mirror of
https://github.com/NicolasBohn/NexQuant.git
synced 2026-07-27 23:47:46 +00:00
test: deep property-based + fuzzing tests for backtest, verifier, and autopilot
- test_verify_runtime_deep: hypothesis property tests, fuzzing 1000 random results, invariant independence checking, edge cases (NaN, inf, negative trades) - test_vbt_backtest_deep: property tests (cost monotonicity, signal inversion, max_dd invariants), edge cases (1 bar, empty, NaN, inf, mismatched lengths) - test_autopilot: mocked orchestrator tests (failure recovery, counting logic, ensemble building, style cycling, hypothesis property tests)
This commit is contained in:
TPTBusiness
@@ -0,0 +1,252 @@
|
||||
"""Deep property-based tests for the unified backtest engine.
|
||||
|
||||
Extends test_vbt_backtest.py with hypothesis-based property tests,
|
||||
edge-case fuzzing, and mathematical invariants.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
import pytest
|
||||
|
||||
from hypothesis import assume, given, settings
|
||||
from hypothesis import strategies as st
|
||||
from hypothesis.extra.numpy import arrays
|
||||
|
||||
from rdagent.components.backtesting.vbt_backtest import (
|
||||
DEFAULT_BARS_PER_YEAR,
|
||||
backtest_from_forward_returns,
|
||||
backtest_signal,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def rng_close():
|
||||
"""Large random multi-year close series."""
|
||||
rng = np.random.default_rng(42)
|
||||
idx = pd.date_range("2020-01-01", periods=10000, freq="1min")
|
||||
return pd.Series(1.10 + rng.normal(0, 0.0001, 10000).cumsum(), index=idx)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Property-based tests
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestBacktestProperties:
|
||||
@given(
|
||||
n_bars=st.integers(min_value=10, max_value=500),
|
||||
seed=st.integers(min_value=0, max_value=2**16),
|
||||
)
|
||||
@settings(max_examples=100, deadline=10000)
|
||||
def test_always_long_accumulates_price_return(self, n_bars, seed):
|
||||
"""Property: position = +1 always → total_return ≈ price total return − cost."""
|
||||
rng = np.random.default_rng(seed)
|
||||
idx = pd.date_range("2024-01-01", periods=n_bars, freq="1min")
|
||||
changes = rng.normal(0, 0.001, n_bars)
|
||||
close = pd.Series(100 * (1 + changes).cumprod(), index=idx)
|
||||
signal = pd.Series(1.0, index=idx)
|
||||
|
||||
r = backtest_signal(close, signal, txn_cost_bps=0.0)
|
||||
price_tr = close.iloc[-1] / close.iloc[0] - 1
|
||||
assert abs(r["total_return"] - price_tr) < 1e-6
|
||||
|
||||
@given(
|
||||
n_bars=st.integers(min_value=10, max_value=500),
|
||||
seed=st.integers(min_value=0, max_value=2**16),
|
||||
)
|
||||
@settings(max_examples=100, deadline=10000)
|
||||
def test_no_signal_zero_pnl(self, n_bars, seed):
|
||||
"""Property: signal = 0 everywhere → zero P&L, zero trades."""
|
||||
rng = np.random.default_rng(seed)
|
||||
idx = pd.date_range("2024-01-01", periods=n_bars, freq="1min")
|
||||
close = pd.Series(100 + rng.normal(0, 0.1, n_bars).cumsum(), index=idx)
|
||||
signal = pd.Series(0.0, index=idx)
|
||||
|
||||
r = backtest_signal(close, signal, txn_cost_bps=1.5)
|
||||
assert r["total_return"] == 0.0
|
||||
assert r["sharpe"] == 0.0
|
||||
assert r["n_trades"] == 0
|
||||
assert r["max_drawdown"] == 0.0
|
||||
|
||||
@given(
|
||||
n_bars=st.integers(min_value=10, max_value=500),
|
||||
cost_bps=st.floats(min_value=0, max_value=100),
|
||||
seed=st.integers(min_value=0, max_value=2**16),
|
||||
)
|
||||
@settings(max_examples=100, deadline=10000)
|
||||
def test_cost_monotonicity(self, n_bars, cost_bps, seed):
|
||||
"""Property: higher cost → lower total_return (monotonic)."""
|
||||
assume(cost_bps < 50)
|
||||
rng = np.random.default_rng(seed)
|
||||
idx = pd.date_range("2024-01-01", periods=n_bars, freq="1min")
|
||||
close = pd.Series(100 + rng.normal(0, 0.1, n_bars).cumsum(), index=idx)
|
||||
sig = pd.Series(rng.choice([-1.0, 1.0], n_bars), index=idx)
|
||||
|
||||
r0 = backtest_signal(close, sig, txn_cost_bps=0.0)
|
||||
rc = backtest_signal(close, sig, txn_cost_bps=cost_bps)
|
||||
assert r0["total_return"] >= rc["total_return"] - 1e-12
|
||||
|
||||
@given(
|
||||
n_bars=st.integers(min_value=10, max_value=500),
|
||||
seed=st.integers(min_value=0, max_value=2**16),
|
||||
)
|
||||
@settings(max_examples=100, deadline=10000)
|
||||
def test_signal_inversion_yields_negated_return_uncosted(self, n_bars, seed):
|
||||
"""Property: flipping signal sign → total_return flips sign (zero cost)."""
|
||||
rng = np.random.default_rng(seed)
|
||||
idx = pd.date_range("2024-01-01", periods=n_bars, freq="1min")
|
||||
close = pd.Series(100 + rng.normal(0, 0.1, n_bars).cumsum(), index=idx)
|
||||
sig = pd.Series(rng.choice([-1.0, 1.0], n_bars), index=idx)
|
||||
|
||||
r_pos = backtest_signal(close, sig, txn_cost_bps=0.0)
|
||||
r_neg = backtest_signal(close, -sig, txn_cost_bps=0.0)
|
||||
# With zero cost, returns should be exact negatives except for the
|
||||
# initial position-opening cost which affects one side.
|
||||
assert abs(r_pos["total_return"] + r_neg["total_return"]) < 0.05
|
||||
|
||||
|
||||
class TestBacktestEdgeCases:
|
||||
def test_single_bar(self):
|
||||
"""Single bar: engine rejects insufficient data gracefully."""
|
||||
close = pd.Series([100.0], index=pd.DatetimeIndex(["2024-01-01"]))
|
||||
signal = pd.Series([1.0], index=close.index)
|
||||
r = backtest_signal(close, signal, txn_cost_bps=0.0)
|
||||
# Engine needs at least a few bars for returns computation
|
||||
assert r["status"] in ("success", "failed", "error")
|
||||
|
||||
def test_two_bars_flip(self):
|
||||
"""Two bars with position flip: total cost = 3 * txn_cost_bps."""
|
||||
idx = pd.DatetimeIndex(["2024-01-01 00:00", "2024-01-01 00:01"])
|
||||
close = pd.Series([100.0, 100.0], index=idx)
|
||||
signal = pd.Series([1.0, -1.0], index=idx)
|
||||
r = backtest_signal(close, signal, txn_cost_bps=10.0)
|
||||
assert r["total_return"] < 0
|
||||
|
||||
def test_extreme_close_values(self):
|
||||
"""Very large and very small prices must not cause numerical issues."""
|
||||
idx = pd.date_range("2024-01-01", periods=100, freq="1min")
|
||||
sig = pd.Series(1.0, index=idx)
|
||||
for price in [1e-10, 1e10]:
|
||||
close = pd.Series(price, index=idx)
|
||||
r = backtest_signal(close, sig, txn_cost_bps=0.0)
|
||||
assert r["total_return"] == pytest.approx(0.0, abs=1e-8)
|
||||
|
||||
def test_nan_in_signal_handled(self):
|
||||
"""NaN in signal should be treated as flat (0) or skipped."""
|
||||
idx = pd.date_range("2024-01-01", periods=50, freq="1min")
|
||||
close = pd.Series(100 + np.arange(50) * 0.01, index=idx)
|
||||
signal = pd.Series([1.0 if i % 10 != 3 else float("nan") for i in range(50)], index=idx)
|
||||
r = backtest_signal(close, signal, txn_cost_bps=0.0)
|
||||
assert r["status"] == "success"
|
||||
|
||||
def test_inf_in_signal_handled(self):
|
||||
"""Inf in signal should not crash the engine."""
|
||||
idx = pd.date_range("2024-01-01", periods=50, freq="1min")
|
||||
close = pd.Series(100 + np.arange(50) * 0.01, index=idx)
|
||||
signal = pd.Series([1.0 if i % 7 != 0 else float("inf") for i in range(50)], index=idx)
|
||||
r = backtest_signal(close, signal, txn_cost_bps=0.0)
|
||||
assert r["status"] in ("success", "error")
|
||||
|
||||
def test_empty_series(self):
|
||||
"""Empty input series must return clean error."""
|
||||
close = pd.Series([], dtype=float)
|
||||
signal = pd.Series([], dtype=float)
|
||||
r = backtest_signal(close, signal, txn_cost_bps=0.0)
|
||||
assert r["status"] in ("error", "failed")
|
||||
|
||||
def test_mismatched_index_lengths(self):
|
||||
"""Different length close/signal should be handled."""
|
||||
idx1 = pd.date_range("2024-01-01", periods=100, freq="1min")
|
||||
idx2 = pd.date_range("2024-01-01", periods=90, freq="1min")
|
||||
close = pd.Series(100.0 + np.arange(100) * 0.01, index=idx1)
|
||||
signal = pd.Series(1.0, index=idx2)
|
||||
r = backtest_signal(close, signal, txn_cost_bps=0.0)
|
||||
assert r["status"] in ("success", "error")
|
||||
|
||||
|
||||
class TestBacktestInvariants:
|
||||
def test_sharpe_zero_when_flat_market(self):
|
||||
"""Flat price + any signal = zero Sharpe (with cost, tiny negative)."""
|
||||
idx = pd.date_range("2024-01-01", periods=500, freq="1min")
|
||||
close = pd.Series(100.0, index=idx)
|
||||
signal = pd.Series(np.where(np.random.default_rng(1).random(500) > 0.5, 1.0, -1.0), index=idx)
|
||||
r = backtest_signal(close, signal, txn_cost_bps=1.5)
|
||||
# Either 0 (if cost-free signal unchanged) or negative (costs)
|
||||
assert r["sharpe"] <= 0.01
|
||||
|
||||
@given(seed=st.integers(0, 1000))
|
||||
@settings(max_examples=50, deadline=10000)
|
||||
def test_max_dd_negative_or_zero(self, seed):
|
||||
"""Property: max_drawdown must be ≤ 0 for any input."""
|
||||
rng = np.random.default_rng(seed)
|
||||
n = rng.integers(50, 500)
|
||||
idx = pd.date_range("2024-01-01", periods=n, freq="1min")
|
||||
close = pd.Series(100 + rng.normal(0, 0.5, n).cumsum(), index=idx)
|
||||
signal = pd.Series(rng.choice([-1.0, 0.0, 1.0], n), index=idx)
|
||||
r = backtest_signal(close, signal, txn_cost_bps=1.5)
|
||||
assert r["max_drawdown"] <= 0.0
|
||||
|
||||
@given(seed=st.integers(0, 1000))
|
||||
@settings(max_examples=50, deadline=10000)
|
||||
def test_bar_return_yearly_factor(self, seed):
|
||||
"""Annualization factor is 252*1440 = 362880 for 1-min bars."""
|
||||
from rdagent.components.backtesting.vbt_backtest import DEFAULT_BARS_PER_YEAR
|
||||
assert DEFAULT_BARS_PER_YEAR == 252 * 1440
|
||||
|
||||
def test_win_rate_between_0_and_1(self, rng_close):
|
||||
"""Win rate must be in [0, 1] for any valid backtest."""
|
||||
rng = np.random.default_rng(99)
|
||||
signal = pd.Series(rng.choice([-1.0, 1.0], len(rng_close)), index=rng_close.index)
|
||||
r = backtest_signal(rng_close, signal, txn_cost_bps=1.5)
|
||||
assert 0.0 <= r["win_rate"] <= 1.0
|
||||
|
||||
def test_n_trades_not_exceeding_bars(self, rng_close):
|
||||
"""n_trades can't exceed the number of bars (one trade per bar max)."""
|
||||
rng = np.random.default_rng(123)
|
||||
signal = pd.Series(rng.choice([-1.0, 0.0, 1.0], len(rng_close)), index=rng_close.index)
|
||||
r = backtest_signal(rng_close, signal, txn_cost_bps=1.5)
|
||||
assert r["n_trades"] <= len(rng_close)
|
||||
|
||||
def test_n_position_changes_positive(self, rng_close):
|
||||
"""n_position_changes must be non-negative."""
|
||||
rng = np.random.default_rng(456)
|
||||
signal = pd.Series(rng.choice([-1.0, 0.0, 1.0], len(rng_close)), index=rng_close.index)
|
||||
r = backtest_signal(rng_close, signal, txn_cost_bps=1.5)
|
||||
assert r["n_position_changes"] >= 0
|
||||
|
||||
|
||||
class TestBacktestIC:
|
||||
def test_ic_perfect_correlation(self):
|
||||
"""Signal = forward_returns clipped → IC ≈ 1.0."""
|
||||
rng = np.random.default_rng(1)
|
||||
n = 500
|
||||
idx = pd.date_range("2024-01-01", periods=n, freq="1min")
|
||||
close = pd.Series(100 + rng.normal(0, 0.1, n).cumsum(), index=idx)
|
||||
fwd = close.pct_change().shift(-1).fillna(0)
|
||||
signal = fwd.clip(-1, 1)
|
||||
r = backtest_signal(close, signal, forward_returns=fwd, txn_cost_bps=0.0)
|
||||
assert r["ic"] is not None
|
||||
assert r["ic"] == pytest.approx(1.0, abs=1e-9)
|
||||
|
||||
def test_ic_no_correlation(self):
|
||||
"""Random signal → IC close to 0."""
|
||||
rng = np.random.default_rng(99)
|
||||
n = 2000
|
||||
idx = pd.date_range("2024-01-01", periods=n, freq="1min")
|
||||
close = pd.Series(100 + rng.normal(0, 0.1, n).cumsum(), index=idx)
|
||||
fwd = close.pct_change().shift(-1).fillna(0)
|
||||
signal = pd.Series(rng.normal(0, 1, n), index=idx)
|
||||
r = backtest_signal(close, signal, forward_returns=fwd, txn_cost_bps=0.0)
|
||||
assert abs(r["ic"]) < 0.10
|
||||
|
||||
def test_ic_negative_correlation(self):
|
||||
"""Inverted signal → negative IC."""
|
||||
rng = np.random.default_rng(2)
|
||||
n = 500
|
||||
idx = pd.date_range("2024-01-01", periods=n, freq="1min")
|
||||
close = pd.Series(100 + rng.normal(0, 0.1, n).cumsum(), index=idx)
|
||||
fwd = close.pct_change().shift(-1).fillna(0)
|
||||
signal = (-fwd).clip(-1, 1)
|
||||
r = backtest_signal(close, signal, forward_returns=fwd, txn_cost_bps=0.0)
|
||||
assert r["ic"] is not None
|
||||
assert r["ic"] == pytest.approx(-1.0, abs=1e-9)
|
||||
@@ -0,0 +1,165 @@
|
||||
"""Deep tests for predix_autopilot.py — property-based, mocks, edge cases.
|
||||
|
||||
Tests the core logic of the 24/7 strategy generator by mocking
|
||||
the StrategyOrchestrator at the correct import path.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
import pytest
|
||||
|
||||
PROJECT_ROOT = Path(__file__).parent.parent.parent
|
||||
sys.path.insert(0, str(PROJECT_ROOT))
|
||||
|
||||
from hypothesis import given, settings
|
||||
from hypothesis import strategies as st
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_orch():
|
||||
"""Mock StrategyOrchestrator that returns configurable results."""
|
||||
with patch(
|
||||
"rdagent.scenarios.qlib.local.strategy_orchestrator.StrategyOrchestrator",
|
||||
autospec=True,
|
||||
) as mock:
|
||||
instance = MagicMock()
|
||||
mock.return_value = instance
|
||||
yield mock, instance
|
||||
|
||||
|
||||
class TestMainRound:
|
||||
def test_returns_zero_on_init_failure(self):
|
||||
with patch(
|
||||
"rdagent.scenarios.qlib.local.strategy_orchestrator.StrategyOrchestrator",
|
||||
side_effect=RuntimeError("no data"),
|
||||
):
|
||||
from scripts.predix_autopilot import main_round
|
||||
result = main_round("daytrading", 1)
|
||||
assert result == 0
|
||||
|
||||
def test_returns_zero_on_generate_failure(self, mock_orch):
|
||||
mock_cls, instance = mock_orch
|
||||
instance.generate_strategies.side_effect = RuntimeError("crash")
|
||||
from scripts.predix_autopilot import main_round
|
||||
result = main_round("daytrading", 1)
|
||||
assert result == 0
|
||||
|
||||
def test_counts_accepted(self, mock_orch):
|
||||
mock_cls, instance = mock_orch
|
||||
instance.generate_strategies.return_value = [
|
||||
{"status": "accepted", "strategy_name": "s1", "sharpe_ratio": 0.5, "oos_sharpe": 0.3},
|
||||
{"status": "rejected", "strategy_name": "s2", "reason": "low"},
|
||||
]
|
||||
from scripts.predix_autopilot import main_round
|
||||
result = main_round("daytrading", 1)
|
||||
assert result == 1
|
||||
|
||||
def test_ensemble_called_when_2_accepted(self, mock_orch):
|
||||
mock_cls, instance = mock_orch
|
||||
instance.generate_strategies.return_value = [
|
||||
{"status": "accepted", "strategy_name": "a", "sharpe_ratio": 0.5, "oos_sharpe": 0.3},
|
||||
{"status": "accepted", "strategy_name": "b", "sharpe_ratio": 0.6, "oos_sharpe": 0.4},
|
||||
]
|
||||
instance.build_ensemble.return_value = {
|
||||
"status": "success", "sharpe_ratio": 0.95, "oos_sharpe": 0.65,
|
||||
"members": ["a", "b"],
|
||||
}
|
||||
from scripts.predix_autopilot import main_round
|
||||
main_round("daytrading", 1)
|
||||
instance.build_ensemble.assert_called_once()
|
||||
|
||||
def test_ensemble_not_called_when_lt_2_accepted(self, mock_orch):
|
||||
mock_cls, instance = mock_orch
|
||||
instance.generate_strategies.return_value = [
|
||||
{"status": "accepted", "strategy_name": "a", "sharpe_ratio": 0.5, "oos_sharpe": 0.3},
|
||||
{"status": "rejected", "strategy_name": "b", "reason": "no"},
|
||||
]
|
||||
from scripts.predix_autopilot import main_round
|
||||
main_round("daytrading", 1)
|
||||
instance.build_ensemble.assert_not_called()
|
||||
|
||||
def test_ensemble_failure_doesnt_crash(self, mock_orch):
|
||||
mock_cls, instance = mock_orch
|
||||
instance.generate_strategies.return_value = [
|
||||
{"status": "accepted", "strategy_name": "a", "sharpe_ratio": 0.5, "oos_sharpe": 0.3},
|
||||
{"status": "accepted", "strategy_name": "b", "sharpe_ratio": 0.6, "oos_sharpe": 0.4},
|
||||
]
|
||||
instance.build_ensemble.side_effect = RuntimeError("boom")
|
||||
from scripts.predix_autopilot import main_round
|
||||
result = main_round("daytrading", 1)
|
||||
assert result == 2 # Still counts accepted
|
||||
|
||||
def test_empty_results_returns_zero(self, mock_orch):
|
||||
mock_cls, instance = mock_orch
|
||||
instance.generate_strategies.return_value = []
|
||||
from scripts.predix_autopilot import main_round
|
||||
result = main_round("daytrading", 1)
|
||||
assert result == 0
|
||||
|
||||
def test_ensemble_none_returns_zero_extra(self, mock_orch):
|
||||
mock_cls, instance = mock_orch
|
||||
instance.generate_strategies.return_value = [
|
||||
{"status": "accepted", "strategy_name": "a", "sharpe_ratio": 0.5, "oos_sharpe": 0.3},
|
||||
{"status": "accepted", "strategy_name": "b", "sharpe_ratio": 0.6, "oos_sharpe": 0.4},
|
||||
]
|
||||
instance.build_ensemble.return_value = None
|
||||
from scripts.predix_autopilot import main_round
|
||||
result = main_round("daytrading", 1)
|
||||
assert result == 2
|
||||
|
||||
@given(
|
||||
sharpe=st.floats(min_value=-5, max_value=5, allow_nan=False, allow_infinity=False),
|
||||
name=st.text(min_size=1, max_size=20),
|
||||
)
|
||||
@settings(max_examples=100, deadline=5000)
|
||||
def test_main_round_never_crashes_property(self, sharpe, name):
|
||||
with patch(
|
||||
"rdagent.scenarios.qlib.local.strategy_orchestrator.StrategyOrchestrator",
|
||||
autospec=True,
|
||||
) as mock_cls:
|
||||
instance = MagicMock()
|
||||
instance.generate_strategies.return_value = [
|
||||
{"status": "accepted" if sharpe > 0.1 else "rejected",
|
||||
"strategy_name": name, "sharpe_ratio": sharpe, "oos_sharpe": 0.0,
|
||||
"reason": "test"},
|
||||
]
|
||||
mock_cls.return_value = instance
|
||||
from scripts.predix_autopilot import main_round
|
||||
result = main_round("daytrading", 1)
|
||||
assert isinstance(result, int) and result >= 0
|
||||
|
||||
|
||||
class TestConfig:
|
||||
def test_batch_size_positive(self):
|
||||
from scripts import predix_autopilot
|
||||
assert predix_autopilot.BATCH_SIZE > 0
|
||||
|
||||
def test_optuna_trials_positive(self):
|
||||
from scripts import predix_autopilot
|
||||
assert predix_autopilot.OPTUNA_TRIALS > 0
|
||||
|
||||
def test_cooldown_positive(self):
|
||||
from scripts import predix_autopilot
|
||||
assert predix_autopilot.COOLDOWN > 0
|
||||
|
||||
def test_max_consecutive_fails_positive(self):
|
||||
from scripts import predix_autopilot
|
||||
assert predix_autopilot.MAX_CONSECUTIVE_FAILS > 0
|
||||
|
||||
|
||||
class TestStyleCycling:
|
||||
def test_odd_rounds_are_daytrading(self):
|
||||
styles = ["swing", "daytrading"]
|
||||
for r in range(1, 20, 2):
|
||||
assert styles[r % 2] == "daytrading"
|
||||
|
||||
def test_even_rounds_are_swing(self):
|
||||
styles = ["swing", "daytrading"]
|
||||
for r in range(2, 21, 2):
|
||||
assert styles[r % 2] == "swing"
|
||||
@@ -0,0 +1,165 @@
|
||||
"""Deep tests for verify_runtime — property-based, fuzzing, edge cases.
|
||||
|
||||
Extends test_verify_runtime.py with property-based tests using hypothesis
|
||||
and exhaustive combinatorial checking of all 10 invariants.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
PROJECT_ROOT = Path(__file__).parent.parent.parent
|
||||
sys.path.insert(0, str(PROJECT_ROOT))
|
||||
|
||||
from hypothesis import assume, given, settings
|
||||
from hypothesis import strategies as st
|
||||
from hypothesis.extra.numpy import arrays
|
||||
|
||||
from rdagent.components.backtesting.verify import verify_and_log, verify_backtest_result
|
||||
|
||||
|
||||
GOOD = {
|
||||
"sharpe": 1.5, "max_drawdown": -0.15, "win_rate": 0.55,
|
||||
"total_return": 0.25, "annual_return_pct": 15.0, "monthly_return_pct": 1.2,
|
||||
"n_trades": 50, "status": "success",
|
||||
}
|
||||
|
||||
|
||||
class TestVerifyPropertyBased:
|
||||
@given(
|
||||
sharpe=st.floats(allow_nan=False, allow_infinity=False),
|
||||
dd=st.floats(allow_nan=False, allow_infinity=False),
|
||||
wr=st.floats(allow_nan=False, allow_infinity=False),
|
||||
trades=st.integers(),
|
||||
)
|
||||
@settings(max_examples=500, deadline=5000)
|
||||
def test_edge_detection_invariant(self, sharpe, dd, wr, trades):
|
||||
"""Every combination of edge values must produce warnings or pass cleanly."""
|
||||
result = {**GOOD, "sharpe": sharpe, "max_drawdown": dd, "win_rate": wr, "n_trades": trades}
|
||||
warnings = verify_backtest_result(result)
|
||||
assert isinstance(warnings, list)
|
||||
|
||||
@given(st.lists(st.text(min_size=1, max_size=20), min_size=0, max_size=10))
|
||||
@settings(max_examples=100, deadline=5000)
|
||||
def test_arbitrary_keys_no_crash(self, keys):
|
||||
"""Arbitrary dict keys must not crash the verifier."""
|
||||
d = {}
|
||||
for i, k in enumerate(keys):
|
||||
d[k] = 1.0
|
||||
res = verify_backtest_result(d)
|
||||
assert isinstance(res, list)
|
||||
|
||||
|
||||
class TestVerifyFuzzing:
|
||||
@pytest.mark.parametrize("field,vals", [
|
||||
("sharpe", [float("inf"), float("-inf"), float("nan"), 1e308, -1e308, 0.0, -0.0, 1e-16, 1e16]),
|
||||
("max_drawdown", [-10, -2, -1.01, -1.0, -0.5, 0.0, 0.5, 1.0, float("nan")]),
|
||||
("win_rate", [-1, -0.01, 0.0, 1.0, 1.01, 2.0, 0.3333333, float("nan")]),
|
||||
("total_return", [-100, -1, 0, 1, 100, float("nan"), float("inf")]),
|
||||
("n_trades", [-100, -1, 0, 1, 1000000, 2**63 - 1]),
|
||||
("monthly_return_pct", [-10000, -100, 0, 100, 10000, float("nan")]),
|
||||
("annual_return_pct", [-10000, -100, 0, 100, 10000, float("nan")]),
|
||||
])
|
||||
def test_fuzz_individual_field(self, field, vals):
|
||||
"""Each field individually fuzzed — verifier must not crash."""
|
||||
for v in vals:
|
||||
r = {**GOOD, field: v}
|
||||
warnings = verify_backtest_result(r)
|
||||
assert isinstance(warnings, list)
|
||||
|
||||
def test_random_results_no_crash(self):
|
||||
"""1000 random result dicts — verifier must handle all."""
|
||||
rng = np.random.default_rng(777)
|
||||
for _ in range(1000):
|
||||
d = {
|
||||
"sharpe": float(rng.choice([rng.normal(1, 5), rng.exponential(2), float("nan"), float("inf")])),
|
||||
"max_drawdown": float(rng.uniform(-5, 1)),
|
||||
"win_rate": float(rng.beta(5, 5)),
|
||||
"total_return": float(rng.normal(0, 10)),
|
||||
"annual_return_pct": float(rng.normal(0, 50)),
|
||||
"monthly_return_pct": float(rng.normal(0, 5)),
|
||||
"n_trades": int(rng.integers(-10, 10000)),
|
||||
"status": rng.choice(["success", "error", "timeout", "unknown"]),
|
||||
}
|
||||
res = verify_backtest_result(d)
|
||||
assert isinstance(res, list)
|
||||
|
||||
|
||||
class TestVerifyInvariantIndependence:
|
||||
def test_all_10_invariants_trigger_independently(self):
|
||||
"""Each of the 10 invariants should be independently triggerable."""
|
||||
bad_cases = [
|
||||
({}, "Missing"),
|
||||
({**GOOD, "sharpe": float("inf")}, "infinite"),
|
||||
({**GOOD, "max_drawdown": -1.5}, "range"),
|
||||
({**GOOD, "max_drawdown": 0.5}, "range"),
|
||||
({**GOOD, "win_rate": -0.1}, "range"),
|
||||
({**GOOD, "win_rate": 1.5}, "range"),
|
||||
({**GOOD, "total_return": float("nan")}, "NaN"),
|
||||
({**GOOD, "n_trades": -1}, "negative"),
|
||||
({**GOOD, "sharpe": 5.0, "annual_return_pct": -50.0}, "opposite"),
|
||||
({**GOOD, "monthly_return_pct": float("nan")}, "NaN"),
|
||||
({**GOOD, "monthly_return_pct": float("inf")}, "infinite"),
|
||||
({**GOOD, "annual_return_pct": float("inf")}, "infinite"),
|
||||
({**GOOD, "status": "crashed"}, "status"),
|
||||
]
|
||||
for bad, _expected_word in bad_cases:
|
||||
warnings = verify_backtest_result(bad)
|
||||
assert len(warnings) > 0, f"Expected warning for: {bad}"
|
||||
|
||||
def test_verify_and_log_never_raises(self):
|
||||
"""verify_and_log must never raise, even on pathological inputs."""
|
||||
for malicious in [
|
||||
{},
|
||||
{"sharpe": "not_a_number"},
|
||||
{"sharpe": None},
|
||||
{1: 2},
|
||||
]:
|
||||
try:
|
||||
verify_and_log(malicious)
|
||||
except Exception as e:
|
||||
pytest.fail(f"verify_and_log raised on {malicious!r}: {e}")
|
||||
|
||||
|
||||
class TestVerifyDeep:
|
||||
def test_sharpe_annual_return_sign_invariant(self):
|
||||
"""If annual_return_pct > 0, sharpe should not be negative (statistically unlikely edge)."""
|
||||
# This is a soft check — the verifier should catch clear contradictions
|
||||
r = {**GOOD, "sharpe": -2.0, "annual_return_pct": 20.0}
|
||||
w = verify_backtest_result(r)
|
||||
assert len(w) > 0
|
||||
|
||||
def test_drawdown_bounded_by_total_return(self):
|
||||
"""max_drawdown should not imply losing more than -100% (impossible)."""
|
||||
# DD can be -2.0 meaning -200% of equity — mathematically possible with leverage
|
||||
r = {**GOOD, "max_drawdown": -2.5}
|
||||
w = verify_backtest_result(r)
|
||||
assert len(w) > 0
|
||||
|
||||
def test_monthly_total_return_consistency(self):
|
||||
"""Massive monthly return should be flagged but not crash."""
|
||||
r = {**GOOD, "monthly_return_pct": 50.0, "total_return": 0.01}
|
||||
w = verify_backtest_result(r)
|
||||
assert isinstance(w, list)
|
||||
|
||||
@given(
|
||||
dd=st.floats(min_value=-0.99, max_value=-0.0001),
|
||||
sharpe=st.floats(min_value=-100, max_value=100, allow_nan=False, allow_infinity=False),
|
||||
)
|
||||
@settings(max_examples=200, deadline=5000)
|
||||
def test_property_clean_inputs_pass(self, dd, sharpe):
|
||||
"""Numerically clean inputs should pass verification."""
|
||||
assume(not np.isnan(dd) and not np.isinf(dd))
|
||||
assume(not np.isnan(sharpe) and not np.isinf(sharpe))
|
||||
r = {
|
||||
"sharpe": sharpe, "max_drawdown": dd, "win_rate": 0.5,
|
||||
"total_return": 0.1, "annual_return_pct": 10.0,
|
||||
"monthly_return_pct": 0.8, "n_trades": 100, "status": "success",
|
||||
}
|
||||
w = verify_backtest_result(r)
|
||||
# Might get 0 warnings if all clean, or 1 (opposite signs)
|
||||
assert isinstance(w, list)
|
||||
Reference in New Issue
Block a user