269 lines
11 KiB
Python
269 lines
11 KiB
Python
"""Backtest replay page (plan 015): replay_window assembly + /api/backtest
|
|
route + cache. Fixtures modeled on test_chart_endpoint.py / test_regime.py's
|
|
_seed_session idiom (direct DashboardState/handler-level calls — the urllib
|
|
HTTP-fixture group in this directory hits a pre-existing local-sandbox
|
|
failure unrelated to this feature)."""
|
|
|
|
import io
|
|
import os
|
|
import uuid
|
|
from datetime import date, datetime, timezone
|
|
|
|
import psycopg2
|
|
import pytest
|
|
|
|
from api.main import DashboardHandler, DashboardState
|
|
from research import regime
|
|
from research.experiments.replay import replay_window
|
|
from storage.event_store import EventStore
|
|
from strategy.population import PopulationConfig
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _isolated_pg_schema(monkeypatch):
|
|
"""Pin every EventStore in this file to one throwaway schema (mirrors
|
|
test_chart_endpoint.py) so seeded setups/bars never touch prod."""
|
|
schema = f"mem_{uuid.uuid4().hex[:12]}"
|
|
monkeypatch.setenv("MOMENTUM_PG_SCHEMA", schema)
|
|
yield
|
|
url = os.environ.get("DATABASE_URL")
|
|
if url:
|
|
conn = psycopg2.connect(url)
|
|
conn.autocommit = True
|
|
conn.cursor().execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')
|
|
conn.close()
|
|
|
|
|
|
class CaptureHandler(DashboardHandler):
|
|
"""Drives do_GET without a real socket (same pattern as test_chart_endpoint.py)."""
|
|
|
|
def __init__(self, state, path):
|
|
self.state = state
|
|
self.path = path
|
|
self.rfile = io.BytesIO(b"")
|
|
self.headers = {}
|
|
self.captured = []
|
|
|
|
def _send_json(self, payload, status=200):
|
|
self.captured.append(("json", status, payload))
|
|
|
|
|
|
def _get(state, path):
|
|
h = CaptureHandler(state, path)
|
|
h.do_GET()
|
|
return h.captured[-1]
|
|
|
|
|
|
SESS1 = date(2026, 6, 10)
|
|
SESS2 = date(2026, 6, 11)
|
|
# a clean runner: rises steadily so simulate_exit resolves inside the session
|
|
# (same bar path idiom test_replay_run.py uses for its setup-sourced fire)
|
|
RUNNER_BARS = [(10 + i * 0.1, 10.2 + i * 0.1, 9.99 + i * 0.1, 10.1 + i * 0.1) for i in range(10)]
|
|
|
|
|
|
def _seed_setup(store, symbol, sess, entry=10.0, stop=9.0):
|
|
store.con.execute(
|
|
"INSERT INTO setup_events (setup_id, symbol, session_date, setup_time, "
|
|
"setup_name, setup_version, entry_reference_price, invalidation_price) "
|
|
"VALUES (?,?,?,?,?,?,?,?)",
|
|
[f"s-{symbol}-{sess}", symbol, sess,
|
|
datetime(sess.year, sess.month, sess.day, 13, 35), "orb", "v1", entry, stop])
|
|
|
|
|
|
def _seed_minutes(store, symbol, sess, rows, start_min=36):
|
|
for i, (o, h, low, c) in enumerate(rows):
|
|
ts = datetime(sess.year, sess.month, sess.day, 13, start_min + i)
|
|
store.con.execute(
|
|
"INSERT INTO minute_bars (symbol, timestamp, session_date, is_premarket, "
|
|
"is_regular_hours, open, high, low, close, volume, vwap, quality_score) "
|
|
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
|
|
[symbol, ts, sess, False, True, o, h, low, c, 1000, c, 1.0])
|
|
|
|
|
|
def _seed_store():
|
|
"""2 sessions x 2 setups (AAA, BBB), each with a runner bar path."""
|
|
store = EventStore("momentum")
|
|
for sess in (SESS1, SESS2):
|
|
for sym in ("AAA", "BBB"):
|
|
_seed_setup(store, sym, sess)
|
|
_seed_minutes(store, sym, sess, RUNNER_BARS)
|
|
return store
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Test plan case 1: replay_window over 2 sessions x 2 setups -> 4 trades
|
|
# --------------------------------------------------------------------------
|
|
def test_replay_window_two_sessions_two_setups():
|
|
store = _seed_store()
|
|
try:
|
|
out = replay_window(store, str(SESS1), str(SESS2))
|
|
finally:
|
|
store.close()
|
|
|
|
assert len(out["trades"]) == 4
|
|
for t in out["trades"]:
|
|
assert t["session_date"] in (str(SESS1), str(SESS2))
|
|
# exit_ts is inside the session it was simulated on
|
|
assert t["exit_ts"].startswith(t["session_date"])
|
|
assert out["skipped"] == 0
|
|
assert out["screened"] == {"population": 0, "regime": 0}
|
|
|
|
assert len(out["daily"]) == 2
|
|
d0, d1 = out["daily"]
|
|
assert d0["date"] == str(SESS1) and d1["date"] == str(SESS2)
|
|
assert d0["n"] == 2 and d1["n"] == 2
|
|
# cumulative is consistent with per-day sums
|
|
assert abs(d0["cum_r"] - d0["sum_r"]) < 1e-9
|
|
assert abs(d1["cum_r"] - (d0["sum_r"] + d1["sum_r"])) < 1e-9
|
|
assert out["summary"]["n"] == 4
|
|
assert abs(out["summary"]["sum_r"] - d1["cum_r"]) < 1e-9
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Test plan case 2: population screen excludes an out-of-band setup
|
|
# --------------------------------------------------------------------------
|
|
def test_replay_window_population_screen_excludes_out_of_band():
|
|
store = _seed_store()
|
|
try:
|
|
# BBB priced outside the default [2,20] band via daily_bars.open
|
|
for sess in (date(2026, 6, 9), SESS1, SESS2):
|
|
store.con.execute(
|
|
"INSERT INTO daily_bars (symbol, trade_date, open, close, volume) "
|
|
"VALUES (?,?,?,?,?)", ["BBB", sess, 25.0, 25.0, 500_000])
|
|
out = replay_window(store, str(SESS1), str(SESS2), population_cfg=PopulationConfig())
|
|
finally:
|
|
store.close()
|
|
|
|
symbols = {t["symbol"] for t in out["trades"]}
|
|
assert "BBB" not in symbols # rejected: price outside $2-20 band
|
|
assert "AAA" in symbols # no daily_bars row -> fail-open, kept
|
|
assert out["screened"]["population"] == 2 # BBB on both SESS1 and SESS2
|
|
assert len(out["trades"]) == 2
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Test plan case 3: regime gate drops an unfavorable session, keeps unscored
|
|
# --------------------------------------------------------------------------
|
|
def test_replay_window_regime_gate_drops_unfavorable_session():
|
|
store = _seed_store()
|
|
try:
|
|
regime.apply_schema(store.con)
|
|
store.con.execute(
|
|
"INSERT INTO regime_scores (model_version, session_date, p_profit, favorable, computed_at) "
|
|
"VALUES (?,?,?,?,?)", ["meta_v1", SESS1, 0.1, False, datetime.now(timezone.utc)])
|
|
# SESS2 has no regime_scores row -> fail-open, kept
|
|
out = replay_window(store, str(SESS1), str(SESS2), regime_gate=True)
|
|
finally:
|
|
store.close()
|
|
|
|
dates = {t["session_date"] for t in out["trades"]}
|
|
assert str(SESS1) not in dates
|
|
assert str(SESS2) in dates
|
|
assert out["screened"]["regime"] == 2 # AAA + BBB on SESS1
|
|
assert len(out["trades"]) == 2
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Test plan case 4: /api/backtest?days=9999 -> clamped
|
|
# --------------------------------------------------------------------------
|
|
def test_api_backtest_clamped_for_huge_days(monkeypatch):
|
|
monkeypatch.setenv("BACKTEST_MAX_SESSIONS", "45") # cap is env-tunable (default 300)
|
|
state = DashboardState("momentum")
|
|
kind, status, payload = _get(state, "/api/backtest?days=9999")
|
|
assert kind == "json" and status == 200
|
|
assert payload["clamped"] is True
|
|
assert payload["sessions_requested"] == 45
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Test plan case 5: cache — two identical calls compute the replay once
|
|
# --------------------------------------------------------------------------
|
|
def test_api_backtest_cache_hits_second_call(monkeypatch):
|
|
_seed_store().close()
|
|
|
|
calls = {"n": 0}
|
|
import research.experiments.replay as replay_mod
|
|
real = replay_mod.replay_window
|
|
|
|
def counting(*a, **kw):
|
|
calls["n"] += 1
|
|
return real(*a, **kw)
|
|
|
|
monkeypatch.setattr(replay_mod, "replay_window", counting)
|
|
|
|
state = DashboardState("momentum")
|
|
kind1, status1, payload1 = _get(state, "/api/backtest?days=10")
|
|
kind2, status2, payload2 = _get(state, "/api/backtest?days=10")
|
|
|
|
assert status1 == 200 and status2 == 200
|
|
assert calls["n"] == 1
|
|
assert payload1["trades"] and payload1["trades"] == payload2["trades"]
|
|
assert payload1["settings"] == payload2["settings"]
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Route wiring: GET /backtest serves the page
|
|
# --------------------------------------------------------------------------
|
|
def test_backtest_route_serves_html():
|
|
kind, status, body = None, None, None
|
|
|
|
class _H(DashboardHandler):
|
|
def __init__(self, state, path):
|
|
self.state = state
|
|
self.path = path
|
|
self.rfile = io.BytesIO(b"")
|
|
self.headers = {}
|
|
self.sent = []
|
|
|
|
def _send_html(self, path):
|
|
self.sent.append(("html", path))
|
|
|
|
h = _H(DashboardState("momentum"), "/backtest")
|
|
h.do_GET()
|
|
assert h.sent and h.sent[0][0] == "html"
|
|
assert h.sent[0][1].name == "backtest.html"
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover
|
|
pytest.main([__file__, "-q"])
|
|
|
|
|
|
def test_replay_window_replays_live_track_only():
|
|
"""/backtest must judge the strategy that actually trades: setup_version v1.
|
|
Shadow research tracks (liq1/vr1/run1/pm1 -- refuted, never traded live)
|
|
share setup_events and would otherwise dominate the replay 8:1 with
|
|
tight-stop artifacts (the 2026-07-17 -1.8R/trade garbage-scale bug)."""
|
|
store = _seed_store()
|
|
# a liq1 shadow fire on the same session/bars -- must NOT be replayed
|
|
store.con.execute(
|
|
"INSERT INTO setup_events (setup_id, symbol, session_date, setup_time, "
|
|
"setup_name, setup_version, entry_reference_price, invalidation_price) "
|
|
"VALUES (?,?,?,?,?,?,?,?)",
|
|
[f"liq-AAA-{SESS1}", "AAA", SESS1,
|
|
datetime(SESS1.year, SESS1.month, SESS1.day, 13, 35),
|
|
"liquid_breakout", "liq1", 10.0, 9.0])
|
|
from research.experiments.replay import replay_window
|
|
out = replay_window(store, str(SESS1), str(SESS2))
|
|
assert len(out["trades"]) == 4 # the 4 v1 fires only; liq1 excluded
|
|
|
|
|
|
def test_backtest_regime_override_bypasses_gate(monkeypatch):
|
|
"""/backtest UX fix (owner: 'no data in backtest'): when the regime gate
|
|
zeroes a whole window the page must be able to show RAW-criteria trades
|
|
via ?regime=0, and the response must echo which mode ran."""
|
|
import api.main as m
|
|
store = _seed_store()
|
|
monkeypatch.setenv("TRADING_REGIME_GATE_ENABLED", "1")
|
|
# unfavorable scores for BOTH sessions -> gated run has zero trades
|
|
from research import regime as rg
|
|
rg.apply_schema(store.con)
|
|
for sess in (SESS1, SESS2):
|
|
store.con.execute(
|
|
"INSERT INTO regime_scores (model_version, session_date, p_profit, favorable, computed_at) "
|
|
"VALUES ('meta_v1', ?, 0.1, false, now()) ON CONFLICT DO NOTHING", [sess])
|
|
state = m.DashboardState("momentum")
|
|
gated = state.backtest(days=10)
|
|
raw = state.backtest(days=10, regime_override=False)
|
|
assert gated["summary"]["n"] == 0 and gated["settings"]["regime_gate"] is True
|
|
assert raw["summary"]["n"] == 4 and raw["settings"]["regime_gate"] is False
|