momentum devandClaude Fable 5
8ac222a4ca
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>
738 lines
34 KiB
Python
738 lines
34 KiB
Python
"""Replay engine — counterfactual interventions over logged fires.
|
|
|
|
For each running experiment: reconstruct the logged fires from the event
|
|
stream (order_submitted[buy] -> order_filled -> position_closed, the
|
|
_rebuild_daily_state idiom), population-filter them through
|
|
``strategy/population.py`` with the experiment's population JSON, load that
|
|
symbol/day's minute bars, and compute:
|
|
|
|
* baseline_r — what actually happened (position_closed, in R units: the live
|
|
truth, not a simulation).
|
|
* variant_r — the intervention applied counterfactually on the bar path.
|
|
|
|
FIDELITY (hard-coded, NOT per-experiment-optional): every counterfactual entry
|
|
AND stop fills +``adverse_ticks`` WORSE (buy higher, sell lower); the
|
|
asym_exit stop width is floored at ``max(atr_frac*ATR14, spread_mult*
|
|
spread_proxy)`` where spread_proxy is the median 1-min high-low ("quiet bar")
|
|
of the session. This is the guard against the zero-slippage tight-stop artifact
|
|
(vr1/liq1 class) — see docs/research/entry_conversion_research.md.
|
|
|
|
ALL variant R is measured in the ORIGINAL 1R unit (entry_actual - stop_actual)
|
|
so variant_r - baseline_r is a comparable per-trade R delta.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from collections import defaultdict
|
|
from dataclasses import dataclass, fields
|
|
from datetime import date, datetime, timezone
|
|
|
|
from research.experiments import registry
|
|
from research.query import query_minute_bars
|
|
from storage.projections import _r_multiple
|
|
from strategy.exits import TRAIL_PRIOR_LOW, ExitConfig, simulate_exit
|
|
from strategy.population import PopulationConfig, population_reject
|
|
|
|
# ponytail: penny tick — correct for the whole $2-20 population; sub-penny NA
|
|
# here. Upgrade to a per-price tick table only if the universe ever goes sub-$1.
|
|
TICK = 0.01
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Fidelity helpers
|
|
# --------------------------------------------------------------------------- #
|
|
def spread_proxy(bars) -> float:
|
|
"""Median 1-min high-low across the session — the research doc's quiet-bar
|
|
spread stand-in (median downweights the loud breakout bars)."""
|
|
if bars is None or not len(bars):
|
|
return 0.0
|
|
hl = (bars["high"].astype(float) - bars["low"].astype(float))
|
|
return float(hl.median()) if len(hl) else 0.0
|
|
|
|
|
|
def _adverse(fidelity: dict) -> float:
|
|
return float(fidelity.get("adverse_ticks", 1)) * TICK
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Fire reconstruction
|
|
# --------------------------------------------------------------------------- #
|
|
@dataclass
|
|
class Fire:
|
|
fire_id: str # order_id
|
|
symbol: str
|
|
session_date: date
|
|
fill_ts: object # entry-fill timestamp (str or datetime)
|
|
entry_price: float | None
|
|
stop_price: float | None
|
|
exit_price: float | None
|
|
baseline_r: float | None
|
|
trigger_price: float | None
|
|
sim_baseline: bool = False # setup-sourced fire: baseline is the LIVE exit simulated on the bar path
|
|
|
|
|
|
def _payload(e: dict) -> dict:
|
|
raw = e.get("payload") or e.get("payload_json")
|
|
if isinstance(raw, str):
|
|
try:
|
|
return json.loads(raw)
|
|
except ValueError:
|
|
return {}
|
|
return raw or {}
|
|
|
|
|
|
def _f(x):
|
|
try:
|
|
return float(x)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _to_date(ts) -> date:
|
|
if isinstance(ts, datetime):
|
|
return ts.date()
|
|
return date.fromisoformat(str(ts)[:10])
|
|
|
|
|
|
def reconstruct_fires(store, since: str | None = None) -> list[Fire]:
|
|
"""Rebuild every logged fire (a filled BUY order) with its baseline outcome.
|
|
|
|
baseline_r is present only when a position_closed for the same symbol/day is
|
|
found (the R baseline). ``coverage`` = fires-with-baseline / total.
|
|
"""
|
|
submits: dict[str, dict] = {}
|
|
for e in store.query_events(event_type="order_submitted", since=since, limit=None):
|
|
p = _payload(e)
|
|
oid = str(p.get("order_id") or "")
|
|
if oid and str(p.get("side", "buy")).lower() == "buy":
|
|
submits[oid] = {"symbol": p.get("symbol"), "trigger_price": p.get("price"),
|
|
"ts": e["timestamp"], "session_date": _to_date(e["timestamp"])}
|
|
|
|
# order_filled gives a PRECISE entry timestamp/price when present — but the live
|
|
# loop fills resting-limit brackets and detects them via position-snapshot
|
|
# diffing, so it emits NO order_filled for real fills (verified: 0 all-time in
|
|
# prod). The submit timestamp is then the entry-time proxy (an ORB bracket fills
|
|
# within a bar of the submit), and position_closed carries the real entry/stop/
|
|
# exit prices. Without this fallback the whole harness reconstructs zero fires.
|
|
fills: dict[str, dict] = {}
|
|
for e in store.query_events(event_type="order_filled", since=since, limit=None):
|
|
p = _payload(e)
|
|
oid = str(p.get("order_id") or "")
|
|
if oid in submits:
|
|
fills[oid] = {"fill_price": p.get("fill_price"), "ts": e["timestamp"]}
|
|
|
|
closes: dict[tuple, list] = defaultdict(list)
|
|
for e in store.query_events(event_type="position_closed", since=since, limit=None):
|
|
p = _payload(e)
|
|
closes[(p.get("symbol"), _to_date(e["timestamp"]))].append(p)
|
|
|
|
out: list[Fire] = []
|
|
for oid, sub in submits.items():
|
|
sym = sub["symbol"]
|
|
f = fills.get(oid)
|
|
# entry timing: precise order_filled if present, else the submit time
|
|
if f:
|
|
fill_ts, sd, entry = f["ts"], _to_date(f["ts"]), _f(f["fill_price"])
|
|
else:
|
|
fill_ts, sd, entry = sub["ts"], sub["session_date"], None
|
|
cl = (closes.get((sym, sd)) or [None])[0]
|
|
# need SOMETHING that places entry on the bar path AND gives a stop for R:
|
|
# a fill (precise) or a close (carries the real entry/stop/exit). Neither -> skip.
|
|
if f is None and cl is None:
|
|
continue
|
|
stop = exit_p = baseline = None
|
|
if cl:
|
|
stop = _f(cl.get("stop_loss_price"))
|
|
exit_p = _f(cl.get("exit_price"))
|
|
e_entry = _f(cl.get("entry_price")) or entry
|
|
entry = entry if entry is not None else e_entry # close's entry when no fill event
|
|
if stop is not None and e_entry is not None and exit_p is not None:
|
|
baseline = _r_multiple(e_entry, stop, exit_p, cl.get("side", "buy"))
|
|
out.append(Fire(fire_id=oid, symbol=sym, session_date=sd, fill_ts=fill_ts,
|
|
entry_price=entry, stop_price=stop, exit_price=exit_p,
|
|
baseline_r=baseline, trigger_price=_f(sub["trigger_price"])))
|
|
return out
|
|
|
|
|
|
def reconstruct_fires_from_setups(store, since: str | None = None,
|
|
limit: int | None = None) -> list[Fire]:
|
|
"""Simulated fires from the labeler's setup_events — the 18k-strong backtestable
|
|
ORB universe, vs the handful of LIVE round-trips. Each setup carries the ORB
|
|
entry (entry_reference_price) and stop (invalidation_price); the exit
|
|
experiments replay over its minute-bar path exactly as for a live fire. The
|
|
baseline here is the LIVE exit config SIMULATED on the same path (sim_baseline),
|
|
so variant_r - baseline_r stays a clean same-path delta with no live P&L needed.
|
|
This is what grows every exit experiment's n from ~15 to thousands."""
|
|
q = ("SELECT setup_id, symbol, session_date, setup_time, entry_reference_price, "
|
|
"invalidation_price FROM setup_events "
|
|
"WHERE entry_reference_price IS NOT NULL AND invalidation_price IS NOT NULL "
|
|
"AND entry_reference_price > invalidation_price")
|
|
args: list = []
|
|
if since:
|
|
q += " AND session_date >= ?"
|
|
args.append(since)
|
|
q += " ORDER BY session_date, setup_time"
|
|
if limit:
|
|
q += f" LIMIT {int(limit)}"
|
|
out: list[Fire] = []
|
|
for sid, sym, sd, st, entry, stop in store.con.execute(q, args).fetchall():
|
|
out.append(Fire(fire_id=f"setup:{sid}", symbol=sym, session_date=_to_date(sd),
|
|
fill_ts=st, entry_price=_f(entry), stop_price=_f(stop),
|
|
exit_price=None, baseline_r=None, trigger_price=_f(entry),
|
|
sim_baseline=True))
|
|
return out
|
|
|
|
|
|
def coverage(fires: list[Fire]) -> tuple[int, int]:
|
|
"""(reconstructable-with-baseline, total) — the STOP-condition coverage number."""
|
|
return sum(1 for f in fires if f.baseline_r is not None), len(fires)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Population
|
|
# --------------------------------------------------------------------------- #
|
|
def population_config(pop_json: dict) -> PopulationConfig:
|
|
names = {fld.name for fld in fields(PopulationConfig)}
|
|
return PopulationConfig(**{k: v for k, v in (pop_json or {}).items() if k in names})
|
|
|
|
|
|
def _fire_in_population(con, cfg: PopulationConfig, fire: Fire) -> bool:
|
|
"""Screen a fire through the SHARED population gate on open-knowable inputs
|
|
(static open gap + open price + rvol), the eod_replay idiom. Fail-open: no
|
|
daily context => keep (the fire already passed the live gate)."""
|
|
rows = con.execute(
|
|
"SELECT open, close, volume FROM daily_bars WHERE symbol=? AND trade_date<=? "
|
|
"ORDER BY trade_date DESC LIMIT 21", [fire.symbol, fire.session_date]).fetchall()
|
|
if len(rows) < 2 or rows[0][0] is None:
|
|
return True
|
|
t_open = _f(rows[0][0]); t_vol = _f(rows[0][2]); prev_close = _f(rows[1][1])
|
|
gap = (t_open / prev_close - 1) if (t_open and prev_close) else 0.0
|
|
prior = [_f(r[2]) or 0.0 for r in rows[1:21]]
|
|
adv = sum(prior) / len(prior) if prior else 0.0
|
|
rvol = (t_vol / adv) if (t_vol and adv) else 0.0
|
|
return population_reject(cfg, gap_pct=gap, price=t_open, volume=None, rvol=rvol) is None
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Bar helpers
|
|
# --------------------------------------------------------------------------- #
|
|
def _entry_index(bars, fill_ts) -> int:
|
|
"""Index of the entry bar (last bar at/before the fill); 0 if none precede."""
|
|
import pandas as pd
|
|
ts = pd.Timestamp(fill_ts)
|
|
prior = bars.index[bars["timestamp"] <= ts]
|
|
return int(prior[-1]) if len(prior) else 0
|
|
|
|
|
|
def _atr14(con, symbol: str, session_date: date) -> float | None:
|
|
row = con.execute(
|
|
"SELECT atr_14 FROM daily_bars WHERE symbol=? AND trade_date<=? AND atr_14 IS NOT NULL "
|
|
"ORDER BY trade_date DESC LIMIT 1", [symbol, session_date]).fetchone()
|
|
return _f(row[0]) if row else None
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Interventions — small pure functions. Each returns (variant_r, detail).
|
|
# variant_r in ORIGINAL R units (risk = entry_actual - stop_actual). None =
|
|
# not computable.
|
|
# --------------------------------------------------------------------------- #
|
|
def iv_time_stop(fire, bars_after, trigger_high, params, fidelity):
|
|
"""Scratch at market (next bar open, +adverse) if within N min there is NO
|
|
new post-entry high AND price never reached +0.5R; else identical to
|
|
baseline (the intervention did nothing)."""
|
|
risk = fire.entry_price - fire.stop_price
|
|
n = int(params.get("minutes", 10))
|
|
window = bars_after.iloc[:n]
|
|
high_water = fire.entry_price
|
|
confirmed = False
|
|
for i in range(len(window)):
|
|
hi = float(window.iloc[i]["high"])
|
|
high_water = max(high_water, hi)
|
|
if hi > trigger_high or high_water >= fire.entry_price + 0.5 * risk:
|
|
confirmed = True
|
|
break
|
|
if confirmed:
|
|
return fire.baseline_r, {"kind": "time_stop", "confirmed": True}
|
|
if n < len(bars_after):
|
|
raw_exit = float(bars_after.iloc[n]["open"])
|
|
elif len(bars_after):
|
|
raw_exit = float(bars_after.iloc[-1]["close"])
|
|
else:
|
|
return fire.baseline_r, {"kind": "time_stop", "confirmed": False, "reason": "no_bars"}
|
|
entry_fill = fire.entry_price + _adverse(fidelity)
|
|
exit_fill = raw_exit - _adverse(fidelity)
|
|
return (exit_fill - entry_fill) / risk, {
|
|
"kind": "time_stop", "confirmed": False, "raw_exit": raw_exit}
|
|
|
|
|
|
def iv_probe_add(fire, bars_after, trigger_high, params, fidelity):
|
|
"""0.5x at the fill; add 0.5x at the first new high above the trigger-bar
|
|
high within N min (fill at that bar's high +adverse); blended R, both legs
|
|
exit at the baseline exit (-adverse)."""
|
|
if fire.exit_price is None:
|
|
return None, {"kind": "probe_add", "reason": "no_exit"}
|
|
risk = fire.entry_price - fire.stop_price
|
|
probe = float(params.get("probe_frac", 0.5))
|
|
n = int(params.get("confirm_minutes", 10))
|
|
adv = _adverse(fidelity)
|
|
entry_fill = fire.entry_price + adv
|
|
exit_fill = fire.exit_price - adv
|
|
r = probe * (exit_fill - entry_fill) / risk
|
|
add_fill = None
|
|
for i in range(min(n, len(bars_after))):
|
|
hi = float(bars_after.iloc[i]["high"])
|
|
if hi > trigger_high:
|
|
add_fill = hi + adv
|
|
break
|
|
if add_fill is not None:
|
|
r += (1.0 - probe) * (exit_fill - add_fill) / risk
|
|
return r, {"kind": "probe_add", "added": add_fill is not None, "add_fill": add_fill}
|
|
|
|
|
|
_LIVE_CFG = None
|
|
|
|
|
|
def _live_exit_cfg():
|
|
"""The LIVE exit config (ExitConfig.from_env) — the baseline a setup-sourced
|
|
fire is measured against. Cached; read once per process."""
|
|
global _LIVE_CFG
|
|
if _LIVE_CFG is None:
|
|
_LIVE_CFG = ExitConfig.from_env()
|
|
return _LIVE_CFG
|
|
|
|
|
|
def _sim_baseline_r(fire, bars_after, fidelity) -> float | None:
|
|
"""Baseline R for a setup fire: the LIVE exit config simulated on the bar path,
|
|
with the SAME adverse-fill accounting the interventions use (fair delta)."""
|
|
risk = fire.entry_price - fire.stop_price
|
|
if risk <= 0 or not len(bars_after):
|
|
return None
|
|
adv = _adverse(fidelity)
|
|
entry_fill = fire.entry_price + adv
|
|
res = simulate_exit(entry_fill, fire.stop_price, bars_after, _live_exit_cfg())
|
|
return sum(fl.frac * ((fl.price - adv) - entry_fill) / risk for fl in res.fills)
|
|
|
|
|
|
def iv_exit_config(fire, bars_after, params, fidelity):
|
|
"""Generic exit-config experiment: run the shared exit engine (simulate_exit)
|
|
with an arbitrary ExitConfig from params['exit_cfg'] and measure R vs the LIVE
|
|
exit baseline, same adverse-fill accounting. Lets ANY target/breakeven/trail
|
|
combination be adjudicated over logged fires without new code — the reusable
|
|
way to answer 'does this exit number retain more profit?'."""
|
|
from strategy.exits import TRAIL_NONE
|
|
risk = fire.entry_price - fire.stop_price
|
|
if risk <= 0 or not len(bars_after):
|
|
return None, {"kind": "exit_config", "reason": "no_bars_or_risk"}
|
|
ec = params.get("exit_cfg", {})
|
|
cfg = ExitConfig(
|
|
target_r=float(ec.get("target_r", 2.0)),
|
|
breakeven_at_r=float(ec.get("breakeven_at_r", 0.0)),
|
|
breakeven_at_pct=float(ec.get("breakeven_at_pct", 0.05)),
|
|
trail_mode=ec.get("trail_mode", TRAIL_NONE),
|
|
trail_pct=float(ec.get("trail_pct", 0.0)),
|
|
trail_after_r=float(ec.get("trail_after_r", 1.0)),
|
|
trail_r_step=float(ec.get("trail_r_step", 0.0)),
|
|
profit_lock_tiers=[], scale_out_r=0.0, first_red_exit=False,
|
|
catastrophe_pct=0.0, enforce_stop_grace_passes=0)
|
|
adv = _adverse(fidelity)
|
|
entry_fill = fire.entry_price + adv
|
|
res = simulate_exit(entry_fill, fire.stop_price, bars_after, cfg)
|
|
v = sum(fl.frac * ((fl.price - adv) - entry_fill) / risk for fl in res.fills)
|
|
return v, {"kind": "exit_config", "cfg": ec, "reason": res.reason}
|
|
|
|
|
|
def iv_volume_fade(fire, bars_after, params, fidelity):
|
|
"""VOLUME-FADE exit: exit at market to LOCK a gain when the momentum volume
|
|
dies, instead of scratching at breakeven.
|
|
|
|
Baseline volume = mean of the first ``baseline_bars`` post-entry bars (the
|
|
entry burst). After a ``min_hold_bars`` warm-up, once the trade is IN PROFIT
|
|
(high-water above entry), the first bar whose volume drops below
|
|
``fade_frac`` x baseline exits at that bar's close (-adverse) — capturing the
|
|
live gain. The original stop still caps the downside (bar low <= stop -> stop
|
|
fill, -adverse). No fade and no stop -> ride to EOD (last close -adverse).
|
|
R is in ORIGINAL units (entry_actual - stop_actual), adverse fills on entry
|
|
AND exit, so it is comparable to baseline and free of the zero-slippage
|
|
artifact."""
|
|
if "volume" not in getattr(bars_after, "columns", []):
|
|
return None, {"kind": "volume_fade", "reason": "no_volume_column"}
|
|
risk = fire.entry_price - fire.stop_price
|
|
if risk <= 0 or not len(bars_after):
|
|
return None, {"kind": "volume_fade", "reason": "no_bars_or_risk"}
|
|
fade_frac = float(params.get("fade_frac", 0.4))
|
|
base_n = int(params.get("baseline_bars", 5))
|
|
min_hold = int(params.get("min_hold_bars", 3))
|
|
require_profit = bool(params.get("require_profit", True))
|
|
adv = _adverse(fidelity)
|
|
entry_fill = fire.entry_price + adv
|
|
stop = fire.stop_price
|
|
|
|
base_vol = float(bars_after.iloc[:base_n]["volume"].mean() or 0.0)
|
|
high_water = fire.entry_price
|
|
for i in range(len(bars_after)):
|
|
b = bars_after.iloc[i]
|
|
lo, hi, close, vol = float(b["low"]), float(b["high"]), float(b["close"]), float(b["volume"] or 0.0)
|
|
high_water = max(high_water, hi)
|
|
if lo <= stop: # downside: original stop fills first
|
|
return (stop - adv - entry_fill) / risk, {
|
|
"kind": "volume_fade", "exit": "stop", "bar": i}
|
|
if i < min_hold or base_vol <= 0:
|
|
continue
|
|
in_profit = (not require_profit) or (high_water > fire.entry_price)
|
|
if in_profit and vol < fade_frac * base_vol: # momentum volume died -> lock the gain
|
|
return (close - adv - entry_fill) / risk, {
|
|
"kind": "volume_fade", "exit": "fade", "bar": i,
|
|
"vol": vol, "base_vol": base_vol, "frac": vol / base_vol}
|
|
eod = float(bars_after.iloc[-1]["close"]) # never faded, never stopped -> EOD
|
|
return (eod - adv - entry_fill) / risk, {"kind": "volume_fade", "exit": "eod"}
|
|
|
|
|
|
def iv_asym_exit(con, fire, bars, bars_after, params, fidelity):
|
|
"""Replace the live exit with stop = max(atr_frac*ATR14, spread_mult*
|
|
spread_proxy), NO target, trail under prior lows to EOD. Simulated through
|
|
strategy.exits.simulate_exit ONLY; R recomputed in original units with
|
|
adverse entry AND sell fills."""
|
|
risk = fire.entry_price - fire.stop_price
|
|
atr14 = _atr14(con, fire.symbol, fire.session_date)
|
|
spread = spread_proxy(bars)
|
|
atr_frac = float(params.get("atr_frac", fidelity.get("atr_frac", 0.25)))
|
|
spread_mult = float(fidelity.get("spread_mult", 3))
|
|
floor_w = max(atr_frac * (atr14 or 0.0), spread_mult * (spread or 0.0))
|
|
if floor_w <= 0:
|
|
return None, {"kind": "asym_exit", "reason": "no_floor"}
|
|
adv = _adverse(fidelity)
|
|
entry_fill = fire.entry_price + adv
|
|
init_stop = entry_fill - floor_w
|
|
cfg = ExitConfig(target_r=0.0, breakeven_at_r=0.0, breakeven_at_pct=0.0,
|
|
trail_mode=TRAIL_PRIOR_LOW, trail_after_r=1.0, trail_r_step=0.0,
|
|
scale_out_r=0.0, first_red_exit=False, profit_lock_tiers=[],
|
|
catastrophe_pct=0.0, enforce_stop_grace_passes=0,
|
|
default_trail_r_pct=0.0)
|
|
res = simulate_exit(entry_fill, init_stop, bars_after, cfg)
|
|
variant_r = 0.0
|
|
for fl in res.fills:
|
|
variant_r += fl.frac * ((fl.price - adv) - entry_fill) / risk
|
|
return variant_r, {"kind": "asym_exit", "exit_sim": "strategy.exits.simulate_exit",
|
|
"reason": res.reason, "floor_w": floor_w,
|
|
"atr14": atr14, "spread_proxy": spread}
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Regime gate (session-level; plan 011) — keep-or-zero a fire based on a
|
|
# PRE-COMPUTED session-level regime score (research.regime.score), the same
|
|
# shape as rank_one_bullet below but keyed on session_date alone (no
|
|
# multi-fire coordination needed, so it plugs into the per-fire _compute_fire
|
|
# path instead of a _run_rank-style batch).
|
|
# --------------------------------------------------------------------------- #
|
|
def iv_regime_gate(fire: "Fire", baseline: float | None, favorable: dict,
|
|
model_version: str | None) -> tuple[float | None, dict]:
|
|
"""variant_r = baseline if the session's regime score says favorable, else
|
|
0.0. A session with no score row is KEPT (fail-open, like the population
|
|
gate — an unscored session must not silently zero out a fire), flagged
|
|
``no_score`` in detail."""
|
|
fav = favorable.get(fire.session_date)
|
|
if fav is None:
|
|
return baseline, {"kind": "regime_gate", "favorable": None,
|
|
"model_version": model_version, "no_score": True}
|
|
if fav:
|
|
return baseline, {"kind": "regime_gate", "favorable": True, "model_version": model_version}
|
|
return 0.0, {"kind": "regime_gate", "favorable": False, "model_version": model_version}
|
|
|
|
|
|
def _load_regime_favorable(con, model_version: str) -> dict:
|
|
"""{session_date: favorable} for one regime model_version — loaded once
|
|
per ``run()`` call for a regime_gate experiment (not per fire)."""
|
|
from research.regime import apply_schema as _regime_apply_schema
|
|
_regime_apply_schema(con)
|
|
rows = con.execute(
|
|
"SELECT session_date, favorable FROM regime_scores WHERE model_version=?",
|
|
[model_version]).fetchall()
|
|
return {_to_date(r[0]): bool(r[1]) for r in rows}
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Rank (portfolio-level; receipts-only, starves without plan 005)
|
|
# --------------------------------------------------------------------------- #
|
|
def ranks_from_receipts(store, since: str | None = None) -> dict:
|
|
"""{(symbol, session_date): rank} from entry_gate_evaluated receipts. NO
|
|
historical event carries rank (armed book is memory-only), so this is empty
|
|
for fires predating plan 005 — rank_one_bullet starves until receipts flow."""
|
|
out: dict = {}
|
|
for e in store.query_events(event_type="entry_gate_evaluated", since=since, limit=None):
|
|
p = _payload(e)
|
|
rank = p.get("rank")
|
|
if rank is None:
|
|
rank = (p.get("gates", {}).get("rank", {}) or {}).get("value")
|
|
if rank is None or not p.get("symbol"):
|
|
continue
|
|
out[(p.get("symbol"), _to_date(e["timestamp"]))] = int(rank)
|
|
return out
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Run
|
|
# --------------------------------------------------------------------------- #
|
|
def _insert_run(con, exp_id: str, fire: Fire, row: dict) -> None:
|
|
con.execute(
|
|
"INSERT INTO experiment_runs "
|
|
"(experiment_id, fire_id, symbol, session_date, baseline_r, variant_r, detail, computed_at) "
|
|
"VALUES (?, ?, ?, ?, ?, ?, ?::jsonb, ?) ON CONFLICT DO NOTHING",
|
|
[exp_id, fire.fire_id, fire.symbol, fire.session_date,
|
|
row["baseline_r"], row["variant_r"], json.dumps(row["detail"]),
|
|
datetime.now(timezone.utc)])
|
|
|
|
|
|
def _compute_fire(con, exp: dict, fire: Fire, favorable: dict | None = None) -> dict | None:
|
|
kind = (exp["intervention"] or {}).get("kind")
|
|
bars = query_minute_bars(con, fire.symbol, fire.session_date)
|
|
if bars is None or not len(bars):
|
|
return None
|
|
if fire.entry_price is None or fire.stop_price is None:
|
|
return {"baseline_r": fire.baseline_r, "variant_r": None,
|
|
"detail": {"kind": kind, "reason": "no_entry_or_stop"}}
|
|
risk = fire.entry_price - fire.stop_price
|
|
if risk <= 0:
|
|
return {"baseline_r": fire.baseline_r, "variant_r": None,
|
|
"detail": {"kind": kind, "reason": "no_risk"}}
|
|
entry_idx = _entry_index(bars, fire.fill_ts)
|
|
bars_after = bars.iloc[entry_idx + 1:].reset_index(drop=True)
|
|
trigger_high = float(bars.iloc[entry_idx]["high"])
|
|
params, fidelity = exp["intervention"], exp["fidelity"]
|
|
if kind == "regime_gate":
|
|
# needs the baseline BEFORE computing variant (unlike the other kinds,
|
|
# which compute variant first and attach fire.sim_baseline's simulated
|
|
# baseline at the very end) — a setup-sim fire's baseline IS the
|
|
# simulated live-exit R, same as every other setup-sourced fire.
|
|
baseline = _sim_baseline_r(fire, bars_after, fidelity) if fire.sim_baseline else fire.baseline_r
|
|
v, d = iv_regime_gate(fire, baseline, favorable or {}, params.get("model_version"))
|
|
if fire.sim_baseline:
|
|
d["source"] = "setup_sim"
|
|
return {"baseline_r": baseline, "variant_r": v, "detail": d}
|
|
if kind == "time_stop":
|
|
v, d = iv_time_stop(fire, bars_after, trigger_high, params, fidelity)
|
|
elif kind == "probe_add":
|
|
v, d = iv_probe_add(fire, bars_after, trigger_high, params, fidelity)
|
|
elif kind == "asym_exit":
|
|
v, d = iv_asym_exit(con, fire, bars, bars_after, params, fidelity)
|
|
elif kind == "volume_fade":
|
|
v, d = iv_volume_fade(fire, bars_after, params, fidelity)
|
|
elif kind == "exit_config":
|
|
v, d = iv_exit_config(fire, bars_after, params, fidelity)
|
|
else:
|
|
return None
|
|
d["fidelity"] = fidelity
|
|
# setup-sourced fires have no live P&L — baseline is the LIVE exit simulated on
|
|
# the same path, so variant vs baseline is a clean same-path delta.
|
|
if fire.sim_baseline:
|
|
baseline = _sim_baseline_r(fire, bars_after, fidelity)
|
|
d["source"] = "setup_sim"
|
|
return {"baseline_r": baseline, "variant_r": v, "detail": d}
|
|
return {"baseline_r": fire.baseline_r, "variant_r": v, "detail": d}
|
|
|
|
|
|
def _run_rank(con, exp: dict, new_fires: list[Fire], ranks: dict) -> int:
|
|
inserted = 0
|
|
by_session: dict = defaultdict(list)
|
|
for fr in new_fires:
|
|
by_session[fr.session_date].append(fr)
|
|
for _sess, fs in by_session.items():
|
|
ranked = [(ranks.get((fr.symbol, fr.session_date)), fr) for fr in fs]
|
|
have = [(r, fr) for r, fr in ranked if r is not None]
|
|
top_id = min(have, key=lambda rf: rf[0])[1].fire_id if have else None
|
|
for r, fr in ranked:
|
|
if r is None:
|
|
row = {"baseline_r": fr.baseline_r, "variant_r": None,
|
|
"detail": {"kind": "rank_one_bullet", "rank_missing": True}}
|
|
elif fr.fire_id == top_id:
|
|
row = {"baseline_r": fr.baseline_r, "variant_r": fr.baseline_r,
|
|
"detail": {"kind": "rank_one_bullet", "kept": True, "rank": r}}
|
|
else:
|
|
row = {"baseline_r": fr.baseline_r, "variant_r": 0.0,
|
|
"detail": {"kind": "rank_one_bullet", "kept": False, "rank": r}}
|
|
_insert_run(con, exp["id"], fr, row)
|
|
inserted += 1
|
|
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
|
|
the count of new run rows."""
|
|
con = store.con
|
|
registry.apply_schema(con)
|
|
live_fires = reconstruct_fires(store, since=since)
|
|
# Setup-sourced simulated fires (the labeler's 18k ORB setups) grow the exit
|
|
# experiments' n from the handful of live round-trips to thousands. Gated so it
|
|
# can be turned off; capped by EXPERIMENTS_SETUP_SIM_LIMIT for a chunked backfill.
|
|
setup_fires: list[Fire] = []
|
|
if os.environ.get("EXPERIMENTS_INCLUDE_SETUP_SIM", "1").strip().lower() in {"1", "true", "yes"}:
|
|
lim = os.environ.get("EXPERIMENTS_SETUP_SIM_LIMIT")
|
|
setup_fires = reconstruct_fires_from_setups(store, since=since,
|
|
limit=int(lim) if lim else None)
|
|
ranks = ranks_from_receipts(store, since=since)
|
|
inserted = 0
|
|
for exp in registry.list_experiments(con, running_only=True):
|
|
kind = (exp["intervention"] or {}).get("kind")
|
|
done = {r[0] for r in con.execute(
|
|
"SELECT fire_id FROM experiment_runs WHERE experiment_id=?", [exp["id"]]).fetchall()}
|
|
pop = population_config(exp["population"])
|
|
# rank_one_bullet needs live receipts, so setup fires (no rank) don't apply;
|
|
# every exit experiment replays over live + setup fires.
|
|
src = live_fires if kind == "rank_one_bullet" else (live_fires + setup_fires)
|
|
new_fires = [f for f in src
|
|
if f.fire_id not in done and _fire_in_population(con, pop, f)]
|
|
if not new_fires:
|
|
continue
|
|
if kind == "rank_one_bullet":
|
|
inserted += _run_rank(con, exp, new_fires, ranks)
|
|
continue
|
|
favorable = None
|
|
if kind == "regime_gate":
|
|
favorable = _load_regime_favorable(con, (exp["intervention"] or {}).get("model_version"))
|
|
for fire in new_fires:
|
|
row = _compute_fire(con, exp, fire, favorable=favorable)
|
|
if row is None:
|
|
continue
|
|
_insert_run(con, exp["id"], fire, row)
|
|
inserted += 1
|
|
return inserted
|