""" 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") # Loose grid values here are deliberate: this test isolates the # leakage-safety question ("is the SELECTED config independent of # test-region data") from the strategy-quality question (already # covered by test_select_frozen_config_rejects_grid_points_below_ # min_trade_count) -- a stub with genuinely zero real skill gets # correctly calibrated toward the base rate by design, so a # realistic min_probability like 0.55-0.70 would legitimately find # no acceptable config here, which is correct behavior but leaves # nothing to compare across variants A/B. 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) # Variant A: base data, unmodified. feat_a = compute_indicators(base) train_a, val_a, test_a = wf.chronological_split(feat_a, wf_cfg) # Variant B: identical through train+val, then a totally different # (differently-seeded) tail spliced in as the "test" region. other_tail = make_synthetic_ohlcv(n=len(base), seed=999, freq="1h") other_tail.index = base.index # align so concat/slicing stays simple 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 # may or may not trade depending on synthetic noise; just must run cleanly 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") # ~6.7 months -> several distinct 30D bins equity = pd.Series(np.linspace(1.0, 1.15, 200), index=idx) # smooth, steady growth -- no single-period dependency 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") # ~6.7 months -> several distinct 30D bins vals = np.ones(200) vals[150:] = 1.30 # one sharp jump near the end, flat everywhere else 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