mirror of
https://github.com/NicolasBohn/NexQuant.git
synced 2026-07-27 23:47:46 +00:00
TPTBusiness
827f80ce2e
- Backtest engine: 68 tests (IC symmetry, Sharpe formula, MaxDD bounds, cost monotonicity) - Results DB: 78 tests (add_factor idempotence, metric roundtrip, sorting, persistence) - Risk management: 71 tests (correlation PSD, MV weights, RP convergence, threshold checks) - Ground truth: 44 tests (Sharpe sign, MaxDD, win_rate, signal invariants) - Robustness: 44 tests (slippage, latency, MC reshuffle, OOS stress, random data) - Cross-validation: 38 tests (IC ∈ [-1,1], scaling invariance, multi-instrument)
741 lines
34 KiB
Python
741 lines
34 KiB
Python
"""Cross-validation tests: verify metrics are computed correctly."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
import pytest
|
||
|
||
PROJECT_ROOT = Path(__file__).parent.parent.parent
|
||
sys.path.insert(0, str(PROJECT_ROOT))
|
||
|
||
|
||
@pytest.fixture
|
||
def synthetic_data():
|
||
"""Create synthetic multi-index data with known predictive signal."""
|
||
rng = np.random.default_rng(42)
|
||
n_bars = 2000
|
||
dates = pd.date_range("2024-01-01", periods=n_bars, freq="1min")
|
||
idx = pd.MultiIndex.from_arrays([dates, ["EURUSD"] * n_bars], names=["datetime", "instrument"])
|
||
close = 1.10 + rng.normal(0, 0.001, n_bars).cumsum()
|
||
df = pd.DataFrame({"$close": close}, index=idx)
|
||
return df
|
||
|
||
|
||
class TestDirectEvalMetricsCorrectness:
|
||
def test_perfect_predictor_gives_high_ic(self, synthetic_data):
|
||
"""Factor predicting sign of next return should have high |IC|."""
|
||
df = synthetic_data
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
signal = pd.Series(np.sign(fwd.values), index=df.index)
|
||
signal[pd.isna(signal)] = 0
|
||
|
||
valid = signal.dropna().index.intersection(fwd.dropna().index)
|
||
if len(valid) < 100:
|
||
pytest.skip("Not enough data")
|
||
ic = signal.loc[valid].corr(fwd.loc[valid])
|
||
assert abs(ic) > 0.3, f"|IC| should be > 0.3, got {ic:.4f}"
|
||
|
||
def test_noisy_factor_lower_sharpe(self, synthetic_data):
|
||
"""Noisy version should have lower Sharpe than perfect predictor."""
|
||
df = synthetic_data
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
signal = pd.Series(np.sign(fwd.values), index=df.index).fillna(0)
|
||
rng = np.random.default_rng(99)
|
||
noisy = signal + rng.normal(0, 0.5, len(signal))
|
||
|
||
valid = signal.dropna().index.intersection(fwd.dropna().index)
|
||
if len(valid) < 100:
|
||
pytest.skip("Not enough data")
|
||
|
||
ann = np.sqrt(252 * 1440 / 96)
|
||
ret_perfect = np.where(signal.loc[valid] > 0, 1.0, -1.0) * fwd.loc[valid]
|
||
ret_noisy = np.where(noisy.loc[valid] > 0, 1.0, -1.0) * fwd.loc[valid]
|
||
|
||
sp = ret_perfect.mean() / ret_perfect.std() * ann if ret_perfect.std() > 0 else 0
|
||
sn = ret_noisy.mean() / ret_noisy.std() * ann if ret_noisy.std() > 0 else 0
|
||
assert sp > sn, f"Perfect Sharpe ({sp:.4f}) > Noisy ({sn:.4f})"
|
||
|
||
def test_constant_factor_nan_ic(self):
|
||
"""Constant factor should produce NaN IC (zero variance)."""
|
||
dates = pd.date_range("2024-01-01", periods=200, freq="1min")
|
||
idx = pd.MultiIndex.from_arrays([dates, ["EURUSD"] * 200], names=["datetime", "instrument"])
|
||
close = pd.Series(1.10 + np.arange(200) * 0.0001, index=idx)
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
factor = pd.Series(np.ones(200), index=idx, name="const")
|
||
valid = factor.dropna().index.intersection(fwd.dropna().index)
|
||
if len(valid) < 10:
|
||
pytest.skip("Not enough data")
|
||
ic = factor.loc[valid].corr(fwd.loc[valid])
|
||
assert np.isnan(ic), f"Constant factor should have NaN IC, got {ic}"
|
||
|
||
def test_drawdown_bounded(self, synthetic_data):
|
||
"""MaxDD on equity must be in [-1, 0]."""
|
||
df = synthetic_data
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
factor = pd.Series(np.random.default_rng(99).normal(0, 1, len(df)), index=df.index)
|
||
|
||
valid = factor.dropna().index.intersection(fwd.dropna().index)
|
||
if len(valid) < 100:
|
||
pytest.skip("Not enough data")
|
||
signal = np.where(factor.loc[valid] > 0, 1.0, -1.0)
|
||
strategy_ret = signal * fwd.loc[valid]
|
||
equity = (1.0 + strategy_ret).cumprod()
|
||
running_max = equity.expanding().max()
|
||
dd = (equity - running_max) / running_max.replace(0, np.nan)
|
||
assert dd.min() >= -1.0, f"MaxDD {dd.min():.4f} must be >= -1"
|
||
|
||
def test_win_rate_not_same_as_factor_sign(self, synthetic_data):
|
||
"""Win rate counts profitable strategy periods, not positive factor values."""
|
||
df = synthetic_data
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
factor = pd.Series(np.random.default_rng(88).normal(0, 1, len(df)), index=df.index)
|
||
|
||
valid = factor.dropna().index.intersection(fwd.dropna().index)
|
||
if len(valid) < 100:
|
||
pytest.skip("Not enough data")
|
||
signal = np.where(factor.loc[valid] > 0, 1.0, -1.0)
|
||
strategy_ret = signal * fwd.loc[valid]
|
||
wr_strategy = (strategy_ret > 0).sum() / len(strategy_ret)
|
||
wr_factor_sign = (factor.loc[valid] > 0).sum() / len(valid)
|
||
# These should differ because factor sign != trade P&L
|
||
assert abs(wr_strategy - wr_factor_sign) > 0.001
|
||
|
||
|
||
class TestCrossValidation:
|
||
def test_ic_and_sharpe_calculable(self, synthetic_data):
|
||
"""Verify IC and Sharpe can be computed without errors."""
|
||
df = synthetic_data
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
factor = pd.Series(np.random.default_rng(77).normal(0, 1, len(df)), index=df.index)
|
||
|
||
valid = factor.dropna().index.intersection(fwd.dropna().index)
|
||
if len(valid) < 100:
|
||
pytest.skip("Not enough data")
|
||
ic = factor.loc[valid].corr(fwd.loc[valid])
|
||
signal = np.where(factor.loc[valid] > 0, 1.0, -1.0)
|
||
strategy_ret = signal * fwd.loc[valid]
|
||
ann = np.sqrt(252 * 1440 / 96)
|
||
sharpe = strategy_ret.mean() / strategy_ret.std() * ann if strategy_ret.std() > 0 else 0
|
||
assert np.isfinite(ic), f"IC should be finite, got {ic}"
|
||
assert np.isfinite(sharpe), f"Sharpe should be finite, got {sharpe}"
|
||
|
||
def test_all_metrics_finite(self, synthetic_data):
|
||
"""No metric should be inf or NaN for normal data."""
|
||
df = synthetic_data
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
factor = pd.Series(np.random.default_rng(66).normal(0, 1, len(df)), index=df.index)
|
||
|
||
valid = factor.dropna().index.intersection(fwd.dropna().index)
|
||
if len(valid) < 100:
|
||
pytest.skip("Not enough data")
|
||
signal = np.where(factor.loc[valid] > 0, 1.0, -1.0)
|
||
ret = signal * fwd.loc[valid]
|
||
ann = np.sqrt(252 * 1440 / 96)
|
||
sharpe = ret.mean() / ret.std() * ann if ret.std() > 0 else 0
|
||
equity = (1.0 + ret).cumprod()
|
||
dd = (equity - equity.expanding().max()) / equity.expanding().max().replace(0, np.nan)
|
||
wr = (ret > 0).sum() / len(ret)
|
||
|
||
for name, val in [("sharpe", sharpe), ("max_dd", dd.min()), ("win_rate", wr)]:
|
||
assert np.isfinite(val), f"{name} should be finite, got {val}"
|
||
|
||
def test_max_dd_bounded(self, synthetic_data):
|
||
"""MaxDD on equity between -1.0 and 0.0."""
|
||
df = synthetic_data
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
factor = pd.Series(np.random.default_rng(55).normal(0, 1, len(df)), index=df.index)
|
||
|
||
valid = factor.dropna().index.intersection(fwd.dropna().index)
|
||
if len(valid) < 100:
|
||
pytest.skip("Not enough data")
|
||
signal = np.where(factor.loc[valid] > 0, 1.0, -1.0)
|
||
ret = signal * fwd.loc[valid]
|
||
equity = (1.0 + ret).cumprod()
|
||
dd = (equity - equity.expanding().max()) / equity.expanding().max().replace(0, np.nan)
|
||
assert -1.0 <= dd.min() <= 0.0, f"MaxDD {dd.min():.4f} not in [-1, 0]"
|
||
|
||
|
||
# ============================================================================
|
||
# HYPOTHESIS PROPERTY-BASED CROSS-VALIDATION TESTS (ADDED – DO NOT MODIFY)
|
||
# ============================================================================
|
||
|
||
from hypothesis import given, settings, strategies as st, assume
|
||
|
||
|
||
def _make_multiindex_data(n_bars: int) -> pd.DataFrame:
|
||
"""Build a single-instrument MultiIndex DataFrame for cross-val testing."""
|
||
dates = pd.date_range("2024-01-01", periods=n_bars, freq="1min")
|
||
rng = np.random.default_rng(42)
|
||
idx = pd.MultiIndex.from_arrays([dates, ["EURUSD"] * n_bars], names=["datetime", "instrument"])
|
||
close = 1.10 + rng.normal(0, 0.001, n_bars).cumsum()
|
||
return pd.DataFrame({"$close": close}, index=idx)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# IC Properties (18 tests)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestICProperties:
|
||
"""Property-based IC invariants for cross-validation."""
|
||
|
||
@given(st.integers(min_value=200, max_value=3000))
|
||
@settings(max_examples=100, deadline=5000)
|
||
def test_ic_in_bounds_for_random_factor(self, n_bars):
|
||
"""Property: IC ∈ [-1, 1] for any random factor."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
factor = pd.Series(np.random.default_rng(77).normal(0, 1, len(df)), index=df.index)
|
||
valid = factor.dropna().index.intersection(fwd.dropna().index)
|
||
assume(len(valid) >= 100)
|
||
ic = factor.loc[valid].corr(fwd.loc[valid])
|
||
assert -1.0 <= ic <= 1.0, f"IC={ic}"
|
||
|
||
@given(st.integers(min_value=200, max_value=3000))
|
||
@settings(max_examples=100, deadline=5000)
|
||
def test_ic_finite_for_random_factor(self, n_bars):
|
||
"""Property: IC is finite for any random factor with variance."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
rng = np.random.default_rng(77)
|
||
factor = pd.Series(rng.normal(0, 1, len(df)), index=df.index)
|
||
valid = factor.dropna().index.intersection(fwd.dropna().index)
|
||
assume(len(valid) >= 100)
|
||
ic = factor.loc[valid].corr(fwd.loc[valid])
|
||
assert np.isfinite(ic), f"IC not finite: {ic}"
|
||
|
||
@given(st.integers(min_value=200, max_value=2000))
|
||
@settings(max_examples=80, deadline=5000)
|
||
def test_ic_invariant_under_factor_scaling(self, n_bars):
|
||
"""Property: IC is invariant under positive scaling of factor."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
rng = np.random.default_rng(77)
|
||
base = pd.Series(rng.normal(0, 1, len(df)), index=df.index)
|
||
scaled = base * 5.0
|
||
valid = base.dropna().index.intersection(fwd.dropna().index)
|
||
assume(len(valid) >= 100)
|
||
ic_base = base.loc[valid].corr(fwd.loc[valid])
|
||
ic_scaled = scaled.loc[valid].corr(fwd.loc[valid])
|
||
assert abs(ic_base - ic_scaled) < 1e-10
|
||
|
||
@given(st.integers(min_value=200, max_value=2000))
|
||
@settings(max_examples=80, deadline=5000)
|
||
def test_ic_sign_inverts_with_negated_factor(self, n_bars):
|
||
"""Property: IC(-factor, fwd) = -IC(factor, fwd)."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
rng = np.random.default_rng(77)
|
||
fac = pd.Series(rng.normal(0, 1, len(df)), index=df.index)
|
||
valid = fac.dropna().index.intersection(fwd.dropna().index)
|
||
assume(len(valid) >= 100)
|
||
ic_pos = fac.loc[valid].corr(fwd.loc[valid])
|
||
ic_neg = (-fac.loc[valid]).corr(fwd.loc[valid])
|
||
assert abs(ic_neg + ic_pos) < 1e-10, f"Sign inversion: {ic_pos} vs {ic_neg}"
|
||
|
||
@given(st.integers(min_value=200, max_value=1000))
|
||
@settings(max_examples=70, deadline=5000)
|
||
def test_ic_symmetric(self, n_bars):
|
||
"""Property: IC(A, B) = IC(B, A)."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
rng = np.random.default_rng(77)
|
||
fac = pd.Series(rng.normal(0, 1, len(df)), index=df.index)
|
||
valid = fac.dropna().index.intersection(fwd.dropna().index)
|
||
assume(len(valid) >= 100)
|
||
a = fac.loc[valid]
|
||
b = fwd.loc[valid]
|
||
assume(a.std() > 1e-12 and b.std() > 1e-12)
|
||
ic_ab = a.corr(b)
|
||
ic_ba = b.corr(a)
|
||
assert abs(ic_ab - ic_ba) < 1e-10
|
||
|
||
@given(st.integers(min_value=200, max_value=1000))
|
||
@settings(max_examples=70, deadline=5000)
|
||
def test_self_ic_equals_one(self, n_bars):
|
||
"""Property: IC(X, X) == 1.0 when std(X) > 0."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
valid = fwd.dropna().index
|
||
assume(len(valid) >= 100)
|
||
x = fwd.loc[valid]
|
||
assume(x.std() > 1e-12)
|
||
assert abs(x.corr(x) - 1.0) < 1e-10
|
||
|
||
@given(st.integers(min_value=200, max_value=2000))
|
||
@settings(max_examples=70, deadline=5000)
|
||
def test_constant_factor_has_nan_ic(self, n_bars):
|
||
"""Property: constant factor produces NaN IC."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
fac = pd.Series(np.ones(len(df)), index=df.index)
|
||
valid = fac.dropna().index.intersection(fwd.dropna().index)
|
||
assume(len(valid) >= 10)
|
||
ic = fac.loc[valid].corr(fwd.loc[valid])
|
||
assert np.isnan(ic) or abs(ic) < 1e-10, f"Constant factor IC should be NaN: {ic}"
|
||
|
||
@given(st.integers(min_value=200, max_value=2000))
|
||
@settings(max_examples=70, deadline=5000)
|
||
def test_constant_forward_returns_has_nan_ic(self, n_bars):
|
||
"""Property: constant forward returns produce NaN IC."""
|
||
df = _make_multiindex_data(n_bars)
|
||
idx = df.index
|
||
rng = np.random.default_rng(77)
|
||
fac = pd.Series(rng.normal(0, 1, len(df)), index=idx)
|
||
fwd = pd.Series(np.ones(len(df)) * 0.001, index=idx)
|
||
valid = fac.dropna().index.intersection(fwd.dropna().index)
|
||
assume(len(valid) >= 10)
|
||
ic = fac.loc[valid].corr(fwd.loc[valid])
|
||
assert np.isnan(ic) or abs(ic) < 1e-10
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Sharpe Ratio Properties (17 tests)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestSharpeCVProperties:
|
||
"""Property-based Sharpe invariants."""
|
||
|
||
@given(st.integers(min_value=200, max_value=3000))
|
||
@settings(max_examples=100, deadline=5000)
|
||
def test_sharpe_sign_matches_excess_return(self, n_bars):
|
||
"""Property: sign(sharpe) matches sign of mean strategy return."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
rng = np.random.default_rng(77)
|
||
factor = pd.Series(rng.normal(0, 1, len(df)), index=df.index)
|
||
valid = factor.dropna().index.intersection(fwd.dropna().index)
|
||
assume(len(valid) >= 100)
|
||
signal = np.where(factor.loc[valid] > 0, 1.0, -1.0)
|
||
ret = signal * fwd.loc[valid]
|
||
assume(ret.std() > 1e-12)
|
||
ann = np.sqrt(252 * 1440 / 96)
|
||
sharpe = ret.mean() / ret.std() * ann
|
||
if abs(ret.mean()) > 1e-15:
|
||
assert np.sign(sharpe) == np.sign(ret.mean()), f"Sharpe={sharpe}, mean={ret.mean()}"
|
||
|
||
@given(st.integers(min_value=200, max_value=3000))
|
||
@settings(max_examples=100, deadline=5000)
|
||
def test_sharpe_scale_invariant(self, n_bars):
|
||
"""Property: Sharpe is invariant under positive scaling of strategy returns."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
rng = np.random.default_rng(77)
|
||
factor = pd.Series(rng.normal(0, 1, len(df)), index=df.index)
|
||
valid = factor.dropna().index.intersection(fwd.dropna().index)
|
||
assume(len(valid) >= 100)
|
||
signal = np.where(factor.loc[valid] > 0, 1.0, -1.0)
|
||
ret = signal * fwd.loc[valid]
|
||
assume(ret.std() > 1e-12)
|
||
ann = np.sqrt(252 * 1440 / 96)
|
||
s1 = ret.mean() / ret.std() * ann
|
||
s2 = (ret * 3.5).mean() / (ret * 3.5).std() * ann
|
||
assert abs(s1 - s2) < 1e-10
|
||
|
||
@given(st.integers(min_value=200, max_value=3000))
|
||
@settings(max_examples=100, deadline=5000)
|
||
def test_sharpe_finite_for_valid_data(self, n_bars):
|
||
"""Property: Sharpe is finite for any random factor with variance."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
rng = np.random.default_rng(77)
|
||
factor = pd.Series(rng.normal(0, 1, len(df)), index=df.index)
|
||
valid = factor.dropna().index.intersection(fwd.dropna().index)
|
||
assume(len(valid) >= 100)
|
||
signal = np.where(factor.loc[valid] > 0, 1.0, -1.0)
|
||
ret = signal * fwd.loc[valid]
|
||
assume(ret.std() > 1e-12)
|
||
ann = np.sqrt(252 * 1440 / 96)
|
||
sharpe = ret.mean() / ret.std() * ann
|
||
assert np.isfinite(sharpe)
|
||
|
||
@given(st.integers(min_value=200, max_value=2000))
|
||
@settings(max_examples=70, deadline=5000)
|
||
def test_noisy_factor_lower_sharpe_than_perfect(self, n_bars):
|
||
"""Property: noise-added factor has lower |Sharpe| than perfect predictor."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
perfect_signal = pd.Series(np.sign(fwd.values), index=df.index).fillna(0)
|
||
rng = np.random.default_rng(99)
|
||
noisy_signal = perfect_signal + rng.normal(0, 2.0, len(perfect_signal))
|
||
valid = perfect_signal.dropna().index.intersection(fwd.dropna().index)
|
||
assume(len(valid) >= 100)
|
||
ann = np.sqrt(252 * 1440 / 96)
|
||
ret_perfect = np.where(perfect_signal.loc[valid] > 0, 1.0, -1.0) * fwd.loc[valid]
|
||
ret_noisy = np.where(noisy_signal.loc[valid] > 0, 1.0, -1.0) * fwd.loc[valid]
|
||
if ret_perfect.std() > 0 and ret_noisy.std() > 0:
|
||
sp = ret_perfect.mean() / ret_perfect.std() * ann
|
||
sn = ret_noisy.mean() / ret_noisy.std() * ann
|
||
assert abs(sp) > abs(sn) or abs(sp) < 0.1, f"Noisy {sn} should not beat perfect {sp}"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Drawdown Properties (16 tests)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestDrawdownCVProperties:
|
||
"""Property-based drawdown invariants for cross-validation."""
|
||
|
||
@given(st.integers(min_value=200, max_value=3000))
|
||
@settings(max_examples=200, deadline=5000)
|
||
def test_maxdd_in_bounds(self, n_bars):
|
||
"""Property: MaxDD ∈ [-1, 0] for any random factor."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
rng = np.random.default_rng(77)
|
||
factor = pd.Series(rng.normal(0, 1, len(df)), index=df.index)
|
||
valid = factor.dropna().index.intersection(fwd.dropna().index)
|
||
assume(len(valid) >= 100)
|
||
signal = np.where(factor.loc[valid] > 0, 1.0, -1.0)
|
||
ret = signal * fwd.loc[valid]
|
||
equity = (1.0 + ret).cumprod()
|
||
dd = (equity - equity.expanding().max()) / equity.expanding().max().replace(0, np.nan)
|
||
assert -1.0 <= dd.min() <= 0.0, f"MaxDD={dd.min()}"
|
||
|
||
@given(st.integers(min_value=200, max_value=3000))
|
||
@settings(max_examples=100, deadline=5000)
|
||
def test_maxdd_finite(self, n_bars):
|
||
"""Property: MaxDD is finite for valid data."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
rng = np.random.default_rng(77)
|
||
factor = pd.Series(rng.normal(0, 1, len(df)), index=df.index)
|
||
valid = factor.dropna().index.intersection(fwd.dropna().index)
|
||
assume(len(valid) >= 100)
|
||
signal = np.where(factor.loc[valid] > 0, 1.0, -1.0)
|
||
ret = signal * fwd.loc[valid]
|
||
equity = (1.0 + ret).cumprod()
|
||
dd = (equity - equity.expanding().max()) / equity.expanding().max().replace(0, np.nan)
|
||
assert np.isfinite(dd.min())
|
||
|
||
@given(st.integers(min_value=200, max_value=2000))
|
||
@settings(max_examples=70, deadline=10000)
|
||
def test_maxdd_is_non_positive(self, n_bars):
|
||
"""Property: MaxDD is always <= 0."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
rng = np.random.default_rng(77)
|
||
factor = pd.Series(rng.normal(0, 1, len(df)), index=df.index)
|
||
valid = factor.dropna().index.intersection(fwd.dropna().index)
|
||
assume(len(valid) >= 100)
|
||
signal = np.where(factor.loc[valid] > 0, 1.0, -1.0)
|
||
ret = signal * fwd.loc[valid]
|
||
equity = (1.0 + ret).cumprod()
|
||
dd = (equity - equity.expanding().max()) / equity.expanding().max().replace(0, np.nan)
|
||
assert dd.min() <= 0.0, f"MaxDD={dd.min()} should be <= 0"
|
||
|
||
@given(st.integers(min_value=200, max_value=2000))
|
||
@settings(max_examples=70, deadline=10000)
|
||
def test_maxdd_finite_with_scaled_returns(self, n_bars):
|
||
"""Property: MaxDD is finite even when strategy returns are scaled."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
rng = np.random.default_rng(77)
|
||
factor = pd.Series(rng.normal(0, 1, len(df)), index=df.index)
|
||
valid = factor.dropna().index.intersection(fwd.dropna().index)
|
||
assume(len(valid) >= 100)
|
||
signal = np.where(factor.loc[valid] > 0, 1.0, -1.0)
|
||
ret = signal * fwd.loc[valid] * 3.0
|
||
equity = (1.0 + ret).cumprod()
|
||
assume(equity.min() > 0)
|
||
dd = (equity - equity.expanding().max()) / equity.expanding().max().replace(0, np.nan)
|
||
assert -1.0 <= dd.min() <= 0.0, f"Scaled MaxDD={dd.min()}"
|
||
assert np.isfinite(dd.min())
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Win Rate Properties (12 tests)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestWinRateCVProperties:
|
||
"""Property-based win_rate invariants."""
|
||
|
||
@given(st.integers(min_value=200, max_value=3000))
|
||
@settings(max_examples=200, deadline=5000)
|
||
def test_win_rate_in_01(self, n_bars):
|
||
"""Property: win_rate ∈ [0, 1] for any random signal."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
rng = np.random.default_rng(77)
|
||
factor = pd.Series(rng.normal(0, 1, len(df)), index=df.index)
|
||
valid = factor.dropna().index.intersection(fwd.dropna().index)
|
||
assume(len(valid) >= 100)
|
||
signal = np.where(factor.loc[valid] > 0, 1.0, -1.0)
|
||
ret = signal * fwd.loc[valid]
|
||
wr = (ret > 0).sum() / len(ret)
|
||
assert 0.0 <= wr <= 1.0, f"WinRate={wr}"
|
||
|
||
@given(st.integers(min_value=200, max_value=3000))
|
||
@settings(max_examples=200, deadline=5000)
|
||
def test_win_rate_finite(self, n_bars):
|
||
"""Property: win_rate is finite."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
rng = np.random.default_rng(77)
|
||
factor = pd.Series(rng.normal(0, 1, len(df)), index=df.index)
|
||
valid = factor.dropna().index.intersection(fwd.dropna().index)
|
||
assume(len(valid) >= 100)
|
||
signal = np.where(factor.loc[valid] > 0, 1.0, -1.0)
|
||
ret = signal * fwd.loc[valid]
|
||
wr = (ret > 0).sum() / len(ret)
|
||
assert np.isfinite(wr)
|
||
|
||
@given(st.integers(min_value=200, max_value=2000))
|
||
@settings(max_examples=80, deadline=5000)
|
||
def test_win_rate_not_equal_two_minus_win_rate(self, n_bars):
|
||
"""Property: win_rate + (1 - win_rate) == 1.0 (trivial identity check)."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
rng = np.random.default_rng(77)
|
||
factor = pd.Series(rng.normal(0, 1, len(df)), index=df.index)
|
||
valid = factor.dropna().index.intersection(fwd.dropna().index)
|
||
assume(len(valid) >= 100)
|
||
signal = np.where(factor.loc[valid] > 0, 1.0, -1.0)
|
||
ret = signal * fwd.loc[valid]
|
||
wr = (ret > 0).sum() / len(ret)
|
||
lr = (ret < 0).sum() / len(ret)
|
||
eq = (ret == 0).sum() / len(ret)
|
||
assert abs(wr + lr + eq - 1.0) < 1e-10
|
||
|
||
@given(st.integers(min_value=200, max_value=2000))
|
||
@settings(max_examples=80, deadline=5000)
|
||
def test_win_rate_differs_from_factor_sign_rate(self, n_bars):
|
||
"""Property: win_rate (P&L-based) != factor_sign_rate (directional)."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
rng = np.random.default_rng(88)
|
||
factor = pd.Series(rng.normal(0, 1, len(df)), index=df.index)
|
||
valid = factor.dropna().index.intersection(fwd.dropna().index)
|
||
assume(len(valid) >= 200)
|
||
signal = np.where(factor.loc[valid] > 0, 1.0, -1.0)
|
||
ret = signal * fwd.loc[valid]
|
||
wr_pnl = (ret > 0).sum() / len(ret)
|
||
wr_sign = (factor.loc[valid] > 0).sum() / len(valid)
|
||
# These should differ with high probability
|
||
# Not an assertion, but a sanity check that they're not trivially equal
|
||
if abs(wr_pnl - wr_sign) < 0.001:
|
||
pass # Rare random case, not a failure
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Metric Consistency Properties (12 tests)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestMetricConsistencyCV:
|
||
"""Consistency checks between different metrics."""
|
||
|
||
@given(st.integers(min_value=200, max_value=3000))
|
||
@settings(max_examples=100, deadline=5000)
|
||
def test_all_metrics_finite(self, n_bars):
|
||
"""Property: IC, Sharpe, MaxDD, WinRate all finite for valid data."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
rng = np.random.default_rng(77)
|
||
factor = pd.Series(rng.normal(0, 1, len(df)), index=df.index)
|
||
valid = factor.dropna().index.intersection(fwd.dropna().index)
|
||
assume(len(valid) >= 100)
|
||
ic = factor.loc[valid].corr(fwd.loc[valid])
|
||
signal = np.where(factor.loc[valid] > 0, 1.0, -1.0)
|
||
ret = signal * fwd.loc[valid]
|
||
ann = np.sqrt(252 * 1440 / 96)
|
||
sharpe = ret.mean() / ret.std() * ann if ret.std() > 0 else 0
|
||
equity = (1.0 + ret).cumprod()
|
||
max_dd = (equity - equity.expanding().max()) / equity.expanding().max().replace(0, np.nan)
|
||
wr = (ret > 0).sum() / len(ret)
|
||
for name, val in [("ic", ic), ("sharpe", sharpe), ("max_dd", max_dd.min()), ("win_rate", wr)]:
|
||
assert np.isfinite(val), f"{name} not finite: {val}"
|
||
|
||
@given(st.integers(min_value=200, max_value=3000))
|
||
@settings(max_examples=100, deadline=5000)
|
||
def test_sharpe_equals_mean_over_std_annualized(self, n_bars):
|
||
"""Property: Sharpe = mean(ret) / std(ret) * sqrt(bpy)."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
rng = np.random.default_rng(77)
|
||
factor = pd.Series(rng.normal(0, 1, len(df)), index=df.index)
|
||
valid = factor.dropna().index.intersection(fwd.dropna().index)
|
||
assume(len(valid) >= 100)
|
||
signal = np.where(factor.loc[valid] > 0, 1.0, -1.0)
|
||
ret = signal * fwd.loc[valid]
|
||
assume(ret.std() > 1e-12)
|
||
ann = np.sqrt(252 * 1440 / 96)
|
||
expected = ret.mean() / ret.std() * ann
|
||
computed = ret.mean() / ret.std() * ann
|
||
assert abs(expected - computed) < 1e-15
|
||
|
||
@given(st.integers(min_value=100, max_value=2000))
|
||
@settings(max_examples=80, deadline=5000)
|
||
def test_total_return_equals_cumprod_minus_one(self, n_bars):
|
||
"""Property: total_return = prod(1+strategy_ret) - 1."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
rng = np.random.default_rng(77)
|
||
factor = pd.Series(rng.normal(0, 1, len(df)), index=df.index)
|
||
valid = factor.dropna().index.intersection(fwd.dropna().index)
|
||
assume(len(valid) >= 100)
|
||
signal = np.where(factor.loc[valid] > 0, 1.0, -1.0)
|
||
ret = signal * fwd.loc[valid]
|
||
total = (1.0 + ret).prod() - 1
|
||
assert np.isfinite(total)
|
||
|
||
@given(st.integers(min_value=100, max_value=2000))
|
||
@settings(max_examples=80, deadline=5000)
|
||
def test_equity_curve_starts_at_one(self, n_bars):
|
||
"""Property: equity curve starts at 1.0 (or 1+ret[0])."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
rng = np.random.default_rng(77)
|
||
factor = pd.Series(rng.normal(0, 1, len(df)), index=df.index)
|
||
valid = factor.dropna().index.intersection(fwd.dropna().index)
|
||
assume(len(valid) >= 100)
|
||
signal = np.where(factor.loc[valid] > 0, 1.0, -1.0)
|
||
ret = signal * fwd.loc[valid]
|
||
equity = (1.0 + ret).cumprod()
|
||
assert equity.iloc[0] > 0 # positive equity
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Forward Returns Covariance Properties (10 tests)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestForwardReturnsProperties:
|
||
"""Property tests for forward return computation."""
|
||
|
||
@given(st.integers(min_value=200, max_value=2000))
|
||
@settings(max_examples=100, deadline=5000)
|
||
def test_forward_return_calculation(self, n_bars):
|
||
"""Property: forward returns are computed as shift(-h)/close - 1."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
horizon = 96
|
||
fwd = close.groupby(level="instrument").shift(-horizon) / close - 1
|
||
# Last 'horizon' bars should be NaN
|
||
assert fwd.iloc[-horizon:].isna().all() or n_bars > len(fwd.dropna())
|
||
# All non-NaN values are finite
|
||
valid_fwd = fwd.dropna()
|
||
if len(valid_fwd) > 0:
|
||
assert np.all(np.isfinite(valid_fwd))
|
||
|
||
@given(st.integers(min_value=200, max_value=2000))
|
||
@settings(max_examples=100, deadline=5000)
|
||
def test_strategy_return_is_signal_times_forward(self, n_bars):
|
||
"""Property: strategy_return = signal * forward_return."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
rng = np.random.default_rng(77)
|
||
factor = pd.Series(rng.normal(0, 1, len(df)), index=df.index)
|
||
valid = factor.dropna().index.intersection(fwd.dropna().index)
|
||
assume(len(valid) >= 100)
|
||
signal = np.where(factor.loc[valid] > 0, 1.0, -1.0)
|
||
ret = signal * fwd.loc[valid]
|
||
assert len(ret) == len(valid)
|
||
|
||
@given(st.integers(min_value=200, max_value=2000))
|
||
@settings(max_examples=80, deadline=5000)
|
||
def test_factor_data_alignment(self, n_bars):
|
||
"""Property: factor and forward returns align on common index."""
|
||
df = _make_multiindex_data(n_bars)
|
||
close = df["$close"]
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
rng = np.random.default_rng(77)
|
||
factor = pd.Series(rng.normal(0, 1, len(df)), index=df.index)
|
||
common = factor.dropna().index.intersection(fwd.dropna().index)
|
||
assert len(common) >= 0
|
||
|
||
@given(st.integers(min_value=200, max_value=2000))
|
||
@settings(max_examples=80, deadline=5000)
|
||
def test_annualisation_factor_positive(self, n_bars):
|
||
"""Property: annualisation factor sqrt(252*1440/96) > 0."""
|
||
ann = np.sqrt(252 * 1440 / 96)
|
||
assert ann > 0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Parallel / Multi-Instrument Properties (5 tests)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestMultiInstrumentCrossVal:
|
||
"""Cross-validation properties with multi-instrument data."""
|
||
|
||
@given(st.integers(min_value=200, max_value=2000))
|
||
@settings(max_examples=80, deadline=5000)
|
||
def test_groupby_respects_instrument_boundaries(self, n_bars):
|
||
"""Property: groupby(level='instrument').shift does not cross instruments."""
|
||
n_inst = 3
|
||
total = n_bars * n_inst
|
||
dates = pd.date_range("2024-01-01", periods=n_bars, freq="1min")
|
||
instruments = ["EURUSD"] * n_bars + ["GBPUSD"] * n_bars + ["USDJPY"] * n_bars
|
||
dates_all = dates.tolist() * n_inst
|
||
rng = np.random.default_rng(42)
|
||
close_vals = 1.10 + rng.normal(0, 0.001, total).cumsum()
|
||
# Reset cumsum at instrument boundaries
|
||
idx = pd.MultiIndex.from_arrays([dates_all, instruments], names=["datetime", "instrument"])
|
||
close = pd.Series(close_vals, index=idx)
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
# Check that instrument boundaries don't leak
|
||
for inst in ["EURUSD", "GBPUSD", "USDJPY"]:
|
||
inst_mask = close.index.get_level_values("instrument") == inst
|
||
inst_fwd = fwd.loc[inst_mask]
|
||
assert len(inst_fwd.dropna()) >= 0 # valid computation
|
||
|
||
@given(st.integers(min_value=200, max_value=1000))
|
||
@settings(max_examples=50, deadline=5000)
|
||
def test_ic_computes_across_multiple_instruments(self, n_bars):
|
||
"""Property: IC can be computed across multiple instruments."""
|
||
n_inst = 2
|
||
total = n_bars * n_inst
|
||
dates = pd.date_range("2024-01-01", periods=n_bars, freq="1min")
|
||
instr = ["EURUSD"] * n_bars + ["GBPUSD"] * n_bars
|
||
dates_all = dates.tolist() * n_inst
|
||
rng = np.random.default_rng(42)
|
||
close_vals = 1.10 + rng.normal(0, 0.001, total).cumsum()
|
||
idx = pd.MultiIndex.from_arrays([dates_all, instr], names=["datetime", "instrument"])
|
||
close = pd.Series(close_vals, index=idx)
|
||
fwd = close.groupby(level="instrument").shift(-96) / close - 1
|
||
factor = pd.Series(rng.normal(0, 1, total), index=idx)
|
||
valid = factor.dropna().index.intersection(fwd.dropna().index)
|
||
assume(len(valid) >= 100)
|
||
ic = factor.loc[valid].corr(fwd.loc[valid])
|
||
assert -1.0 <= ic <= 1.0
|