From 8ac222a4ca0507be2b415efd1bdaba097a793e80 Mon Sep 17 00:00:00 2001 From: momentum dev Date: Fri, 17 Jul 2026 08:42:43 -0500 Subject: [PATCH] =?UTF-8?q?research-tooling:=20/backtest=20replay=20dashbo?= =?UTF-8?q?ard=20=E2=80=94=20theoretical=20trades=20under=20current=20sett?= =?UTF-8?q?ings?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- api/main.py | 116 ++++++- api/static/backtest.html | 401 +++++++++++++++++++++++ api/static/dashboard.html | 1 + research/experiments/replay.py | 137 ++++++++ tests/unit/api/test_backtest_endpoint.py | 227 +++++++++++++ 5 files changed, 877 insertions(+), 5 deletions(-) create mode 100644 api/static/backtest.html create mode 100644 tests/unit/api/test_backtest_endpoint.py diff --git a/api/main.py b/api/main.py index d51a100..fe0579a 100644 --- a/api/main.py +++ b/api/main.py @@ -8,8 +8,11 @@ GET /api/stream Server-Sent Events: pushes snapshots on GET /api/approval-queue pending approvals only GET /api/criteria?symbol=XYZ per-criterion pass/fail breakdown GET /api/bars?symbol=XYZ[&minutes=60] recent 1-min bars for a sparkline -GET /api/chart?symbol=XYZ[&hours=24] 24h candles + fill markers + entry/stop/target levels +GET /api/chart?symbol=XYZ[&hours=24][&date=YYYY-MM-DD] candles + fill markers + levels; + date= anchors the window to that session (else "now") GET /api/decisions[?symbol=&limit=&date=] entry_gate_evaluated decision receipts (plan 014) +GET /backtest theoretical-replay dashboard HTML (plan 015) +GET /api/backtest[?days=10|since=&until=] theoretical trades under CURRENT settings (plan 015) GET /static/vendor/... vendored JS libs (e.g. Lightweight Charts) POST /api/trading/approvals/approve {"order_id": ..., "approved_by"?: ...} POST /api/trading/approvals/reject {"order_id": ..., "reason"?: ...} @@ -31,7 +34,7 @@ import os import threading import time import uuid -from datetime import datetime, timedelta, timezone +from datetime import date, datetime, timedelta, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs, urlparse @@ -249,6 +252,11 @@ class DashboardState: self._lock = threading.Lock() # lazily-built, process-wide SSE change signal (shared by all clients) self._change_signal: _ChangeSignal | None = None + # backtest replay cache (plan 015) — a DEDICATED lock, not self._lock, + # so a slow replay (10-30s) never blocks the live snapshot/SSE path. + # key = (since, until, repr(settings)) -> (result, monotonic timestamp). + self._backtest_lock = threading.Lock() + self._backtest_cache: dict = {} def change_stamp(self): """SSE change token from the shared once-per-second signal (see _ChangeSignal).""" @@ -381,13 +389,19 @@ class DashboardState: ] return {"symbol": symbol, "bars": bars} - def chart(self, symbol: str, hours: int = 24) -> dict: + def chart(self, symbol: str, hours: int = 24, anchor_date: str | None = None) -> dict: """24h candlestick bars + trade markers + entry/stop/target levels for the symbol drill-in chart (plan 013). Bars come from the research DB; markers from the fills feed; levels from the same account snapshots ``snapshots()`` already loads. Timestamps in ``minute_bars``/``events`` are naive US/Eastern — every timestamp returned here is a UTC epoch second (what Lightweight Charts wants). + + ``anchor_date`` (plan 015): anchors the window to that session's close + instead of "now" — the live modal never passes it (unchanged + behavior), but the backtest replay's drill-in needs bars from + arbitrary past sessions the plain ``hours``-back-from-now window can + never reach (it's capped at 72h). """ if self.research_con is None: return {"symbol": symbol, "bars": [], "reason": "no research db attached"} @@ -395,7 +409,11 @@ class DashboardState: et = ZoneInfo("America/New_York") hours = max(1, min(int(hours or 24), 72)) - now_et = datetime.now(et) + if anchor_date: + d = date.fromisoformat(str(anchor_date)) + now_et = datetime(d.year, d.month, d.day, 20, 0, tzinfo=et) # well after the close + else: + now_et = datetime.now(et) cutoff_et = now_et - timedelta(hours=hours) cutoff_naive = cutoff_et.replace(tzinfo=None) # naive ET, matches storage cutoff_epoch = int(cutoff_et.timestamp()) @@ -492,6 +510,84 @@ class DashboardState: row["epoch"] = _to_epoch_et(ts) if ts else None return rows + def backtest(self, days: int | None = 10, since: str | None = None, + until: str | None = None) -> dict: + """Theoretical trades under CURRENT live settings (plan 015) — + assembly only, the simulation lives in research.experiments.replay. + + ``days`` (default path): the most recent N trading sessions with + setup_events, counted backwards. ``since``/``until``: an explicit + date range instead. Either way capped at 45 sessions (``clamped`` + reports when the request exceeded it). Cached 10 min behind a + DEDICATED lock (not self._lock) so a slow replay never blocks the + live snapshot/SSE path — two concurrent identical requests compute + the replay exactly once. + """ + from strategy.exits import ExitConfig + from strategy.population import PopulationConfig + + MAX_SESSIONS = 45 + pop = PopulationConfig.from_env() + regime_on = os.environ.get("TRADING_REGIME_GATE_ENABLED", "0").strip().lower() in { + "1", "true", "yes", "on"} + fidelity = {"atr_frac": 0.25, "spread_mult": 3, "adverse_ticks": 1} + settings = { + "population": {"price_min": pop.price_min, "price_max": pop.price_max, + "min_entry_gap_pct": pop.min_entry_gap_pct, + "min_day_volume": pop.min_day_volume}, + "regime_gate": regime_on, + "fidelity": fidelity, + "target_r": ExitConfig.from_env().target_r, + } + empty = {"trades": [], "daily": [], "skipped": 0, + "summary": {"n": 0, "sum_r": 0.0, "avg_r": 0.0, "hit_rate": 0.0, "sessions": 0}, + "screened": {"population": 0, "regime": 0}} + + store = self.open_store() + try: + base_q = ("SELECT DISTINCT session_date FROM setup_events WHERE " + "entry_reference_price IS NOT NULL AND invalidation_price IS NOT NULL") + if since and until: + rows = store.con.execute( + base_q + " AND session_date >= ? AND session_date <= ? " + "ORDER BY session_date DESC", [since, until]).fetchall() + all_dates = sorted(r[0] for r in rows) + clamped = len(all_dates) > MAX_SESSIONS + dates = all_dates[-MAX_SESSIONS:] if clamped else all_dates + sessions_requested = len(all_dates) + else: + requested = int(days or 10) + clamped = requested > MAX_SESSIONS + n_sessions = min(max(requested, 1), MAX_SESSIONS) + rows = store.con.execute( + base_q + " ORDER BY session_date DESC LIMIT ?", [n_sessions]).fetchall() + dates = sorted(r[0] for r in rows) + sessions_requested = n_sessions + + if not dates: + return {**empty, "settings": settings, "clamped": clamped, + "sessions_requested": sessions_requested} + win_since, win_until = str(dates[0]), str(dates[-1]) + + key = (win_since, win_until, repr(settings)) + with self._backtest_lock: + cached = self._backtest_cache.get(key) + if cached and (time.monotonic() - cached[1]) < 600: + result = cached[0] + else: + from research.experiments.replay import replay_window + result = replay_window(store, win_since, win_until, population_cfg=pop, + regime_gate=regime_on, fidelity=fidelity) + for t in result["trades"]: + t["entry_epoch"] = _to_epoch_et(t["entry_ts"]) + t["exit_epoch"] = _to_epoch_et(t["exit_ts"]) + self._backtest_cache[key] = (result, time.monotonic()) + finally: + store.close() + + return {**result, "settings": settings, "clamped": clamped, + "sessions_requested": sessions_requested} + def _with_service(self, fn): if self.execution_service is None: return None @@ -665,6 +761,8 @@ class DashboardHandler(BaseHTTPRequestHandler): route = urlparse(self.path).path if route in ("/", "/index.html", "/dashboard"): self._send_html(STATIC_DIR / "dashboard.html") + elif route == "/backtest": + self._send_html(STATIC_DIR / "backtest.html") elif route.startswith("/static/vendor/"): # path-traversal guard: resolve then require it stays inside # STATIC_DIR/vendor (e.g. "/static/vendor/../dashboard.html" 404s). @@ -693,7 +791,8 @@ class DashboardHandler(BaseHTTPRequestHandler): q = self._query() symbol = (q.get("symbol") or [""])[0] hours = int((q.get("hours") or ["24"])[0]) - self._send_json(self.state.chart(symbol, hours)) + anchor = (q.get("date") or [""])[0] or None + self._send_json(self.state.chart(symbol, hours, anchor_date=anchor)) elif route == "/api/decisions": q = self._query() symbol = (q.get("symbol") or [""])[0] or None @@ -701,6 +800,13 @@ class DashboardHandler(BaseHTTPRequestHandler): d = (q.get("date") or [""])[0] or None self._send_json({"decisions": self.state.decisions(symbol=symbol, limit=limit, for_date=d)}) + elif route == "/api/backtest": + q = self._query() + since = (q.get("since") or [""])[0] or None + until = (q.get("until") or [""])[0] or None + days = (q.get("days") or [""])[0] + self._send_json(self.state.backtest( + days=int(days) if days else 10, since=since, until=until)) elif route == "/api/attribution": # per-component stage ratings (research/attribution.py); computed on # demand — a handful of indexed event queries, fine per click diff --git a/api/static/backtest.html b/api/static/backtest.html new file mode 100644 index 0000000..4bb87b7 --- /dev/null +++ b/api/static/backtest.html @@ -0,0 +1,401 @@ + + + + + +Momentum — Backtest Replay + + + + +
+

Backtest Replay ← live dashboard

+
+ + + + + + + +
+
+ + +
+ + + + +
+
+ +
+

Equity curve (R) click a day to filter the trade table

+
+
+ +
+

Trades

+
+ + + +
DateSymbolEntryExitReasonR
+
+
+
+ +
+
entries are setup_events-sourced ORB triggers (the labeler's population); entry fill = trigger + adverse ticks — optimistic vs real slippage on thin names.
+
+
+ +
+
+
+ + + +
+
+
blue = theoretical entry · purple = theoretical exit (reason labeled) · Charts by TradingView
+
+
+ + + + + diff --git a/api/static/dashboard.html b/api/static/dashboard.html index 7edc271..1876f8f 100644 --- a/api/static/dashboard.html +++ b/api/static/dashboard.html @@ -471,6 +471,7 @@ + BACKTEST
connecting...
diff --git a/research/experiments/replay.py b/research/experiments/replay.py index b00e88f..ba62ac0 100644 --- a/research/experiments/replay.py +++ b/research/experiments/replay.py @@ -556,6 +556,143 @@ def _run_rank(con, exp: dict, new_fires: list[Fire], ranks: dict) -> int: return inserted +# --------------------------------------------------------------------------- # +# Backtest replay (plan 015) — assembly only, no new simulation math. Screens +# + simulates the SAME pieces run() uses per-fire, just returned to a caller +# (the /api/backtest route) instead of written to experiment_runs. +# --------------------------------------------------------------------------- # +def _session_bars(con, session_date: date, symbols: list[str]) -> dict: + """All minute bars for `symbols` on one session, grouped by symbol — ONE + query per session instead of one per fire (copied from research.regime's + _session_bars idiom: the per-fire round trip was the build-features + bottleneck). Each per-symbol frame matches query_minute_bars' shape (same + cols, timestamp-sorted, fresh RangeIndex) so _entry_index works unchanged.""" + if not symbols: + return {} + ph = ",".join(["?"] * len(symbols)) + df = con.execute( + "SELECT symbol, timestamp, session_date, is_premarket, is_regular_hours, " + "open, high, low, close, volume, vwap, quality_score FROM minute_bars " + f"WHERE session_date = ? AND symbol IN ({ph}) ORDER BY symbol, timestamp", + [session_date, *symbols]).df() + if df is None or not len(df): + return {} + return {sym: g.reset_index(drop=True) for sym, g in df.groupby("symbol")} + + +def _iso(ts) -> str: + return ts.isoformat() if hasattr(ts, "isoformat") else str(ts) + + +def replay_window(store, since, until, *, population_cfg=None, + regime_gate: bool = False, fidelity: dict | None = None) -> dict: + """Theoretical trades for [since, until] under the CURRENT live exit + config. Reuses reconstruct_fires_from_setups + _fire_in_population + + simulate_exit with the harness's adverse-fill accounting (the same R + formula _sim_baseline_r uses, so numbers match the exit-experiment + harness) — assembly only, no new simulation logic.""" + fidelity = fidelity or {"atr_frac": 0.25, "spread_mult": 3, "adverse_ticks": 1} + until_d = _to_date(until) + fires = [f for f in reconstruct_fires_from_setups(store, since=str(since)) + if f.session_date <= until_d] + + screened_population = 0 + if population_cfg is not None: + kept = [] + for f in fires: + if _fire_in_population(store.con, population_cfg, f): + kept.append(f) + else: + screened_population += 1 + fires = kept + + screened_regime = 0 + if regime_gate: + favorable = _load_regime_favorable(store.con, "meta_v1") + kept = [] + for f in fires: + if favorable.get(f.session_date) is False: + screened_regime += 1 + else: + kept.append(f) + fires = kept + + by_session: dict = defaultdict(list) + for f in fires: + by_session[f.session_date].append(f) + + adv = _adverse(fidelity) + cfg = _live_exit_cfg() + trades: list[dict] = [] + skipped = 0 + daily_totals: dict = {} + for sd in sorted(by_session): + day_fires = by_session[sd] + bars_by_sym = _session_bars(store.con, sd, sorted({f.symbol for f in day_fires})) + day_r, day_n = 0.0, 0 + for f in day_fires: + bars = bars_by_sym.get(f.symbol) + risk = (f.entry_price - f.stop_price) if ( + f.entry_price is not None and f.stop_price is not None) else None + if bars is None or not len(bars) or not risk or risk <= 0: + skipped += 1 + continue + idx = _entry_index(bars, f.fill_ts) + bars_after = bars.iloc[idx + 1:].reset_index(drop=True) + if not len(bars_after): + skipped += 1 + continue + entry_fill = f.entry_price + adv + res = simulate_exit(entry_fill, f.stop_price, bars_after, cfg) + if not res.fills: + skipped += 1 + continue + r_multiple = sum(fl.frac * ((fl.price - adv) - entry_fill) / risk + for fl in res.fills) + exit_row = bars_after.iloc[res.exit_index] + trades.append({ + "symbol": f.symbol, + "session_date": str(sd), + "entry_ts": _iso(f.fill_ts), + "entry_price": f.entry_price, + "entry_fill": entry_fill, + "stop_price": f.stop_price, + "exit_ts": _iso(exit_row["timestamp"]), + "exit_reason": res.reason, + "r_multiple": r_multiple, + "fills": [{"frac": fl.frac, "price": fl.price, "reason": fl.reason} + for fl in res.fills], + }) + day_r += r_multiple + day_n += 1 + if day_n: + daily_totals[sd] = (day_n, day_r) + + daily: list[dict] = [] + cum = 0.0 + for sd in sorted(daily_totals): + n, sum_r = daily_totals[sd] + cum += sum_r + daily.append({"date": str(sd), "n": n, "sum_r": sum_r, "cum_r": cum}) + + n_total = len(trades) + sum_r_total = sum(t["r_multiple"] for t in trades) + summary = { + "n": n_total, + "sum_r": sum_r_total, + "avg_r": (sum_r_total / n_total) if n_total else 0.0, + "hit_rate": (sum(1 for t in trades if t["r_multiple"] > 0) / n_total) if n_total else 0.0, + "sessions": len(daily), + } + return { + "trades": trades, + "daily": daily, + "summary": summary, + "skipped": skipped, + "screened": {"population": screened_population, "regime": screened_regime}, + } + + def run(store, since: str | None = None) -> int: """Replay every running experiment over fires not yet in experiment_runs. Idempotent (PK (experiment_id, fire_id) + ON CONFLICT DO NOTHING). Returns diff --git a/tests/unit/api/test_backtest_endpoint.py b/tests/unit/api/test_backtest_endpoint.py new file mode 100644 index 0000000..5b4f04a --- /dev/null +++ b/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"])