| """ |
| walk_forward.py tests. test_frozen_selection_is_unaffected_by_anything_ |
| in_the_test_window is the single most important test in this file -- |
| it's a direct, executable proof of the leakage-safety claim in that |
| module's docstring, not just a check that functions run without error. |
| """ |
| import numpy as np |
| import pandas as pd |
|
|
| import config as cfg |
| import walk_forward as wf |
| from features import compute_indicators |
| from fixtures import make_synthetic_ohlcv |
| from moirai_model import QuantileForecast |
|
|
|
|
| def test_chronological_split_preserves_time_order_and_proportions(): |
| df = make_synthetic_ohlcv(n=1000, seed=70, freq="1h") |
| train, val, test = wf.chronological_split(df, cfg.WalkForwardConfig(train_frac=0.5, val_frac=0.3)) |
| assert len(train) + len(val) + len(test) == len(df) |
| assert train.index[-1] < val.index[0] <= val.index[-1] < test.index[0] |
| assert abs(len(train) / len(df) - 0.5) < 0.01 |
| assert abs(len(val) / len(df) - 0.3) < 0.01 |
|
|
|
|
| def test_chronological_split_never_shuffles(): |
| df = make_synthetic_ohlcv(n=500, seed=71, freq="1h") |
| train, val, test = wf.chronological_split(df) |
| for part in (train, val, test): |
| assert part.index.is_monotonic_increasing |
|
|
|
|
| def test_chronological_split_rejects_too_short_data(): |
| df = make_synthetic_ohlcv(n=20, seed=72, freq="1h") |
| try: |
| wf.chronological_split(df) |
| raise AssertionError("expected ValueError for too-short data") |
| except ValueError: |
| pass |
|
|
|
|
| class _NoisyStub: |
| """Raw probability is a noisy but genuinely informative function of |
| the actual forward return -- used to build a calibration dataset |
| that isn't pure garbage, without needing the real model.""" |
| def __init__(self, seed=0): |
| self.rng = np.random.default_rng(seed) |
|
|
| def predict(self, context_df, target_col, past_covariate_cols, prediction_length, freq): |
| last_close = float(context_df[target_col].iloc[-1]) |
| drift = self.rng.normal(0, 0.003) |
| p50 = last_close * (1 + drift) |
| spread = 0.004 |
| prices = [p50 * (1 - spread), p50 * (1 - spread / 2), p50, p50 * (1 + spread / 2), p50 * (1 + spread)] |
| return QuantileForecast(horizon=prediction_length, quantile_levels=[0.1, 0.25, 0.5, 0.75, 0.9], |
| quantiles=np.array([prices]), samples=None, |
| context_length_used=len(context_df), prediction_length=prediction_length) |
|
|
|
|
| def test_frozen_selection_is_unaffected_by_anything_in_the_test_window(): |
| """THE core leakage-safety test: two frames identical through the |
| train/val boundary but with completely different (differently |
| seeded, differently distributed) data after it must produce an |
| IDENTICAL calibrator and frozen DecisionConfig from |
| select_frozen_config -- since that function is never given the test |
| split at all, this is a structural guarantee, and this test proves |
| it holds in practice, not just in the code's own claim about |
| itself.""" |
| base = make_synthetic_ohlcv(n=900, seed=73, freq="1h") |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| wf_cfg = cfg.WalkForwardConfig(train_frac=0.5, val_frac=0.25, |
| min_probability_grid=(0.40, 0.50), buy_threshold_grid=(0.0003, 0.0006), |
| min_trades_for_calibration=10, min_trades_for_acceptance=1) |
|
|
| |
| feat_a = compute_indicators(base) |
| train_a, val_a, test_a = wf.chronological_split(feat_a, wf_cfg) |
|
|
| |
| |
| other_tail = make_synthetic_ohlcv(n=len(base), seed=999, freq="1h") |
| other_tail.index = base.index |
| combined = base.copy() |
| val_end = int(len(base) * (wf_cfg.train_frac + wf_cfg.val_frac)) |
| combined.iloc[val_end:] = other_tail.iloc[val_end:] |
| feat_b = compute_indicators(combined) |
| train_b, val_b, test_b = wf.chronological_split(feat_b, wf_cfg) |
|
|
| assert train_a.equals(train_b), "train split should be identical by construction" |
| assert val_a.equals(val_b), "val split should be identical by construction" |
| assert not test_a.equals(test_b), "test split MUST differ for this to be a meaningful test" |
|
|
| calib_a, dc_a, report_a = wf.select_frozen_config( |
| _NoisyStub(seed=5), train_a, val_a, timeframe="1h", asset_class="forex", |
| horizon=5, context_length=32, wf=wf_cfg) |
| calib_b, dc_b, report_b = wf.select_frozen_config( |
| _NoisyStub(seed=5), train_b, val_b, timeframe="1h", asset_class="forex", |
| horizon=5, context_length=32, wf=wf_cfg) |
|
|
| assert np.array_equal(calib_a.bin_edges, calib_b.bin_edges) |
| assert np.array_equal(calib_a.calibrated_values, calib_b.calibrated_values) |
| assert dc_a.min_probability == dc_b.min_probability |
| assert dc_a.buy_threshold == dc_b.buy_threshold |
| assert report_a["trials"] == report_b["trials"] |
|
|
|
|
| def test_select_frozen_config_grid_is_bounded_not_exhaustive_search(): |
| """Item 12: "do not search thousands of combinations" -- the number |
| of backtest runs select_frozen_config performs must equal exactly |
| len(min_probability_grid) * len(buy_threshold_grid), not something |
| open-ended.""" |
| df = make_synthetic_ohlcv(n=700, seed=74, freq="1h") |
| feat = compute_indicators(df) |
| wf_cfg = cfg.WalkForwardConfig(train_frac=0.5, val_frac=0.3, |
| min_probability_grid=(0.55, 0.60, 0.65), |
| buy_threshold_grid=(0.0015, 0.0025), |
| min_trades_for_calibration=10, min_trades_for_acceptance=1) |
| train, val, _ = wf.chronological_split(feat, wf_cfg) |
| _, _, report = wf.select_frozen_config(_NoisyStub(seed=1), train, val, timeframe="1h", |
| asset_class="forex", horizon=5, context_length=32, wf=wf_cfg) |
| assert len(report["trials"]) == 3 * 2 |
|
|
|
|
| def test_select_frozen_config_rejects_grid_points_below_min_trade_count(): |
| """Item 13: a combination profitable on a handful of trades must not |
| be selectable -- verified by setting min_trades_for_acceptance |
| absurdly high and confirming selection then fails outright rather |
| than quietly accepting a thin result.""" |
| df = make_synthetic_ohlcv(n=700, seed=75, freq="1h") |
| feat = compute_indicators(df) |
| wf_cfg = cfg.WalkForwardConfig(train_frac=0.5, val_frac=0.3, |
| min_probability_grid=(0.55,), buy_threshold_grid=(0.0015,), |
| min_trades_for_calibration=10, min_trades_for_acceptance=100_000) |
| train, val, _ = wf.chronological_split(feat, wf_cfg) |
| try: |
| wf.select_frozen_config(_NoisyStub(seed=2), train, val, timeframe="1h", |
| asset_class="forex", horizon=5, context_length=32, wf=wf_cfg) |
| raise AssertionError("expected ValueError: no grid point can reach 100,000 trades") |
| except ValueError as e: |
| assert "no (min_probability, buy_threshold)" in str(e) |
|
|
|
|
| def test_baseline_forecasters_run_through_the_real_backtest_engine(): |
| """Both baselines must be usable exactly like the real forecaster -- |
| same WalkForwardBacktester, same trading_summary -- for the |
| baseline comparison to mean anything.""" |
| df = make_synthetic_ohlcv(n=300, seed=76, freq="1h") |
| feat = compute_indicators(df) |
| dc = cfg.DecisionConfig(min_probability=0.55, buy_threshold=0.0005, sell_threshold=-0.0005) |
| zero_summary = wf.run_baseline(wf.ZeroSkillForecaster, feat, "1h", "forex", 5, 32, dc) |
| assert zero_summary.get("num_trades", 0) == 0, "zero-skill baseline predicts no change -> never clears any edge threshold -> always HOLD" |
|
|
| momentum_summary = wf.run_baseline(wf.NaiveMomentumForecaster, feat, "1h", "forex", 5, 32, dc) |
| assert "num_trades" in momentum_summary |
|
|
|
|
| def test_check_acceptance_rejects_on_insufficient_trades(): |
| summary = {"num_trades": 5, "net_pnl": 0.05, "profit_factor": 1.5, "max_drawdown": -0.02} |
| result = wf.check_acceptance(summary, {}, pd.Series([1.0, 1.01, 1.05]), |
| wf=cfg.WalkForwardConfig(min_trades_for_acceptance=30)) |
| assert result["accepted"] is False |
| assert result["checks"]["sufficient_trade_count"] is False |
|
|
|
|
| def test_check_acceptance_rejects_negative_net_return_even_with_many_trades(): |
| summary = {"num_trades": 100, "net_pnl": -0.05, "profit_factor": 0.8, "max_drawdown": -0.02} |
| result = wf.check_acceptance(summary, {}, pd.Series([1.0, 0.99, 0.95]), |
| wf=cfg.WalkForwardConfig(min_trades_for_acceptance=30)) |
| assert result["accepted"] is False |
| assert result["checks"]["positive_net_return"] is False |
| assert result["checks"]["profit_factor_above_1"] is False |
|
|
|
|
| def test_check_acceptance_rejects_when_baseline_wins(): |
| summary = {"num_trades": 50, "net_pnl": 0.01, "profit_factor": 1.1, "max_drawdown": -0.02} |
| baselines = {"momentum": {"net_pnl": 0.05}} |
| result = wf.check_acceptance(summary, baselines, pd.Series([1.0, 1.005, 1.01]), |
| wf=cfg.WalkForwardConfig(min_trades_for_acceptance=30)) |
| assert result["checks"]["beats_baseline_momentum"] is False |
| assert result["accepted"] is False |
|
|
|
|
| def test_check_acceptance_can_accept_a_genuinely_good_result(): |
| idx = pd.date_range("2024-01-01", periods=200, freq="1D") |
| equity = pd.Series(np.linspace(1.0, 1.15, 200), index=idx) |
| summary = {"num_trades": 50, "net_pnl": 0.15, "profit_factor": 1.8, "max_drawdown": -0.03} |
| baselines = {"zero_skill": {"net_pnl": 0.0}, "momentum": {"net_pnl": 0.02}} |
| result = wf.check_acceptance(summary, baselines, equity, wf=cfg.WalkForwardConfig(min_trades_for_acceptance=30)) |
| assert result["accepted"] is True, result["checks"] |
|
|
|
|
| def test_check_acceptance_rejects_when_one_period_explains_most_profit(): |
| idx = pd.date_range("2024-01-01", periods=200, freq="1D") |
| vals = np.ones(200) |
| vals[150:] = 1.30 |
| equity = pd.Series(vals, index=idx) |
| summary = {"num_trades": 50, "net_pnl": 0.30, "profit_factor": 5.0, "max_drawdown": -0.01} |
| result = wf.check_acceptance(summary, {}, equity, wf=cfg.WalkForwardConfig(min_trades_for_acceptance=30, |
| max_single_period_profit_share=0.70)) |
| assert result["checks"]["not_single_period_dependent"] is False |
| assert result["accepted"] is False |
|
|
