momentum devandClaude Fable 5
ab2810896c
iv_exit_config replays ANY target/breakeven/trail combo through the shared exit engine vs the live-exit baseline — reusable way to adjudicate exit numbers. Counterfactual over setup fires: v1 (slow, 1.37R median winners) gains most from prior_low trailing (win 17->24%, +0.036R); pm1 (fast, 2.64R median, 10R tail) gains most from NO target + step0.5 (+0.273R, win 24->34%) — the fixed target throws away pm1's fat tail. Registered both as tracked experiments. +test. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
555 lines
26 KiB
Python
555 lines
26 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}
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 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) -> 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 == "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
|
|
|
|
|
|
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
|
|
for fire in new_fires:
|
|
row = _compute_fire(con, exp, fire)
|
|
if row is None:
|
|
continue
|
|
_insert_run(con, exp["id"], fire, row)
|
|
inserted += 1
|
|
return inserted
|