Repository Public
1"""Split enforcement and the robustness toolbox."""2from __future__ import annotations34import json56import numpy as np7import pandas as pd8import pytest910from conftest import make_data11from spq.engine import BacktestConfig12from spq.research import (Experiment, Splits, TestSplitError, cartesian_grid, one_at_a_time_grid, read_ledger)13from spq.robustness import (_stationary_indices, bootstrap_sharpe_ci, drawdown_episodes, monte_carlo,14 parameter_sensitivity, select_params, stress_windows, walk_forward)15from spq.strategy import from_function1617CFG = BacktestConfig("1h", rebalance_threshold=0.05)181920def trend(data, params):21 w = int(params["window"])22 return {s: (1.0 / len(data) if df["close"].iloc[-1] > df["close"].iloc[-w] else 0.0) for s, df in data.items()}232425def make_strategy(name="trend", window=24):26 return from_function(trend, {"name": name, "version": "1", "universe": ["AAA", "BBB"], "interval": "1h",27 "lookback": 100, "params": {"window": window}})282930@pytest.fixture31def exp(tmp_path):32 return Experiment(make_data(2000, seed=3), CFG, ledger=tmp_path / "ledger.jsonl")333435def test_splits_are_chronological_and_disjoint():36 idx = pd.date_range("2024-01-01", periods=1000, freq="1h", tz="UTC")37 s = Splits.chronological(idx)38 assert s.train == (idx[0], idx[600]) and s.validation == (idx[600], idx[800])39 assert s.test == (idx[800], idx[-1] + pd.Timedelta(hours=1))40 assert s.window("development") == (idx[0], idx[800])41 rep = s.to_report()42 assert rep["train"] == [idx[0].isoformat(), idx[599].isoformat()]43 assert rep["validation"] == [idx[600].isoformat(), idx[799].isoformat()]44 assert rep["test"] == [idx[800].isoformat(), idx[999].isoformat()]45 assert Splits.chronological(idx, (0.5, 0.25, 0.25)).test[0] == idx[750]46 for bad in ((0.6, 0.2), (0.6, 0.2, 0.3), (0.8, 0.2, 0.0)):47 with pytest.raises(ValueError):48 Splits.chronological(idx, bad)495051def test_selection_code_cannot_obtain_test_bars(exp):52 s = make_strategy()53 test_start = exp.splits.test[0]54 assert all(df.index.max() < test_start for df in exp.dev_data().values())55 with pytest.raises(TestSplitError, match="final_test"):56 exp.run(s, "test")57 with pytest.raises(TestSplitError):58 exp.split_returns(exp.trial(s), "test")59 with pytest.raises(TestSplitError, match="reaches into the test split"):60 exp.run_window(s, exp.splits.validation[0], test_start + pd.Timedelta(hours=1))61 for split in ("train", "validation", "development"):62 res = exp.run(s, split)63 assert res.equity.index.max() < test_start64 assert exp.trial(s).equity.index.max() < test_start65 with pytest.raises(ValueError, match="train_\\* or validation_\\*"):66 select_params([{"params": {}, "test_sharpe": 1.0}], metric="test_sharpe")676869def test_test_split_is_evaluated_once_then_selection_is_closed(exp):70 s = make_strategy()71 exp.trial(s, {"window": 12})72 exp.trial(s, {"window": 48})73 final = exp.final_test(s, {"window": 12})74 assert final.equity.index[0] == exp.splits.test[0] and final.equity.index[-1] == exp.splits.last["test"]75 # the same evaluation can be read again, it is not a new look76 again = exp.final_test(s, {"window": 12})77 pd.testing.assert_series_equal(final.equity, again.equity)78 assert len(exp.test_evaluations) == 179 # another parameter set, or another version under the same name, is refused80 with pytest.raises(TestSplitError, match="already evaluated"):81 exp.final_test(s, {"window": 48})82 with pytest.raises(TestSplitError, match="already evaluated"):83 exp.final_test(s) # META params (window 24) differ from the sealed ones84 # and no more tuning for that strategy85 for call in (lambda: exp.trial(s, {"window": 36}), lambda: exp.run(s, "validation"),86 lambda: exp.run_window(s, exp.splits.train[0], exp.splits.validation[0]),87 lambda: parameter_sensitivity(exp, s, [{"window": 36}]),88 lambda: walk_forward(exp, s, [{"window": 36}])):89 with pytest.raises(TestSplitError, match="selection is closed"):90 call()91 # another strategy is not affected; benchmarks have no parameters and stay available92 other = make_strategy("other")93 exp.trial(other)94 assert set(exp.benchmarks("test")) == {"buy_and_hold", "equal_weight", "buy_and_hold_AAA", "buy_and_hold_BBB"}959697def test_trials_are_counted_once_and_written_to_the_ledger(exp):98 s = make_strategy()99 rows = parameter_sensitivity(exp, s, [{"window": 12}, {"window": 24}, {"window": 48}, {"window": 24}])100 assert len(rows) == 4 and len(exp.trials) == 3 # the repeated configuration is not a new trial101 assert all({"train_sharpe", "validation_sharpe", "validation_return", "validation_trades"} <= set(r) for r in rows)102 assert len(exp.trial_sharpes("validation")) == 3103 best = select_params(rows)104 assert best == max(rows, key=lambda r: r["validation_sharpe"])["params"]105 exp.final_test(s, best)106 ledger = read_ledger(exp.ledger_path)107 assert [r["kind"] for r in ledger] == ["trial", "trial", "trial", "test"]108 assert ledger[0]["params"] == {"window": 12} and ledger[0]["validation_bars"] > 0109 assert json.loads(exp.ledger_path.read_text().splitlines()[-1])["params"] == best110 # a failing configuration is reported, not fatal111 exp2 = Experiment(make_data(600, seed=1), CFG)112 rows = parameter_sensitivity(exp2, make_strategy(), [{"window": 24}, {"window": 5000}])113 assert "error" in rows[1] and "IndexError" in rows[1]["error"] and "validation_sharpe" in rows[0]114115116def test_grids():117 assert one_at_a_time_grid({"fast": 24, "slow": 168, "z": 2.0, "flag": True, "pairs": [["A", "B"]]}) == [118 {"fast": 24, "slow": 168, "z": 2.0, "flag": True, "pairs": [["A", "B"]]},119 {"fast": 12, "slow": 168, "z": 2.0, "flag": True, "pairs": [["A", "B"]]},120 {"fast": 48, "slow": 168, "z": 2.0, "flag": True, "pairs": [["A", "B"]]},121 {"fast": 24, "slow": 84, "z": 2.0, "flag": True, "pairs": [["A", "B"]]},122 {"fast": 24, "slow": 336, "z": 2.0, "flag": True, "pairs": [["A", "B"]]},123 {"fast": 24, "slow": 168, "z": 1.0, "flag": True, "pairs": [["A", "B"]]},124 {"fast": 24, "slow": 168, "z": 4.0, "flag": True, "pairs": [["A", "B"]]},125 ]126 grid = cartesian_grid({"fast": 24, "slow": 168}, {"fast": [12, 24], "slow": [168, 336]})127 assert len(grid) == 4 and grid[0] == {"fast": 24, "slow": 168}128 with pytest.raises(ValueError, match="not in META"):129 cartesian_grid({"fast": 24}, {"nope": [1]})130131132def test_walk_forward_folds_are_ordered_and_inside_development(exp):133 s = make_strategy()134 grid = [{"window": 12}, {"window": 24}, {"window": 48}]135 folds = walk_forward(exp, s, grid, n_folds=4)136 assert len(folds) == 4137 test_start = exp.splits.test[0]138 for f in folds:139 is_a, is_b = map(pd.Timestamp, f["in_sample"])140 oos_a, oos_b = map(pd.Timestamp, f["out_of_sample"])141 assert is_a < is_b == oos_a < oos_b <= test_start142 assert f["params"] in grid and "out_of_sample_sharpe" in f and "out_of_sample_return" in f143 assert [f["out_of_sample"][0] for f in folds] == sorted(f["out_of_sample"][0] for f in folds)144 assert len(exp.trials) == 3 # walk-forward reuses the three development runs145146147def test_stationary_bootstrap_blocks_and_interval():148 rng = np.random.default_rng(0)149 idx = _stationary_indices(500, 200, 20, rng)150 assert idx.shape == (200, 500) and idx.min() >= 0 and idx.max() < 500151 consecutive = (np.diff(idx, axis=1) % 500 == 1).mean()152 assert consecutive == pytest.approx(1 - 1 / 20, abs=0.01) # a new block starts with probability 1/20153154 r = np.random.default_rng(1).normal(0.0005, 0.01, 4000)155 out = bootstrap_sharpe_ci(r, "1h", n_samples=1000, seed=7)156 lo, hi = out["ci"]157 point = r.mean() / r.std(ddof=1) * np.sqrt(8760)158 assert lo < point < hi and out["mean_block"] == 24 and out["n_samples"] == 1000159 half_width = 1.96 * np.sqrt(8760 / 4000) # asymptotic standard error of a Sharpe ratio160 assert hi - lo == pytest.approx(2 * half_width, rel=0.15)161 assert bootstrap_sharpe_ci(r, "1h", n_samples=1000, seed=7)["ci"] == [lo, hi] # reproducible162 assert bootstrap_sharpe_ci(np.zeros(100), "1h")["ci"] == [None, None]163 assert bootstrap_sharpe_ci(r[:5], "1h")["ci"] == [None, None]164165166def test_monte_carlo_summaries():167 r = np.random.default_rng(2).normal(0.0002, 0.01, 1500)168 lots = np.array([5.0, -2.0, 3.0, -1.0, 4.0, -6.0, 2.0])169 mc = monte_carlo(r, lots, 1000.0, n_samples=1000, seed=3)170 b, t = mc["bar_returns"], mc["lots"]171 assert b["total_return_p05"] < b["total_return_p50"] < b["total_return_p95"]172 assert 0 <= b["probability_of_loss"] <= 1 and 0 < b["max_drawdown_p50"] <= b["max_drawdown_p95"] < 1173 assert b["total_return_p50"] == pytest.approx(np.exp(np.log1p(r).sum()) - 1, abs=0.15)174 assert t["n_lots"] == 7 and t["pnl_fraction_p05"] < t["pnl_fraction_p50"] < t["pnl_fraction_p95"]175 assert t["pnl_fraction_p50"] == pytest.approx(lots.sum() / 1000, abs=0.004)176 assert monte_carlo(r[:5], lots[:2], 1000.0) == {"n_samples": 2000, "bar_returns": None, "lots": None}177178179def test_stress_windows_are_the_benchmark_drawdowns():180 idx = pd.date_range("2024-01-01", periods=9, freq="1h", tz="UTC")181 bench = pd.Series([100, 120, 90, 96, 130, 65, 70, 140, 133], index=idx, dtype=float)182 eps = drawdown_episodes(bench, top=3)183 assert [(e["peak"], e["trough"]) for e in eps] == [(idx[4], idx[5]), (idx[1], idx[2]), (idx[7], idx[8])]184 assert [e["depth"] for e in eps] == pytest.approx([0.5, 0.25, 0.05])185 strat = pd.Series([100, 100, 100, 100, 110, 99, 99, 99, 99], index=idx, dtype=float)186 out = stress_windows(strat, bench, top=2)187 assert out[0] == {"window": [idx[4].isoformat(), idx[5].isoformat()], "benchmark_return": pytest.approx(-0.5),188 "strategy_return": pytest.approx(-0.1), "strategy_max_drawdown": pytest.approx(0.1)}189 assert out[1]["strategy_return"] == 0 and out[1]["benchmark_return"] == pytest.approx(-0.25)190