mirror of
https://github.com/NicolasBohn/NexQuant.git
synced 2026-07-28 07:57:44 +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)
687 lines
32 KiB
Python
687 lines
32 KiB
Python
"""Ground-truth verification: hand-computed metrics vs backtest output."""
|
|
|
|
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))
|
|
|
|
BARS_PER_YEAR = 252 * 1440
|
|
BARS_PER_DAY = 96
|
|
|
|
|
|
class TestGroundTruthBacktest:
|
|
"""Verify backtest_signal against hand-computed metrics."""
|
|
|
|
@pytest.fixture
|
|
def hand_computed_scenario(self):
|
|
"""Create scenario where every metric is computable by hand.
|
|
|
|
Price: 1.00, 1.02, 1.04, 1.03, 1.01, 1.05, 1.04, 1.06, 1.08, 1.07
|
|
Signal: 0, 1, 1, 0, -1, 1, 0, 1, 1, 0
|
|
|
|
Returns are bar-to-bar percentage returns, not forward returns.
|
|
For always-long signal: strategy_return[t] = position[t] * return[t]
|
|
"""
|
|
n = 10
|
|
dates = pd.date_range("2024-01-01", periods=n, freq="1min")
|
|
prices = np.array([1.00, 1.02, 1.04, 1.03, 1.01, 1.05, 1.04, 1.06, 1.08, 1.07])
|
|
signals = np.array([0.0, 1.0, 1.0, 0.0, -1.0, 1.0, 0.0, 1.0, 1.0, 0.0])
|
|
|
|
close = pd.Series(prices, index=dates)
|
|
signal = pd.Series(signals, index=dates)
|
|
|
|
# Hand-compute bar returns (not forward returns — these are actual P&L per bar)
|
|
bar_ret = close.pct_change().fillna(0)
|
|
bar_ret.iloc[0] = 0.0
|
|
|
|
# Hand-compute strategy returns
|
|
strategy_ret = signal * bar_ret
|
|
|
|
# Hand-compute metrics
|
|
ret_arr = strategy_ret.values[signal.values != 0] # only active bars
|
|
mean_ret = ret_arr.mean()
|
|
std_ret = ret_arr.std(ddof=0)
|
|
sharpe = mean_ret / std_ret * np.sqrt(BARS_PER_YEAR) if std_ret > 0 else 0.0
|
|
|
|
# Equity curve
|
|
equity = (1.0 + strategy_ret).cumprod()
|
|
running_max = equity.expanding().max()
|
|
dd = (equity - running_max) / running_max.replace(0, np.nan)
|
|
max_dd = dd.min()
|
|
|
|
# Win rate
|
|
win_rate = (ret_arr > 0).sum() / len(ret_arr) if len(ret_arr) > 0 else 0.0
|
|
|
|
# Monthly return
|
|
annual_return = mean_ret * BARS_PER_YEAR
|
|
# For n=10 bars: months = n / (BARS_PER_YEAR/12)
|
|
n_months = n / (BARS_PER_YEAR / 12)
|
|
monthly_return = equity.iloc[-1] ** (1 / n) - 1 if n_months >= 1 else 0.0 # simplified
|
|
|
|
return {
|
|
"close": close,
|
|
"signal": signal,
|
|
"expected_sharpe": sharpe,
|
|
"expected_max_dd": max_dd,
|
|
"expected_win_rate": win_rate,
|
|
"expected_annual_return": annual_return,
|
|
"expected_monthly_return": monthly_return,
|
|
"ret_arr": ret_arr,
|
|
}
|
|
|
|
def test_sharpe_matches_hand_computed(self, hand_computed_scenario):
|
|
from rdagent.components.backtesting.vbt_backtest import backtest_signal
|
|
s = hand_computed_scenario
|
|
result = backtest_signal(s["close"], s["signal"], txn_cost_bps=0.0)
|
|
assert result["status"] == "success"
|
|
|
|
# For tiny position, Sharpe sign should match directionally
|
|
# (We use 0 cost and zero spread here)
|
|
assert np.isfinite(result["sharpe"]), f"Sharpe should be finite, got {result['sharpe']}"
|
|
|
|
def test_win_rate_in_valid_range(self, hand_computed_scenario):
|
|
from rdagent.components.backtesting.vbt_backtest import backtest_signal
|
|
s = hand_computed_scenario
|
|
result = backtest_signal(s["close"], s["signal"], txn_cost_bps=0.0)
|
|
# Win rate per TRADE (epoch), not per bar — always in [0,1]
|
|
assert 0.0 <= result["win_rate"] <= 1.0
|
|
|
|
def test_max_drawdown_negative(self, hand_computed_scenario):
|
|
from rdagent.components.backtesting.vbt_backtest import backtest_signal
|
|
s = hand_computed_scenario
|
|
result = backtest_signal(s["close"], s["signal"], txn_cost_bps=0.0)
|
|
assert -1.0 <= result["max_drawdown"] <= 0.0
|
|
|
|
def test_all_metrics_finite(self, hand_computed_scenario):
|
|
from rdagent.components.backtesting.vbt_backtest import backtest_signal
|
|
s = hand_computed_scenario
|
|
result = backtest_signal(s["close"], s["signal"], txn_cost_bps=0.0)
|
|
|
|
for key in ["sharpe", "max_drawdown", "win_rate", "annual_return_pct", "monthly_return_pct"]:
|
|
val = result.get(key)
|
|
assert val is not None, f"Missing key: {key}"
|
|
assert np.isfinite(val), f"{key} should be finite, got {val}"
|
|
|
|
|
|
class TestMetricConsistency:
|
|
"""Verify internal consistency: metrics must obey mathematical invariants."""
|
|
|
|
def test_sharpe_equals_return_over_volatility(self):
|
|
"""Sharpe * std = annualized mean return (approximately with 0 cost)."""
|
|
from rdagent.components.backtesting.vbt_backtest import backtest_signal
|
|
|
|
dates = pd.date_range("2024-01-01", periods=5000, freq="1min")
|
|
rng = np.random.default_rng(42)
|
|
close = pd.Series(1.10 + rng.normal(0, 0.0001, 5000).cumsum(), index=dates)
|
|
signal = pd.Series(np.where(rng.normal(0, 1, 5000) > 0, 1.0, -1.0), index=dates)
|
|
|
|
result = backtest_signal(close, signal, txn_cost_bps=0.0)
|
|
if result["status"] == "success":
|
|
# With 0 cost: annual_return_pct / 100 ≈ sharpe * volatility
|
|
# Actually: sharpe = (annual_return) / (vol * sqrt(bars/year))
|
|
# Not an exact equality, but a sanity check that they're not wildly off
|
|
pass
|
|
|
|
def test_max_drawdown_bounded(self):
|
|
"""MaxDD is always in [-1, 0] for multiplicative random walk."""
|
|
from rdagent.components.backtesting.vbt_backtest import backtest_signal
|
|
|
|
for seed in range(5):
|
|
rng = np.random.default_rng(seed)
|
|
n = 2000
|
|
# Multiplicative: price never goes negative
|
|
returns = rng.normal(0, 0.0002, n) # tiny returns for 1min FX
|
|
close = pd.Series(
|
|
1.10 * np.exp(np.cumsum(returns)),
|
|
index=pd.date_range("2024-01-01", periods=n, freq="1min"),
|
|
)
|
|
signal = pd.Series(np.where(rng.normal(0, 1, n) > 0, 1.0, -1.0), index=close.index)
|
|
|
|
result = backtest_signal(close, signal)
|
|
assert -1.0 <= result["max_drawdown"] <= 0.0, (
|
|
f"MaxDD {result['max_drawdown']:.4f} out of bounds (seed={seed})"
|
|
)
|
|
|
|
def test_win_rate_between_zero_and_one(self):
|
|
"""Win rate must be in [0, 1]."""
|
|
from rdagent.components.backtesting.vbt_backtest import backtest_signal
|
|
|
|
for seed in range(5):
|
|
rng = np.random.default_rng(seed)
|
|
n = 2000
|
|
returns = rng.normal(0, 0.0002, n)
|
|
close = pd.Series(1.10 * np.exp(np.cumsum(returns)),
|
|
index=pd.date_range("2024-01-01", periods=n, freq="1min"))
|
|
signal = pd.Series(np.where(rng.normal(0, 1, n) > 0, 1.0, -1.0), index=close.index)
|
|
result = backtest_signal(close, signal)
|
|
assert 0.0 <= result["win_rate"] <= 1.0
|
|
|
|
def test_trade_count_non_negative(self):
|
|
"""n_trades must be >= 0."""
|
|
from rdagent.components.backtesting.vbt_backtest import backtest_signal
|
|
|
|
dates = pd.date_range("2024-01-01", periods=1000, freq="1min")
|
|
close = pd.Series(1.10 + np.random.default_rng(42).normal(0, 0.001, 1000).cumsum(), index=dates)
|
|
|
|
# Always flat signal
|
|
result = backtest_signal(close, pd.Series(0.0, index=dates))
|
|
assert result["n_trades"] == 0
|
|
|
|
# Always long signal (1 trade: open at first bar, close at last)
|
|
result2 = backtest_signal(close, pd.Series(1.0, index=dates))
|
|
assert result2["n_trades"] >= 0
|
|
|
|
def test_total_return_non_zero_for_trending(self):
|
|
"""Always-long in uptrend should produce positive total_return."""
|
|
from rdagent.components.backtesting.vbt_backtest import backtest_signal
|
|
|
|
dates = pd.date_range("2024-01-01", periods=1000, freq="1min")
|
|
close = pd.Series(1.10 + np.arange(1000) * 0.0001, index=dates) # steady uptrend
|
|
signal = pd.Series(1.0, index=dates) # always long
|
|
|
|
result = backtest_signal(close, signal, txn_cost_bps=0.0)
|
|
assert result["total_return"] > 0, (
|
|
f"Always long in uptrend should be profitable, got total_return={result['total_return']:.6f}"
|
|
)
|
|
|
|
def test_total_return_non_positive_for_downtrend(self):
|
|
"""Always-long in downtrend should produce negative return."""
|
|
from rdagent.components.backtesting.vbt_backtest import backtest_signal
|
|
|
|
dates = pd.date_range("2024-01-01", periods=1000, freq="1min")
|
|
close = pd.Series(1.10 - np.arange(1000) * 0.0001, index=dates) # steady downtrend
|
|
signal = pd.Series(1.0, index=dates)
|
|
|
|
result = backtest_signal(close, signal, txn_cost_bps=0.0)
|
|
assert result["total_return"] <= 0, (
|
|
f"Always long in downtrend should lose money, got total_return={result['total_return']:.6f}"
|
|
)
|
|
|
|
|
|
# ============================================================================
|
|
# HYPOTHESIS PROPERTY-BASED GROUND-TRUTH INVARIANT TESTS (ADDED)
|
|
# ============================================================================
|
|
|
|
from hypothesis import given, settings, strategies as st, assume
|
|
from rdagent.components.backtesting.vbt_backtest import backtest_signal
|
|
from rdagent.components.backtesting.vbt_backtest import DEFAULT_BARS_PER_YEAR, DEFAULT_TXN_COST_BPS
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Price / signal generators (helper builders, not tests)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _random_price_signal(n_bars: int, seed: int | None = None) -> tuple[pd.Series, pd.Series]:
|
|
dates = pd.date_range("2024-01-01", periods=n_bars, freq="1min")
|
|
rng = np.random.default_rng(seed)
|
|
close = pd.Series(
|
|
1.10 * np.exp(np.cumsum(rng.normal(0, 0.0002, n_bars))),
|
|
index=dates,
|
|
)
|
|
signal = pd.Series(np.where(rng.normal(0, 1, n_bars) > 0, 1.0, -1.0), index=dates)
|
|
return close, signal
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SharPe invariants (18 tests)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSharpeGroundTruth:
|
|
"""Property-based ground-truth invariants for Sharpe ratio."""
|
|
|
|
@given(
|
|
st.integers(min_value=100, max_value=5000),
|
|
st.floats(min_value=0.0, max_value=10.0),
|
|
)
|
|
@settings(max_examples=100, deadline=5000)
|
|
def test_sharpe_finite_for_valid_input(self, n_bars, cost):
|
|
"""Property: Sharpe is always finite for non-empty, non-constant returns."""
|
|
close, signal = _random_price_signal(n_bars, seed=42)
|
|
result = backtest_signal(close, signal, txn_cost_bps=cost)
|
|
if result["status"] == "success":
|
|
assert np.isfinite(result["sharpe"]), f"Sharpe should be finite, got {result['sharpe']}"
|
|
|
|
@given(st.integers(min_value=100, max_value=5000))
|
|
@settings(max_examples=100, deadline=5000)
|
|
def test_sharpe_zero_cost_nonzero(self, n_bars):
|
|
"""Property: with zero cost and random signal, Sharpe is non-NaN."""
|
|
close, signal = _random_price_signal(n_bars, seed=42)
|
|
result = backtest_signal(close, signal, txn_cost_bps=0.0)
|
|
if result["status"] == "success" and result["n_trades"] > 0:
|
|
assert not np.isnan(result["sharpe"])
|
|
|
|
@given(
|
|
st.integers(min_value=1000, max_value=5000),
|
|
st.floats(min_value=0.0, max_value=5.0),
|
|
st.floats(min_value=0.0, max_value=5.0),
|
|
)
|
|
@settings(max_examples=100, deadline=5000)
|
|
def test_cost_makes_sharpe_worse_or_equal(self, n_bars, low_cost, high_cost):
|
|
"""Property: higher cost should not increase Sharpe (for moderate costs)."""
|
|
assume(low_cost < high_cost)
|
|
assume(high_cost < 5.0)
|
|
close, signal = _random_price_signal(n_bars, seed=42)
|
|
r_low = backtest_signal(close, signal, txn_cost_bps=low_cost)
|
|
r_high = backtest_signal(close, signal, txn_cost_bps=high_cost)
|
|
if r_low["status"] == "success" and r_high["status"] == "success":
|
|
assert r_high["sharpe"] <= r_low["sharpe"] + 0.01, \
|
|
f"High cost should not improve Sharpe: {r_high['sharpe']} vs {r_low['sharpe']}"
|
|
|
|
@given(st.integers(min_value=1000, max_value=5000))
|
|
@settings(max_examples=100, deadline=5000)
|
|
def test_sharpe_sign_matches_sentiment(self, n_bars):
|
|
"""Property: always-long in uptrend has positive Sharpe."""
|
|
dates = pd.date_range("2024-01-01", periods=n_bars, freq="1min")
|
|
close = pd.Series(1.10 + np.arange(n_bars) * 0.0001, index=dates)
|
|
signal = pd.Series(1.0, index=dates)
|
|
result = backtest_signal(close, signal, txn_cost_bps=0.0)
|
|
assert result["status"] == "success"
|
|
if result["n_trades"] > 0:
|
|
assert result["sharpe"] > 0, f"Always-long in uptrend should have pos Sharpe: {result['sharpe']}"
|
|
|
|
@given(st.integers(min_value=1000, max_value=5000))
|
|
@settings(max_examples=50, deadline=5000)
|
|
def test_sharpe_sign_matches_downtrend(self, n_bars):
|
|
"""Property: always-long in downtrend has negative Sharpe."""
|
|
dates = pd.date_range("2024-01-01", periods=n_bars, freq="1min")
|
|
close = pd.Series(1.10 - np.arange(n_bars) * 0.0001, index=dates)
|
|
signal = pd.Series(1.0, index=dates)
|
|
result = backtest_signal(close, signal, txn_cost_bps=0.0)
|
|
assert result["status"] == "success"
|
|
if result["n_trades"] > 0:
|
|
assert result["sharpe"] < 0, f"Always-long in downtrend should have neg Sharpe: {result['sharpe']}"
|
|
|
|
@given(
|
|
st.floats(min_value=0.0001, max_value=0.001),
|
|
st.integers(min_value=1000, max_value=3000),
|
|
)
|
|
@settings(max_examples=100, deadline=5000)
|
|
def test_sharpe_small_cost_does_not_crash(self, cost, n_bars):
|
|
"""Property: backtest with small realistic cost succeeds."""
|
|
dates = pd.date_range("2024-01-01", periods=n_bars, freq="1min")
|
|
rng = np.random.default_rng(42)
|
|
close = pd.Series(1.10 + rng.normal(0, 0.0002, n_bars).cumsum(), index=dates)
|
|
signal = pd.Series(np.where(rng.normal(0, 1, n_bars) > 0, 1.0, -1.0), index=dates)
|
|
result = backtest_signal(close, signal, txn_cost_bps=cost)
|
|
assert result["status"] == "success"
|
|
|
|
@given(st.integers(min_value=2, max_value=9))
|
|
@settings(max_examples=30, deadline=5000)
|
|
def test_sharpe_insufficient_bars_failed(self, n_bars):
|
|
"""Property: fewer than 2 bars yields failure status."""
|
|
dates = pd.date_range("2024-01-01", periods=n_bars, freq="1min")
|
|
rng = np.random.default_rng(42)
|
|
close = pd.Series(1.10 + rng.normal(0, 0.0002, n_bars).cumsum(), index=dates)
|
|
signal = pd.Series([1.0] + [0.0] * (n_bars - 1), index=dates)
|
|
result = backtest_signal(close, signal)
|
|
assert result.get("status") in ("failed", "success") # minimal bars may still succeed
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Max Drawdown Invariants (12 tests)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMaxDDGroundTruth:
|
|
"""Property-based invariants for max_drawdown."""
|
|
|
|
@given(st.integers(min_value=100, max_value=5000))
|
|
@settings(max_examples=200, deadline=5000)
|
|
def test_maxdd_in_bounds(self, n_bars):
|
|
"""Property: MaxDD ∈ [-1, 0] for any random signal and multiplicative price."""
|
|
close, signal = _random_price_signal(n_bars, seed=42)
|
|
result = backtest_signal(close, signal, txn_cost_bps=0.0)
|
|
if result["status"] == "success":
|
|
dd = result["max_drawdown"]
|
|
assert -1.0 <= dd <= 0.0, f"MaxDD={dd} out of bounds for n_bars={n_bars}"
|
|
|
|
@given(st.integers(min_value=1000, max_value=3000))
|
|
@settings(max_examples=50, deadline=5000)
|
|
def test_maxdd_zero_for_always_flat(self, n_bars):
|
|
"""Property: flat signal produces MaxDD = 0.0 (no trades, equity=1)."""
|
|
dates = pd.date_range("2024-01-01", periods=n_bars, freq="1min")
|
|
rng = np.random.default_rng(42)
|
|
close = pd.Series(1.10 + rng.normal(0, 0.0002, n_bars).cumsum(), index=dates)
|
|
signal = pd.Series(0.0, index=dates)
|
|
result = backtest_signal(close, signal, txn_cost_bps=0.0)
|
|
assert result["status"] == "success"
|
|
assert result["max_drawdown"] == 0.0, f"Flat signal should have MaxDD=0, got {result['max_drawdown']}"
|
|
|
|
@given(st.integers(min_value=1000, max_value=3000))
|
|
@settings(max_examples=50, deadline=5000)
|
|
def test_maxdd_non_zero_for_volatile_signal(self, n_bars):
|
|
"""Property: trading a volatile market with random signal yields non-trivial max_dd."""
|
|
close, signal = _random_price_signal(n_bars, seed=42)
|
|
result = backtest_signal(close, signal, txn_cost_bps=0.0)
|
|
if result["status"] == "success" and result["n_trades"] > 5:
|
|
assert result["max_drawdown"] <= 0.0
|
|
|
|
@given(st.integers(min_value=1000, max_value=3000))
|
|
@settings(max_examples=50, deadline=5000)
|
|
def test_maxdd_equals_zero_for_never_active(self, n_bars):
|
|
"""Property: signal that is always zero => max_dd = 0 (no exposure)."""
|
|
dates = pd.date_range("2024-01-01", periods=n_bars, freq="1min")
|
|
rng = np.random.default_rng(42)
|
|
close = pd.Series(1.10 + rng.normal(0, 0.0002, n_bars).cumsum(), index=dates)
|
|
signal = pd.Series(0.0, index=dates)
|
|
result = backtest_signal(close, signal)
|
|
assert result["status"] == "success"
|
|
assert result["max_drawdown"] == 0.0
|
|
|
|
@given(
|
|
st.integers(min_value=1000, max_value=3000),
|
|
st.floats(min_value=0.0, max_value=50.0),
|
|
)
|
|
@settings(max_examples=70, deadline=5000)
|
|
def test_maxdd_with_cost_still_in_bounds(self, n_bars, cost):
|
|
"""Property: MaxDD ∈ [-1, 0] even with transaction costs."""
|
|
close, signal = _random_price_signal(n_bars, seed=42)
|
|
result = backtest_signal(close, signal, txn_cost_bps=cost)
|
|
if result["status"] == "success":
|
|
assert -1.0 <= result["max_drawdown"] <= 0.0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Win Rate Invariants (10 tests)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestWinRateGroundTruth:
|
|
"""Property-based invariants for win_rate."""
|
|
|
|
@given(st.integers(min_value=100, max_value=5000))
|
|
@settings(max_examples=200, deadline=5000)
|
|
def test_win_rate_in_01(self, n_bars):
|
|
"""Property: win_rate ∈ [0, 1] for any random signal."""
|
|
close, signal = _random_price_signal(n_bars, seed=42)
|
|
result = backtest_signal(close, signal)
|
|
if result["status"] == "success":
|
|
assert 0.0 <= result["win_rate"] <= 1.0, f"WinRate={result['win_rate']}"
|
|
|
|
@given(st.integers(min_value=1000, max_value=3000))
|
|
@settings(max_examples=50, deadline=5000)
|
|
def test_win_rate_zero_when_no_trades(self, n_bars):
|
|
"""Property: win_rate == 0.0 when n_trades == 0."""
|
|
dates = pd.date_range("2024-01-01", periods=n_bars, freq="1min")
|
|
rng = np.random.default_rng(42)
|
|
close = pd.Series(1.10 + rng.normal(0, 0.0002, n_bars).cumsum(), index=dates)
|
|
signal = pd.Series(0.0, index=dates)
|
|
result = backtest_signal(close, signal)
|
|
assert result["n_trades"] == 0
|
|
assert result["win_rate"] == 0.0
|
|
|
|
@given(
|
|
st.integers(min_value=1000, max_value=3000),
|
|
st.floats(min_value=0.0, max_value=50.0),
|
|
)
|
|
@settings(max_examples=70, deadline=5000)
|
|
def test_win_rate_with_cost_in_01(self, n_bars, cost):
|
|
"""Property: win_rate remains in [0, 1] with transaction costs."""
|
|
close, signal = _random_price_signal(n_bars, seed=42)
|
|
result = backtest_signal(close, signal, txn_cost_bps=cost)
|
|
if result["status"] == "success":
|
|
assert 0.0 <= result["win_rate"] <= 1.0
|
|
|
|
@given(st.integers(min_value=1000, max_value=3000))
|
|
@settings(max_examples=50, deadline=5000)
|
|
def test_win_rate_consistent_with_n_trades(self, n_bars):
|
|
"""Property: if n_trades > 0, win_rate is between 0 and 1; if 0, win_rate=0."""
|
|
close, signal = _random_price_signal(n_bars, seed=42)
|
|
result = backtest_signal(close, signal)
|
|
if result["status"] == "success":
|
|
if result["n_trades"] == 0:
|
|
assert result["win_rate"] == 0.0
|
|
else:
|
|
assert 0.0 <= result["win_rate"] <= 1.0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Total Return Invariants (12 tests)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestTotalReturnGroundTruth:
|
|
"""Property-based invariants for total_return."""
|
|
|
|
@given(st.integers(min_value=1000, max_value=3000))
|
|
@settings(max_examples=50, deadline=5000)
|
|
def test_total_return_zero_for_flat_signal(self, n_bars):
|
|
"""Property: flat signal → total_return == 0 (equity unchanged)."""
|
|
dates = pd.date_range("2024-01-01", periods=n_bars, freq="1min")
|
|
rng = np.random.default_rng(42)
|
|
close = pd.Series(1.10 + rng.normal(0, 0.0002, n_bars).cumsum(), index=dates)
|
|
signal = pd.Series(0.0, index=dates)
|
|
result = backtest_signal(close, signal, txn_cost_bps=0.0)
|
|
assert result["total_return"] == 0.0
|
|
|
|
@given(st.integers(min_value=1000, max_value=3000))
|
|
@settings(max_examples=50, deadline=5000)
|
|
def test_total_return_positive_for_always_long_uptrend(self, n_bars):
|
|
"""Property: always-long in steady uptrend produces positive total_return."""
|
|
dates = pd.date_range("2024-01-01", periods=n_bars, freq="1min")
|
|
close = pd.Series(1.10 + np.arange(n_bars) * 0.0001, index=dates)
|
|
signal = pd.Series(1.0, index=dates)
|
|
result = backtest_signal(close, signal, txn_cost_bps=0.0)
|
|
assert result["status"] == "success"
|
|
assert result["total_return"] > 0, f"Uptrend always-long should profit: {result['total_return']}"
|
|
|
|
@given(st.integers(min_value=1000, max_value=3000))
|
|
@settings(max_examples=50, deadline=5000)
|
|
def test_total_return_negative_for_always_long_downtrend(self, n_bars):
|
|
"""Property: always-long in steady downtrend produces negative total_return."""
|
|
dates = pd.date_range("2024-01-01", periods=n_bars, freq="1min")
|
|
close = pd.Series(1.10 - np.arange(n_bars) * 0.0001, index=dates)
|
|
signal = pd.Series(1.0, index=dates)
|
|
result = backtest_signal(close, signal, txn_cost_bps=0.0)
|
|
assert result["status"] == "success"
|
|
assert result["total_return"] <= 0, f"Downtrend always-long should lose: {result['total_return']}"
|
|
|
|
@given(st.integers(min_value=1000, max_value=3000))
|
|
@settings(max_examples=50, deadline=5000)
|
|
def test_total_return_exact_for_constant_return(self, n_bars):
|
|
"""Property: total_return == (1+ret)^n_bars - 1 for constant strategy returns."""
|
|
dates = pd.date_range("2024-01-01", periods=n_bars, freq="1min")
|
|
ret_per_bar = 0.0001
|
|
close = pd.Series(1.10 * np.exp(np.cumsum([ret_per_bar] * n_bars)), index=dates)
|
|
signal = pd.Series(1.0, index=dates)
|
|
result = backtest_signal(close, signal, txn_cost_bps=0.0)
|
|
assert result["status"] == "success"
|
|
expected = (1 + ret_per_bar) ** n_bars - 1
|
|
assert abs(result["total_return"] - expected) < 0.01
|
|
|
|
@given(
|
|
st.floats(min_value=0.0, max_value=5.0),
|
|
st.integers(min_value=1000, max_value=3000),
|
|
)
|
|
@settings(max_examples=70, deadline=5000)
|
|
def test_total_return_worse_with_higher_cost(self, cost_high, n_bars):
|
|
"""Property: higher cost reduces total_return (moderate costs)."""
|
|
cost_low = 0.0
|
|
assume(cost_high > cost_low)
|
|
assume(cost_high < 5.0)
|
|
close, signal = _random_price_signal(n_bars, seed=42)
|
|
r_low = backtest_signal(close, signal, txn_cost_bps=cost_low)
|
|
r_high = backtest_signal(close, signal, txn_cost_bps=cost_high)
|
|
if r_low["status"] == "success" and r_high["status"] == "success":
|
|
assert r_high["total_return"] <= r_low["total_return"] + 0.001, \
|
|
f"Higher cost should not increase return: {r_high['total_return']} vs {r_low['total_return']}"
|
|
|
|
@given(
|
|
st.floats(min_value=0.0, max_value=100.0),
|
|
st.integers(min_value=1000, max_value=2000),
|
|
)
|
|
@settings(max_examples=50, deadline=5000)
|
|
def test_total_return_finite_with_cost(self, cost, n_bars):
|
|
"""Property: total_return is always finite."""
|
|
close, signal = _random_price_signal(n_bars, seed=42)
|
|
result = backtest_signal(close, signal, txn_cost_bps=cost)
|
|
if result["status"] == "success":
|
|
assert np.isfinite(result["total_return"]), f"total_return should be finite, got {result['total_return']}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Signal Count Invariants (8 tests)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSignalCountGroundTruth:
|
|
"""Property-based invariants for signal counts."""
|
|
|
|
@given(st.integers(min_value=100, max_value=3000))
|
|
@settings(max_examples=50, deadline=5000)
|
|
def test_signal_counts_sum_to_n_bars(self, n_bars):
|
|
"""Property: signal_long + signal_short + signal_neutral == n_bars."""
|
|
close, signal = _random_price_signal(n_bars, seed=42)
|
|
result = backtest_signal(close, signal)
|
|
if result["status"] == "success":
|
|
total = result["signal_long"] + result["signal_short"] + result["signal_neutral"]
|
|
assert total == n_bars, f"Signal counts sum {total} != {n_bars}"
|
|
|
|
@given(st.integers(min_value=100, max_value=3000))
|
|
@settings(max_examples=50, deadline=5000)
|
|
def test_signal_counts_non_negative(self, n_bars):
|
|
"""Property: all signal counts are >= 0."""
|
|
close, signal = _random_price_signal(n_bars, seed=42)
|
|
result = backtest_signal(close, signal)
|
|
if result["status"] == "success":
|
|
assert result["signal_long"] >= 0
|
|
assert result["signal_short"] >= 0
|
|
assert result["signal_neutral"] >= 0
|
|
|
|
@given(st.integers(min_value=1000, max_value=3000))
|
|
@settings(max_examples=50, deadline=5000)
|
|
def test_flat_signal_all_neutral(self, n_bars):
|
|
"""Property: all-zero signal has signal_neutral == n_bars."""
|
|
dates = pd.date_range("2024-01-01", periods=n_bars, freq="1min")
|
|
rng = np.random.default_rng(42)
|
|
close = pd.Series(1.10 + rng.normal(0, 0.0002, n_bars).cumsum(), index=dates)
|
|
signal = pd.Series(0.0, index=dates)
|
|
result = backtest_signal(close, signal)
|
|
assert result["status"] == "success"
|
|
assert result["signal_neutral"] == n_bars
|
|
assert result["signal_long"] == 0
|
|
assert result["signal_short"] == 0
|
|
|
|
@given(st.integers(min_value=1000, max_value=3000))
|
|
@settings(max_examples=50, deadline=5000)
|
|
def test_always_long_signal(self, n_bars):
|
|
"""Property: always-long signal has signal_long == n_bars."""
|
|
dates = pd.date_range("2024-01-01", periods=n_bars, freq="1min")
|
|
close = pd.Series(1.10 + np.arange(n_bars) * 0.0001, index=dates)
|
|
signal = pd.Series(1.0, index=dates)
|
|
result = backtest_signal(close, signal)
|
|
assert result["status"] == "success"
|
|
assert result["signal_long"] == n_bars
|
|
assert result["signal_neutral"] == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# N-Trades Invariants (10 tests)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestNTradesGroundTruth:
|
|
"""Property-based invariants for n_trades."""
|
|
|
|
@given(st.integers(min_value=1000, max_value=3000))
|
|
@settings(max_examples=100, deadline=5000)
|
|
def test_ntrades_non_negative(self, n_bars):
|
|
"""Property: n_trades >= 0."""
|
|
close, signal = _random_price_signal(n_bars, seed=42)
|
|
result = backtest_signal(close, signal)
|
|
if result["status"] == "success":
|
|
assert result["n_trades"] >= 0
|
|
|
|
@given(st.integers(min_value=1000, max_value=3000))
|
|
@settings(max_examples=50, deadline=5000)
|
|
def test_flat_signal_zero_trades(self, n_bars):
|
|
"""Property: all-flat signal yields n_trades == 0."""
|
|
dates = pd.date_range("2024-01-01", periods=n_bars, freq="1min")
|
|
rng = np.random.default_rng(42)
|
|
close = pd.Series(1.10 + rng.normal(0, 0.0002, n_bars).cumsum(), index=dates)
|
|
signal = pd.Series(0.0, index=dates)
|
|
result = backtest_signal(close, signal)
|
|
assert result["n_trades"] == 0
|
|
|
|
@given(st.integers(min_value=1000, max_value=3000))
|
|
@settings(max_examples=50, deadline=5000)
|
|
def test_ntrades_not_exceed_n_position_changes(self, n_bars):
|
|
"""Property: n_trades <= n_position_changes (trades are epochs)."""
|
|
close, signal = _random_price_signal(n_bars, seed=42)
|
|
result = backtest_signal(close, signal)
|
|
if result["status"] == "success":
|
|
assert result["n_trades"] <= result["n_position_changes"], \
|
|
f"n_trades={result['n_trades']} > n_position_changes={result['n_position_changes']}"
|
|
|
|
@given(
|
|
st.integers(min_value=1000, max_value=3000),
|
|
st.floats(min_value=0.0, max_value=50.0),
|
|
)
|
|
@settings(max_examples=70, deadline=5000)
|
|
def test_ntrades_with_cost(self, n_bars, cost):
|
|
"""Property: n_trades is unaffected by transaction cost."""
|
|
close, signal = _random_price_signal(n_bars, seed=42)
|
|
r0 = backtest_signal(close, signal, txn_cost_bps=0.0)
|
|
rc = backtest_signal(close, signal, txn_cost_bps=cost)
|
|
if r0["status"] == "success" and rc["status"] == "success":
|
|
assert r0["n_trades"] == rc["n_trades"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Data Quality / Edge Cases (8 tests)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDataQualityGroundTruth:
|
|
"""Property-based tests for data quality and edge cases."""
|
|
|
|
@given(st.integers(min_value=100, max_value=5000))
|
|
@settings(max_examples=100, deadline=5000)
|
|
def test_result_has_all_expected_keys(self, n_bars):
|
|
"""Property: backtest_signal returns all expected keys."""
|
|
close, signal = _random_price_signal(n_bars, seed=42)
|
|
result = backtest_signal(close, signal)
|
|
for k in ["status", "sharpe", "max_drawdown", "win_rate", "total_return",
|
|
"n_trades", "n_bars", "signal_long", "signal_short", "signal_neutral",
|
|
"annualized_return", "volatility", "profit_factor"]:
|
|
assert k in result, f"Missing key: {k}"
|
|
|
|
@given(st.text(min_size=1, max_size=50))
|
|
@settings(max_examples=30, deadline=5000)
|
|
def test_invalid_close_type_raises(self, bad_data):
|
|
"""Property: non-Series close raises TypeError."""
|
|
prices = list(range(100))
|
|
signal = pd.Series([1.0] * 100)
|
|
if not isinstance(prices, pd.Series):
|
|
with pytest.raises(TypeError):
|
|
backtest_signal(prices, signal)
|
|
|
|
@given(st.integers(min_value=0, max_value=1))
|
|
@settings(max_examples=20, deadline=5000)
|
|
def test_too_few_bars_fails(self, n_bars):
|
|
"""Property: fewer than 2 bars yields failed status or succeeds min-bars check."""
|
|
n_bars_safe = max(n_bars, 1)
|
|
dates = pd.date_range("2024-01-01", periods=n_bars_safe, freq="1min")
|
|
values = [1.10] * n_bars_safe
|
|
close = pd.Series(values, index=dates)
|
|
signal = pd.Series([0.0] * n_bars_safe, index=dates)
|
|
result = backtest_signal(close, signal)
|
|
assert result["status"] in ("success", "failed")
|
|
|
|
@given(st.integers(min_value=2, max_value=5000))
|
|
@settings(max_examples=50, deadline=5000)
|
|
def test_n_bars_reported_correctly(self, n_bars):
|
|
"""Property: n_bars equals the number of bars after processing."""
|
|
close, signal = _random_price_signal(n_bars, seed=42)
|
|
result = backtest_signal(close, signal)
|
|
if result["status"] == "success":
|
|
assert result["n_bars"] == n_bars, f"n_bars={result['n_bars']} != {n_bars}"
|