Repository Public
1"""Fill timing and cost accounting of the backtester, on hand-written bars."""2from __future__ import annotations34import numpy as np5import pandas as pd6import pytest78from conftest import frame, make_data9from spq.data import DataError10from spq.engine import BacktestConfig, run_backtest11from spq.strategy import WeightsError, from_function1213META = {"name": "t", "version": "1", "universe": ["X"], "interval": "1h", "lookback": 1, "params": {}}141516def strat(func, lookback=1, universe=("X",)):17 return from_function(func, {**META, "lookback": lookback, "universe": list(universe)})181920def test_fill_at_next_open_with_costs_by_hand():21 # bar: 0 1 2 3 422 data = {"X": frame([100, 102, 104, 108, 110], [101, 103, 107, 109, 111])}23 idx = data["X"].index24 # long after seeing the close of bar 0, flat after seeing the close of bar 225 s = strat(lambda d, p: {"X": 1.0 if len(d["X"]) < 3 else 0.0}, lookback=5)26 s.meta["lookback"] = 127 seen = []2829 def f(d, p):30 seen.append(d["X"].index[-1])31 return {"X": 1.0 if d["X"].index[-1] < idx[2] else 0.0}3233 cfg = BacktestConfig("1h", fee_bps=10, slippage_bps=20, initial_cash=1000)34 res = run_backtest(strat(f), data, cfg)35 fee, slip = 0.001, 0.0023637 # decision on bar 0 -> buy at the OPEN of bar 1 (102), not at the close of bar 0 (101)38 gross_buy = 1000 / (1 + fee)39 units = gross_buy / (102 * (1 + slip))40 # decision on bar 2 -> sell at the OPEN of bar 3 (108)41 gross_sell = units * 108 * (1 - slip)42 cash_end = gross_sell * (1 - fee)43 np.testing.assert_allclose(res.equity.to_numpy(), [1000, units * 103, units * 107, cash_end, cash_end])44 np.testing.assert_allclose(res.exposure.to_numpy(), [0, 1, 1, 0, 0], atol=1e-12)4546 t = res.trades47 assert list(t["side"]) == ["buy", "sell"]48 assert list(t["ts"]) == [idx[1], idx[3]] and list(t["decided_at"]) == [idx[0], idx[2]]49 assert t["price"].tolist() == pytest.approx([102 * 1.002, 108 * 0.998])50 assert t["notional"].tolist() == pytest.approx([gross_buy, gross_sell])51 assert t["fee"].tolist() == pytest.approx([gross_buy * fee, gross_sell * fee])52 assert t["slippage"].tolist() == pytest.approx([units * 102 * slip, units * 108 * slip])53 assert t["realised_pnl"].iloc[1] == pytest.approx(cash_end - 1000) # whole round trip, all costs included54 # money is conserved: final cash = initial + price moves - fees - slippage55 assert cash_end == pytest.approx(1000 + units * (108 - 102) - t["fee"].sum() - t["slippage"].sum())56 assert res.open_positions == [] and seen == list(idx)575859def test_zero_cost_fill_is_exactly_the_open():60 data = {"X": frame([100, 50, 50], [100, 50, 80])}61 res = run_backtest(strat(lambda d, p: {"X": 1.0}), data, BacktestConfig("1h", fee_bps=0, slippage_bps=0))62 assert res.trades["price"].iloc[0] == 50 and res.equity.tolist() == pytest.approx([1000, 1000, 1600])636465def test_sells_fund_buys_and_cash_never_negative():66 a = frame([100, 100, 100, 100], [100, 100, 100, 100])67 data = {"A": a, "B": a.copy()}68 idx = a.index6970 def rotate(d, p):71 return {"A": 1.0, "B": 0.0} if d["A"].index[-1] < idx[1] else {"A": 0.0, "B": 1.0}7273 res = run_backtest(strat(rotate, universe=("A", "B")), data, BacktestConfig("1h", fee_bps=10, slippage_bps=10))74 t = res.trades75 assert list(zip(t["ts"], t["symbol"], t["side"])) == [(idx[1], "A", "buy"), (idx[2], "A", "sell"),76 (idx[2], "B", "buy")]77 assert (res.cash >= 0).all() and res.cash.iloc[-1] == pytest.approx(0, abs=1e-9)78 cost = 0.998 * 0.998 # (1 - fee)(1 - slippage) approximately, three legs79 assert res.equity.iloc[-1] == pytest.approx(1000 / (1.001 * 1.001) * (0.999 * 0.999) / (1.001 * 1.001))80 assert res.equity.iloc[-1] < 1000 * cost818283def test_rebalance_threshold_and_min_order():84 data = make_data(400, ("A", "B"), seed=3)85 half = strat(lambda d, p: {"A": 0.5, "B": 0.5}, universe=("A", "B"))86 always = run_backtest(half, data, BacktestConfig("1h", rebalance_threshold=0.0))87 banded = run_backtest(half, data, BacktestConfig("1h", rebalance_threshold=0.05))88 assert len(always.trades) > 300 # rebalances every bar89 assert 2 <= len(banded.trades) < 30 # initial purchase, then only when a weight drifts by 5 points90 assert (banded.trades["notional"] / 1000 > 0.03).all()91 sized = run_backtest(half, data, BacktestConfig("1h", rebalance_threshold=0.0, min_order=20.0))92 assert len(sized.trades) < len(always.trades) and (sized.trades["notional"] >= 20.0 / 1.002).all()93 blocked = run_backtest(half, data, BacktestConfig("1h", min_order=10_000.0))94 assert len(blocked.trades) == 0 and (blocked.equity == 1000).all()959697def test_full_exit_is_not_blocked_by_the_threshold():98 data = {"X": frame([100] * 6, [100] * 6)}99 idx = data["X"].index100 f = lambda d, p: {"X": 0.04 if d["X"].index[-1] < idx[2] else 0.0} # noqa: E731101 res = run_backtest(strat(f), data, BacktestConfig("1h", rebalance_threshold=0.03))102 assert list(res.trades["side"]) == ["buy", "sell"] and res.open_positions == []103104105def test_history_before_start_is_lookback_only():106 data = make_data(300, ("X",), seed=5)107 idx = data["X"].index108 lengths = []109110 def f(d, p):111 lengths.append((d["X"].index[0], d["X"].index[-1], len(d["X"])))112 return {"X": 1.0}113114 res = run_backtest(strat(f, lookback=50), data, BacktestConfig("1h"), start=idx[200], end=idx[260])115 assert res.equity.index[0] == idx[200] and res.equity.index[-1] == idx[259] and res.equity.iloc[0] == 1000116 assert lengths[0] == (idx[151], idx[200], 50) # 50 closed bars ending at the decision bar117 assert all(n == 50 for _, _, n in lengths)118 assert res.trades["ts"].iloc[0] == idx[201]119 # without history the engine waits until `lookback` bars exist120 lengths.clear()121 res = run_backtest(strat(f, lookback=50), data, BacktestConfig("1h"), end=idx[100])122 assert lengths[0] == (idx[0], idx[49], 50) and res.trades["ts"].iloc[0] == idx[50]123 assert (res.equity.iloc[:50] == 1000).all()124125126def test_decision_every_is_anchored_on_utc():127 data = make_data(24 * 6, ("X",), seed=6, start="2024-01-01T05:00")128 calls = []129130 def f(d, p):131 calls.append(d["X"].index[-1])132 return {"X": 1.0}133134 res = run_backtest(strat(f), data, BacktestConfig("1h", decision_every=24))135 assert all(ts.hour == 23 for ts in calls) and len(calls) == 6 # bars closing at 00:00 UTC136 assert res.trades["ts"].iloc[0].hour == 0137138139def test_strategy_cannot_corrupt_engine_data():140 data = make_data(200, ("X",), seed=7)141 before = data["X"].copy()142143 def vandal(d, p):144 d["X"].loc[:, "close"] = 0.0145 d["X"].iloc[:, 0] = -1.0146 return {"X": 1.0}147148 a = run_backtest(strat(vandal, lookback=20), data, BacktestConfig("1h"))149 b = run_backtest(strat(lambda d, p: {"X": 1.0}, lookback=20), data, BacktestConfig("1h"))150 pd.testing.assert_frame_equal(data["X"], before)151 pd.testing.assert_series_equal(a.equity, b.equity)152153154def test_invalid_weights_and_bad_data_stop_the_run():155 data = make_data(50, ("X",), seed=8)156 for bad in ({"X": 1.2}, {"X": -0.1}, {"X": float("nan")}, {"Y": 0.5}, [1.0], {"X": "1"}):157 with pytest.raises(WeightsError):158 run_backtest(strat(lambda d, p, bad=bad: bad), data, BacktestConfig("1h"))159 two = make_data(50, ("A", "B"), seed=8)160 with pytest.raises(WeightsError, match="sum"):161 run_backtest(strat(lambda d, p: {"A": 0.6, "B": 0.6}, universe=("A", "B")), two, BacktestConfig("1h"))162 two["B"] = two["B"].iloc[1:]163 with pytest.raises(DataError, match="index differs"):164 run_backtest(strat(lambda d, p: {"A": 0.5, "B": 0.5}, universe=("A", "B")), two, BacktestConfig("1h"))165 with pytest.raises(ValueError):166 BacktestConfig("1h", fee_bps=-1)167168169def test_missing_symbol_means_zero_weight_and_partial_allocation_keeps_cash():170 data = make_data(50, ("A", "B"), seed=9)171 res = run_backtest(strat(lambda d, p: {"A": 0.3}, universe=("A", "B")), data,172 BacktestConfig("1h", fee_bps=0, slippage_bps=0))173 assert set(res.trades["symbol"]) == {"A"}174 assert res.exposure.iloc[1:].between(0.25, 0.35).all() and res.cash.iloc[-1] > 600175