Repository Public
1"""The strategy must never see the bar it is filled on."""2from __future__ import annotations34import numpy as np5import pandas as pd67from conftest import make_data8from spq.engine import BacktestConfig, run_backtest9from spq.metrics import sharpe_ratio10from spq.robustness import lookahead_check11from spq.strategy import from_function1213FREE = BacktestConfig("1h", fee_bps=0, slippage_bps=0)141516def meta(lookback=30, universe=("X",)):17 return {"name": "t", "version": "1", "universe": list(universe), "interval": "1h", "lookback": lookback,18 "params": {}}192021def peek(d, p):22 """Long when the newest bar it can see closed above its open.2324 If that newest bar were the fill bar (bought at its open), this would be a money machine."""25 last = d["X"].iloc[-1]26 return {"X": 1.0 if last["close"] > last["open"] else 0.0}272829def test_inputs_end_at_the_decision_bar_and_fills_come_strictly_later():30 data = make_data(500, ("X",), seed=1)31 seen = []3233 def spy(d, p):34 seen.append((d["X"].index[0], d["X"].index[-1], len(d["X"])))35 return peek(d, p)3637 res = run_backtest(from_function(spy, meta(30)), data, FREE)38 idx = data["X"].index39 pos = {ts: i for i, ts in enumerate(idx)}40 assert len(seen) == len(res.decisions) == len(idx) - 2941 for (first, last, n), decided in zip(seen, res.decisions.index):42 assert last == decided and n == 30 and pos[last] - pos[first] == 29 # never more than `lookback` bars43 for _, t in res.trades.iterrows():44 assert pos[t["ts"]] == pos[t["decided_at"]] + 1 # filled on the NEXT bar45 assert t["price"] == data["X"]["open"].loc[t["ts"]] # at its open464748def test_a_strategy_that_tries_to_use_the_future_gains_nothing():49 """On a random walk with independent increments, knowing bar t says nothing about bar t+1.5051 The same rule gets rich as soon as the data are leaked by one bar (each row carrying the values of the52 next bar), which shows that the test would detect an engine that hands the fill bar to the strategy."""53 n = 600054 data = make_data(n, ("X",), seed=11, sigma=0.01)55 honest = run_backtest(from_function(peek, meta(5)), data, FREE)56 r = honest.returns.to_numpy()57 assert abs(sharpe_ratio(r)) < 0.05 # per-bar Sharpe; its standard error is 1/sqrt(n) = 0.01358 many = [sharpe_ratio(run_backtest(from_function(peek, meta(5)), make_data(n, ("X",), seed=100 + k), FREE)59 .returns.to_numpy()) for k in range(8)]60 assert abs(np.mean(many)) < 0.02 # no systematic edge across independent samples6162 df = data["X"]63 idx = df.index64 position = {ts: i for i, ts in enumerate(idx)}6566 def leaky(view): # what an engine with an off-by-one would do: the window ends on the fill bar67 end = position[view["X"].index[-1]]68 return peek({"X": df.iloc[max(0, end - 3): end + 2]}, {})6970 leaked = run_backtest(leaky, data, FREE, lookback=5)71 assert sharpe_ratio(leaked.returns.to_numpy()) > 0.572 assert leaked.equity.iloc[-1] > 1e6 * honest.equity.iloc[-1]737475def test_shifted_data_gives_shifted_results():76 data = make_data(800, ("X", "Y"), seed=21)7778 def trend(d, p):79 return {s: 0.5 if df["close"].iloc[-1] > df["close"].iloc[0] else 0.0 for s, df in d.items()}8081 s = from_function(trend, meta(48, ("X", "Y")))82 cfg = BacktestConfig("1h", fee_bps=10, slippage_bps=10, rebalance_threshold=0.02)83 base = run_backtest(s, data, cfg)8485 # 1. the same bars stamped 37 hours later: identical results, 37 hours later86 shift = pd.Timedelta(hours=37)87 later = {k: df.set_axis(df.index + shift) for k, df in data.items()}88 moved = run_backtest(s, later, cfg)89 assert moved.equity.index.equals(base.equity.index + shift)90 np.testing.assert_array_equal(moved.equity.to_numpy(), base.equity.to_numpy())91 assert list(moved.trades["ts"]) == list(base.trades["ts"] + shift)92 np.testing.assert_array_equal(moved.decisions.to_numpy(), base.decisions.to_numpy())9394 # 2. values delayed by k bars (row t holds what row t-k held): every decision and fill happens k bars later95 k = 596 delayed = {name: df.shift(k).iloc[k:] for name, df in data.items()}97 late = run_backtest(s, delayed, cfg)98 np.testing.assert_allclose(late.equity.to_numpy(), base.equity.to_numpy()[:-k], rtol=1e-12)99 assert list(late.trades["ts"]) == list((base.trades["ts"] + pd.Timedelta(hours=k))[base.trades["ts"] < base.equity.index[-k]])100101 # 3. dropping old history that is beyond the lookback changes nothing102 idx = data["X"].index103 full = run_backtest(s, data, cfg, start=idx[300])104 cut = run_backtest(s, {name: df.iloc[200:] for name, df in data.items()}, cfg, start=idx[300])105 pd.testing.assert_series_equal(full.equity, cut.equity)106107108def test_changing_the_future_does_not_change_the_past():109 data = make_data(600, ("X",), seed=31)110 s = from_function(peek, meta(10))111 base = run_backtest(s, data, BacktestConfig("1h"))112 t = 400113 altered = {"X": data["X"].copy()}114 rng = np.random.default_rng(0)115 altered["X"].iloc[t + 1:, :4] *= rng.uniform(0.5, 1.5, size=(600 - t - 1, 1))116 other = run_backtest(s, altered, BacktestConfig("1h"))117 idx = data["X"].index118 np.testing.assert_array_equal(other.equity.iloc[:t + 1].to_numpy(), base.equity.iloc[:t + 1].to_numpy())119 pd.testing.assert_frame_equal(other.decisions.loc[:idx[t]], base.decisions.loc[:idx[t]])120 assert not np.array_equal(other.equity.iloc[t + 2:].to_numpy(), base.equity.iloc[t + 2:].to_numpy())121122123def test_lookahead_check_passes_on_the_engine_and_fails_on_a_leaky_one():124 data = make_data(900, ("X",), seed=41)125 s = from_function(peek, meta(10))126 ok = lookahead_check(s, data, BacktestConfig("1h"))127 assert ok["status"] == "pass", ok128129 def leaky_runner(strategy, d, config, **kw):130 df = d["X"]131 position = {ts: i for i, ts in enumerate(df.index)}132133 def decide(view):134 end = position[view["X"].index[-1]]135 return strategy({"X": df.iloc[max(0, end - 8): end + 2]}) # one bar too many when it exists136137 return run_backtest(decide, d, config, start=kw.get("start"), end=kw.get("end"), lookback=10)138139 bad = lookahead_check(s, data, BacktestConfig("1h"), runner=leaky_runner)140 assert bad["status"] == "fail" and "decisions" in bad["detail"]141142 short = lookahead_check(s, make_data(8, ("X",)), BacktestConfig("1h"))143 assert short["status"] == "fail"144