research-tooling: /backtest replay dashboard — theoretical trades under current settings

Adds a read-only /backtest page showing how the bot WOULD have traded past
sessions under the CURRENTLY configured live settings: theoretical ORB fires
from setup_events, replayed through the shared exit engine (simulate_exit +
ExitConfig.from_env(), same adverse-fill accounting the exit-experiment
harness uses), screened through the shared population gate and (optionally)
the live regime gate. No new simulation math — replay.replay_window() is pure
assembly of existing pieces (reconstruct_fires_from_setups, _fire_in_population,
_load_regime_favorable, simulate_exit); api/main.py adds the /backtest and
/api/backtest routes plus a dedicated-lock 10-min cache so a slow replay never
blocks the live dashboard's SSE stream. Also extends /api/chart with an
optional date= anchor so the trade drill-in chart modal can reach sessions
older than the existing 72h-from-now window, and adds a one-line BACKTEST nav
link to the live dashboard header.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

tests/unit/api/test_backtest_endpoint.py

@@ -0,0 +1,227 @@
+"""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():
+    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"])