diff --git a/experiments/regime_meta_v1.json b/experiments/regime_meta_v1.json new file mode 100644 index 0000000..60f4a7f --- /dev/null +++ b/experiments/regime_meta_v1.json @@ -0,0 +1,8 @@ +{ + "id": "regime_meta_v1", + "hypothesis": "A shallow walk-forward GradientBoostingClassifier metalabeling P(our ORB makes money THIS session) from the same free pre-open features beats the threshold_v1 rule net, out-of-sample, and zeroing the ORB on unfavorable sessions raises whole-population expR.", + "population": {"min_entry_gap_pct": 0.10, "min_day_volume": 300000, "price_min": 2.0, "price_max": 20.0}, + "intervention": {"kind": "regime_gate", "model_version": "meta_v1"}, + "bars": {"expR_delta_min": 0.05, "min_n": 300}, + "fidelity": {"atr_frac": 0.25, "spread_mult": 3, "adverse_ticks": 1} +} diff --git a/experiments/regime_threshold_v1.json b/experiments/regime_threshold_v1.json new file mode 100644 index 0000000..2955484 --- /dev/null +++ b/experiments/regime_threshold_v1.json @@ -0,0 +1,8 @@ +{ + "id": "regime_threshold_v1", + "hypothesis": "A trivial walk-forward threshold on free pre-open features (trailing bot expR + tape regime + SPY vol side) separates runner-regime sessions from chop sessions and zeroing the ORB on unfavorable sessions raises whole-population expR.", + "population": {"min_entry_gap_pct": 0.10, "min_day_volume": 300000, "price_min": 2.0, "price_max": 20.0}, + "intervention": {"kind": "regime_gate", "model_version": "threshold_v1"}, + "bars": {"expR_delta_min": 0.05, "min_n": 300}, + "fidelity": {"atr_frac": 0.25, "spread_mult": 3, "adverse_ticks": 1} +} diff --git a/research/experiments/replay.py b/research/experiments/replay.py index adc33a3..b00e88f 100644 --- a/research/experiments/replay.py +++ b/research/experiments/replay.py @@ -419,6 +419,39 @@ def iv_asym_exit(con, fire, bars, bars_after, params, fidelity): "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) # --------------------------------------------------------------------------- # @@ -451,7 +484,7 @@ def _insert_run(con, exp_id: str, fire: Fire, row: dict) -> None: datetime.now(timezone.utc)]) -def _compute_fire(con, exp: dict, fire: Fire) -> dict | None: +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): @@ -467,6 +500,16 @@ def _compute_fire(con, exp: dict, fire: Fire) -> dict | None: 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": @@ -545,8 +588,11 @@ def run(store, since: str | None = None) -> int: 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) + row = _compute_fire(con, exp, fire, favorable=favorable) if row is None: continue _insert_run(con, exp["id"], fire, row) diff --git a/research/regime.py b/research/regime.py new file mode 100644 index 0000000..ac99f78 --- /dev/null +++ b/research/regime.py @@ -0,0 +1,493 @@ +"""research/regime.py — Plan 011: a FREE-data daily signal that decides +WHETHER the ORB trades at all today (a session-level metalabel), orthogonal to +WHICH NAME to trade (strategy/population.py + rank_one_bullet). Predicts +``P(our ORB makes money THIS session)`` from pre-open features; the label is +our OWN per-session realized expR (Chan metalabeling) — the mean of +``research.experiments.replay._sim_baseline_r`` (the LIVE exit config +simulated on the bar path) over that session's setup fires. + + python -m research.regime probe # Step 0 STOP gate + python -m research.regime build-features # Step 1: regime_features + python -m research.regime score --model threshold_v1 # Step 1: walk-forward rule + python -m research.regime score --model meta_v1 # Step 2: walk-forward GBM + +DISCIPLINE (see docs/research/regime_notes.md / plan 011): + * Every score is WALK-FORWARD: favorable(D) uses ONLY rows with + session_date < D (1-session embargo). Nothing here may read a full-sample + statistic — see ``score_threshold``/``score_meta`` and + ``test_regime_score_is_walk_forward``, the leakage guard. + * ``label_expr``/``label_pos`` are OUTCOME columns; FEATURE_COLS never + includes them (``test_regime_features_no_label_leak``). + * ``full_sample_fit_threshold`` (Step 4) is DELIBERATELY LEAKY — a sanity + check only, never the shipped flag. + +Imports nothing from the live loop (trading_execution.py / run_live_paper.py / +strategy/exits.py's live call sites / strategy/population.py are never +imported here). Reuses research.experiments.replay's fire reconstruction + +_sim_baseline_r as BOTH the fire source and the label source. +""" + +from __future__ import annotations + +import argparse +import os +from bisect import bisect_left +from collections import defaultdict +from datetime import date, datetime, timezone +from pathlib import Path +from statistics import mean, median, pstdev + +_SCHEMA_SQL = Path(__file__).resolve().parent.parent / "storage" / "regime_schema.sql" + +# Pre-open / prior-close-lagged features only. See module docstring + plan 011 +# Step 1 for the source of each (breadth_* and tape_regime come from the +# session's OWN setup_events rows — the same "as_of=open" contract +# strategy/population.py relies on; spy_* are lagged to the PRIOR close; +# bot_trailing_expr_10d is a rolling mean of PRIOR sessions' labels). +FEATURE_COLS = ( + "breadth_gap", "breadth_rvol", "tape_regime", + "spy_mom5", "spy_mom20", "spy_vol20", "bot_trailing_expr_10d", +) +LABEL_COLS = ("label_expr", "label_pos") + +TAPE_ORDINAL = {"trend_down": -1.0, "chop": 0.0, "trend_up": 1.0} + +MIN_LABELABLE_SESSIONS = 150 # Step 0 STOP threshold (thin sample) +SPY_LOOKBACK_DAYS = 20 +TRAILING_WINDOW = 10 # bot_trailing_expr_10d +MIN_THRESHOLD_PRIOR = 5 # threshold_v1: below this, can't infer a sign -> fail open +MIN_TRAIN_SESSIONS = 60 # meta_v1: below this, don't gate on an unfit model -> fail open +P_MIN_DEFAULT = 0.5 + +# Fidelity floors used to simulate the session label (matches the harness's +# default adverse-fill accounting for setup-sourced fires — see replay.py). +_LABEL_FIDELITY = {"atr_frac": 0.25, "spread_mult": 3, "adverse_ticks": 1} + + +def apply_schema(con) -> None: + """Create regime_features/regime_scores idempotently. Honors PG_SKIP_SCHEMA + exactly like research.experiments.registry.apply_schema (owner-provisioned + DDL escape hatch).""" + if os.environ.get("PG_SKIP_SCHEMA", "").strip().lower() in {"1", "true", "yes"}: + return + con.execute(_SCHEMA_SQL.read_text()) + + +def _as_date(v) -> date: + return v if isinstance(v, date) else date.fromisoformat(str(v)[:10]) + + +# --------------------------------------------------------------------------- # +# Step 0 — data-availability probe (STOP gate) +# --------------------------------------------------------------------------- # +def _labelable_sessions(con) -> list[date]: + rows = con.execute( + "SELECT DISTINCT session_date FROM setup_events " + "WHERE entry_reference_price IS NOT NULL AND invalidation_price IS NOT NULL " + "ORDER BY session_date").fetchall() + return [_as_date(r[0]) for r in rows] + + +def _spy_daily_dates(con) -> list[date]: + rows = con.execute( + "SELECT trade_date FROM daily_bars WHERE symbol='SPY' ORDER BY trade_date").fetchall() + return [_as_date(r[0]) for r in rows] + + +def probe(con) -> dict: + """Step 0 STOP gate. Pure counts + a `stop` verdict — under ~150 labelable + sessions is a thin sample (STOP). SPY coverage is reported separately + (never a STOP by itself): when it doesn't cover the setup range, build- + features falls back to the non-SPY features (spy_* stay NULL) rather than + fabricating SPY history.""" + sessions = _labelable_sessions(con) + n_labelable = len(sessions) + spy_dates = _spy_daily_dates(con) + n_with_spy20 = sum(1 for d in sessions if bisect_left(spy_dates, d) >= SPY_LOOKBACK_DAYS) + spy_span = (spy_dates[0], spy_dates[-1]) if spy_dates else (None, None) + spy_overlaps_range = bool(spy_dates) and bool(sessions) and spy_dates[0] <= sessions[-1] + return { + "labelable_sessions": n_labelable, + "spy_daily_count": len(spy_dates), + "spy_span": spy_span, + "sessions_with_spy20": n_with_spy20, + "spy_coverage_ok": spy_overlaps_range and n_with_spy20 >= 0.5 * max(n_labelable, 1), + "stop": n_labelable < MIN_LABELABLE_SESSIONS, + } + + +# --------------------------------------------------------------------------- # +# Step 1 — build-features +# --------------------------------------------------------------------------- # +class _ConShim: + """Wraps a raw connection as a store-like object (`.con`) so + replay.reconstruct_fires_from_setups can be reused verbatim without a full + EventStore.""" + + def __init__(self, con): + self.con = con + + +def _session_labels(con, sessions: list[date]) -> dict[date, float]: + """{session_date: label_expr} = mean(_sim_baseline_r) over that session's + setup fires — the SAME simulation the exit-experiment harness uses for a + setup-sourced fire's baseline. Sessions with no computable fire (no bars, + no risk) are omitted, not zero-filled.""" + from research.experiments.replay import _entry_index, _sim_baseline_r, reconstruct_fires_from_setups + from research.query import query_minute_bars + + fires = reconstruct_fires_from_setups(_ConShim(con)) + by_session: dict[date, list] = defaultdict(list) + for f in fires: + by_session[f.session_date].append(f) + + out: dict[date, float] = {} + for d in sessions: + rs = [] + for f in by_session.get(d, []): + if f.entry_price is None or f.stop_price is None or f.entry_price - f.stop_price <= 0: + continue + bars = query_minute_bars(con, f.symbol, f.session_date) + if bars is None or not len(bars): + continue + idx = _entry_index(bars, f.fill_ts) + bars_after = bars.iloc[idx + 1:].reset_index(drop=True) + r = _sim_baseline_r(f, bars_after, _LABEL_FIDELITY) + if r is not None: + rs.append(r) + if rs: + out[d] = mean(rs) + return out + + +def _session_context(con, sessions: list[date]) -> dict[date, dict]: + """{session_date: {breadth_gap, breadth_rvol, tape_regime}} from that + session's OWN labelable setup_events rows (gap_pct/relative_volume/ + tape_regime_at_trigger — the same open-knowable columns + strategy/population.py and the labeler already treat as pre-entry + context). tape_regime is the EARLIEST (by setup_time) setup's reading.""" + rows = con.execute( + "SELECT session_date, gap_pct, relative_volume, tape_regime_at_trigger, setup_time " + "FROM setup_events WHERE entry_reference_price IS NOT NULL " + "AND invalidation_price IS NOT NULL ORDER BY session_date, setup_time").fetchall() + by_session: dict[date, list] = defaultdict(list) + for sd, gap, rvol, tape, _st in rows: + by_session[_as_date(sd)].append((gap, rvol, tape)) + + out = {} + for d in sessions: + items = by_session.get(d, []) + if not items: + out[d] = {"breadth_gap": None, "breadth_rvol": None, "tape_regime": None} + continue + gaps = [g for g, _r, _t in items if g is not None] + rvols = [r for _g, r, _t in items if r is not None] + out[d] = { + "breadth_gap": (sum(1 for g in gaps if g > 0) / len(gaps)) if gaps else None, + "breadth_rvol": median(rvols) if rvols else None, + "tape_regime": TAPE_ORDINAL.get(items[0][2]), + } + return out + + +def _spy_features(con, sessions: list[date]) -> dict[date, dict]: + """{session_date: {spy_mom5, spy_mom20, spy_vol20}} strictly lagged to the + PRIOR close (all SPY daily bars used are trade_date < session_date). + Missing/insufficient SPY history -> None (never fabricated).""" + rows = con.execute( + "SELECT trade_date, close FROM daily_bars WHERE symbol='SPY' ORDER BY trade_date").fetchall() + dates = [_as_date(r[0]) for r in rows] + closes = [float(r[1]) for r in rows] + + out = {} + for d in sessions: + idx = bisect_left(dates, d) # count of SPY rows strictly before d + prior = closes[:idx] + mom5 = (prior[-1] / prior[-6] - 1) if len(prior) >= 6 else None + mom20 = (prior[-1] / prior[-21] - 1) if len(prior) >= 21 else None + vol20 = None + if len(prior) >= 21: + rets = [prior[i] / prior[i - 1] - 1 for i in range(-20, 0)] + vol20 = pstdev(rets) + out[d] = {"spy_mom5": mom5, "spy_mom20": mom20, "spy_vol20": vol20} + return out + + +def _trailing_expr(labels: dict[date, float], sessions: list[date]) -> dict[date, float | None]: + """mean(label_expr) over the PRIOR (up to TRAILING_WINDOW) labelable + sessions — strictly backward-looking: `sessions` must be sorted ascending, + and a session's own label is appended to history AFTER its trailing value + is read.""" + out: dict[date, float | None] = {} + hist: list[float] = [] + for d in sessions: + window = hist[-TRAILING_WINDOW:] + out[d] = mean(window) if window else None + if d in labels: + hist.append(labels[d]) + return out + + +def build_features(con, since: str | None = None) -> int: + """Step 1: one regime_features row per labelable session with a computable + label. Idempotent (PK session_date + ON CONFLICT DO UPDATE — features are + safe to rebuild as more setup/SPY data lands).""" + apply_schema(con) + all_sessions = _labelable_sessions(con) + labels = _session_labels(con, all_sessions) # full history: label per session + trailing = _trailing_expr(labels, all_sessions) # full history: correct rolling window + + sessions = all_sessions + if since: + cutoff = date.fromisoformat(str(since)) + sessions = [d for d in sessions if d >= cutoff] # only THESE rows get written + + ctx = _session_context(con, sessions) + spy = _spy_features(con, sessions) + + now = datetime.now(timezone.utc) + n = 0 + for d in sessions: + if d not in labels: + continue # no computable fire for this session -> unlabelable, skip + row = {**ctx[d], **spy[d], "bot_trailing_expr_10d": trailing[d], + "label_expr": labels[d], "label_pos": labels[d] > 0} + con.execute( + "INSERT INTO regime_features (session_date, breadth_gap, breadth_rvol, " + "tape_regime, spy_mom5, spy_mom20, spy_vol20, bot_trailing_expr_10d, " + "label_expr, label_pos, computed_at) VALUES (?,?,?,?,?,?,?,?,?,?,?) " + "ON CONFLICT (session_date) DO UPDATE SET " + "breadth_gap=EXCLUDED.breadth_gap, breadth_rvol=EXCLUDED.breadth_rvol, " + "tape_regime=EXCLUDED.tape_regime, spy_mom5=EXCLUDED.spy_mom5, " + "spy_mom20=EXCLUDED.spy_mom20, spy_vol20=EXCLUDED.spy_vol20, " + "bot_trailing_expr_10d=EXCLUDED.bot_trailing_expr_10d, " + "label_expr=EXCLUDED.label_expr, label_pos=EXCLUDED.label_pos, " + "computed_at=EXCLUDED.computed_at", + [d, row["breadth_gap"], row["breadth_rvol"], row["tape_regime"], + row["spy_mom5"], row["spy_mom20"], row["spy_vol20"], + row["bot_trailing_expr_10d"], row["label_expr"], row["label_pos"], now]) + n += 1 + return n + + +# --------------------------------------------------------------------------- # +# Step 1 — threshold_v1 (no ML, walk-forward) +# --------------------------------------------------------------------------- # +def _tape_favorable_set(prior_rows: list[dict]) -> set[float]: + """Which tape_regime ordinal values have positive mean label_expr over + PRIOR rows ONLY — recomputed at every D, never a fixed global set (or it + would be a full-sample fit).""" + by_tape: dict[float, list[float]] = defaultdict(list) + for r in prior_rows: + if r.get("tape_regime") is not None and r.get("label_expr") is not None: + by_tape[r["tape_regime"]].append(r["label_expr"]) + return {t for t, rs in by_tape.items() if mean(rs) > 0} + + +def _vol_side_ok(prior_rows: list[dict], today_vol) -> bool: + """Whichever SPY-vol side (> median vs <= median, computed over PRIOR rows + only) has the higher mean label_expr; True if today's spy_vol20 is on that + side. Fail-open (True) when uncomputable — do NOT hardcode low-vol=good; + let prior data pick the sign.""" + have = [r for r in prior_rows if r.get("spy_vol20") is not None and r.get("label_expr") is not None] + if today_vol is None or len(have) < 10: + return True + vols = sorted(r["spy_vol20"] for r in have) + med = vols[len(vols) // 2] + hi = [r["label_expr"] for r in have if r["spy_vol20"] > med] + lo = [r["label_expr"] for r in have if r["spy_vol20"] <= med] + if not hi or not lo: + return True + prefer_hi = mean(hi) > mean(lo) + return (today_vol > med) == prefer_hi + + +def score_threshold(rows: list[dict]) -> list[dict]: + """Walk-forward threshold_v1 rule (no fit, cannot leak by construction): + favorable(D) reads row D's OWN features plus `rows[:i]` (strictly PRIOR + sessions) only — never anything at index >= i. `rows` must be sorted by + session_date ascending. Returns [{session_date, p_profit, favorable}, ...]. + """ + out = [] + for i, row in enumerate(rows): + prior = rows[:i] + if len(prior) < MIN_THRESHOLD_PRIOR: + out.append({"session_date": row["session_date"], "p_profit": 1.0, "favorable": True}) + continue + trailing_ok = (row.get("bot_trailing_expr_10d") or 0.0) > 0 + tape_ok = row.get("tape_regime") in _tape_favorable_set(prior) + vol_ok = _vol_side_ok(prior, row.get("spy_vol20")) + favorable = bool(trailing_ok and tape_ok and vol_ok) + out.append({"session_date": row["session_date"], + "p_profit": 1.0 if favorable else 0.0, "favorable": favorable}) + return out + + +def full_sample_fit_threshold(rows: list[dict]) -> list[dict]: + """DELIBERATELY LEAKY variant of score_threshold: uses the WHOLE row set + (including D and everything after) to derive the tape/vol sign instead of + `rows[:i]`. Step 4 sanity check ONLY — never the shipped flag. If this + variant crushes the walk-forward one on held-out expR, the walk-forward + gate's edge is real; if it doesn't, something upstream is leaking.""" + out = [] + for row in rows: + trailing_ok = (row.get("bot_trailing_expr_10d") or 0.0) > 0 + tape_ok = row.get("tape_regime") in _tape_favorable_set(rows) + vol_ok = _vol_side_ok(rows, row.get("spy_vol20")) + favorable = bool(trailing_ok and tape_ok and vol_ok) + out.append({"session_date": row["session_date"], + "p_profit": 1.0 if favorable else 0.0, "favorable": favorable}) + return out + + +# --------------------------------------------------------------------------- # +# Step 2 — meta_v1 (sklearn GBM, walk-forward) +# --------------------------------------------------------------------------- # +def _build_X(rows: list[dict], feature_cols=FEATURE_COLS): + """The feature matrix passed to the classifier — ONLY feature_cols, missing + values filled 0.0. Never includes label_expr/label_pos + (test_regime_features_no_label_leak asserts the column set).""" + import pandas as pd + return pd.DataFrame( + [{c: (r.get(c) if r.get(c) is not None else 0.0) for c in feature_cols} for r in rows]) + + +def score_meta(rows: list[dict], p_min: float = P_MIN_DEFAULT, + min_train: int = MIN_TRAIN_SESSIONS) -> list[dict]: + """Walk-forward sklearn.GradientBoostingClassifier(max_depth=3, <=8 + features). Fits on `rows[:i]` only (never index >= i — the leakage guard); + below `min_train` prior labeled sessions, favorable=True (fail-open, don't + gate on an unfit model).""" + from sklearn.ensemble import GradientBoostingClassifier + + out = [] + for i, row in enumerate(rows): + prior = rows[:i] + train = [r for r in prior if r.get("label_pos") is not None] + y_train = [bool(r["label_pos"]) for r in train] + if len(train) < min_train or len(set(y_train)) < 2: + out.append({"session_date": row["session_date"], "p_profit": None, "favorable": True}) + continue + clf = GradientBoostingClassifier(max_depth=3, n_estimators=50, random_state=0) + clf.fit(_build_X(train), y_train) + p = float(clf.predict_proba(_build_X([row]))[0, list(clf.classes_).index(True)]) + out.append({"session_date": row["session_date"], "p_profit": p, "favorable": p >= p_min}) + return out + + +# --------------------------------------------------------------------------- # +# score — DB wrapper +# --------------------------------------------------------------------------- # +def _feature_rows(con) -> list[dict]: + cols = ("session_date", *FEATURE_COLS, *LABEL_COLS) + q = f"SELECT {', '.join(cols)} FROM regime_features ORDER BY session_date" + out = [] + for vals in con.execute(q).fetchall(): + row = dict(zip(cols, vals)) + row["session_date"] = _as_date(row["session_date"]) + out.append(row) + return out + + +def score(con, model_version: str, p_min: float = P_MIN_DEFAULT) -> int: + """Step 1/2 `score`: write regime_scores rows for one model_version, walk- + forward over the finished regime_features table. Idempotent (PK + (model_version, session_date) + ON CONFLICT DO UPDATE).""" + apply_schema(con) + rows = _feature_rows(con) + if model_version == "threshold_v1": + scored = score_threshold(rows) + elif model_version == "meta_v1": + scored = score_meta(rows, p_min=p_min) + else: + raise ValueError(f"unknown model_version: {model_version!r}") + + now = datetime.now(timezone.utc) + for s in scored: + con.execute( + "INSERT INTO regime_scores (model_version, session_date, p_profit, favorable, computed_at) " + "VALUES (?, ?, ?, ?, ?) ON CONFLICT (model_version, session_date) DO UPDATE SET " + "p_profit=EXCLUDED.p_profit, favorable=EXCLUDED.favorable, computed_at=EXCLUDED.computed_at", + [model_version, s["session_date"], s["p_profit"], s["favorable"], now]) + return len(scored) + + +# --------------------------------------------------------------------------- # +# Step 4 — OOS split + leakage-audit helpers (supervised run wires these up; +# never executed against production here). +# --------------------------------------------------------------------------- # +def oos_split(rows: list[dict], frac: float = 0.3) -> tuple[list[dict], list[dict]]: + """(head, held_out_tail) — `rows` sorted by session_date ascending; the + tail is the most-recent `frac` of sessions.""" + n = len(rows) + k = int(round(n * (1 - frac))) + return rows[:k], rows[k:] + + +def sequential_folds(rows: list[dict], k: int = 3) -> list[list[dict]]: + """`rows` (sorted ascending) split into k sequential chronological folds of + ~equal size, for the 'expR_delta positive across ~3 sequential folds' OOS + check.""" + n = len(rows) + size = max(1, -(-n // k)) # ceil(n/k) so k folds cover all rows + return [rows[i:i + size] for i in range(0, n, size)][:k] + + +def session_expr_delta(rows: list[dict], scored: list[dict]) -> float | None: + """Fire-free session-level expR_delta proxy: mean(label_expr over + favorable sessions) - mean(label_expr over ALL sessions in `rows`). Usable + directly on regime_features rows (no experiment_runs needed) for the + OOS/fold/leakage checks in this step.""" + fav = {s["session_date"] for s in scored if s.get("favorable")} + fav_labels = [r["label_expr"] for r in rows if r["session_date"] in fav and r.get("label_expr") is not None] + all_labels = [r["label_expr"] for r in rows if r.get("label_expr") is not None] + if not fav_labels or not all_labels: + return None + return mean(fav_labels) - mean(all_labels) + + +# --------------------------------------------------------------------------- # +# CLI +# --------------------------------------------------------------------------- # +def _store(): + from storage.event_store import EventStore + return EventStore(os.environ.get("WATCHER_EVENT_DB_PATH", "momentum")) + + +def main(argv=None) -> int: + ap = argparse.ArgumentParser(prog="python -m research.regime") + sub = ap.add_subparsers(dest="cmd", required=True) + sub.add_parser("probe") + bf = sub.add_parser("build-features") + bf.add_argument("--since", help="only rebuild sessions >= this date (YYYY-MM-DD)") + sc = sub.add_parser("score") + sc.add_argument("--model", required=True, choices=["threshold_v1", "meta_v1"]) + sc.add_argument("--p-min", type=float, default=P_MIN_DEFAULT) + args = ap.parse_args(argv) + + store = _store() + try: + if args.cmd == "probe": + info = probe(store.con) + for k, v in info.items(): + print(f"{k}: {v}") + if info["stop"]: + print(f"STOP: {info['labelable_sessions']} labelable sessions " + f"< {MIN_LABELABLE_SESSIONS} (thin sample)") + return 1 + return 0 + if args.cmd == "build-features": + n = build_features(store.con, since=args.since) + print(f"regime_features: {n} rows") + return 0 + if args.cmd == "score": + n = score(store.con, args.model, p_min=args.p_min) + print(f"regime_scores[{args.model}]: {n} rows") + return 0 + return 2 + finally: + store.close() + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/storage/regime_schema.sql b/storage/regime_schema.sql new file mode 100644 index 0000000..5cc5ef5 --- /dev/null +++ b/storage/regime_schema.sql @@ -0,0 +1,34 @@ +-- Regime gate schema (plan 011) — a FREE-data daily signal (session-level +-- metalabel) that decides whether the ORB trades AT ALL today, separate from +-- which NAME to trade (strategy/population.py). Applied by +-- research.regime.apply_schema on first run (same idiom as +-- storage/experiments_schema.sql). Idempotent: safe to run repeatedly. + +-- One row per labelable session_date: PRE-OPEN/prior-close-lagged features + +-- the session's OWN outcome label. label_* are OUTCOME columns computed from +-- the session's setup fires — NEVER read back as model features (see +-- research.regime.FEATURE_COLS / test_regime_features_no_label_leak). +CREATE TABLE IF NOT EXISTS regime_features ( + session_date DATE PRIMARY KEY, + breadth_gap DOUBLE PRECISION, -- fraction of session's setups with gap_pct>0 + breadth_rvol DOUBLE PRECISION, -- median relative_volume across session's setups + tape_regime DOUBLE PRECISION, -- ordinal: trend_down=-1, chop=0, trend_up=1 + spy_mom5 DOUBLE PRECISION, -- SPY 5-session momentum as of PRIOR close + spy_mom20 DOUBLE PRECISION, -- SPY 20-session momentum as of PRIOR close + spy_vol20 DOUBLE PRECISION, -- SPY 20-session realized vol as of PRIOR close + bot_trailing_expr_10d DOUBLE PRECISION, -- mean label_expR over the PRIOR 10 labelable sessions + label_expr DOUBLE PRECISION, -- OUTCOME: mean _sim_baseline_r over the session's setup fires + label_pos BOOLEAN, -- OUTCOME: label_expr > 0 + computed_at TIMESTAMP +); + +-- One row per (model, session): the walk-forward gate decision consumed by +-- research.experiments.replay's iv_regime_gate. +CREATE TABLE IF NOT EXISTS regime_scores ( + model_version TEXT NOT NULL, + session_date DATE NOT NULL, + p_profit DOUBLE PRECISION, + favorable BOOLEAN NOT NULL, + computed_at TIMESTAMP NOT NULL, + PRIMARY KEY (model_version, session_date) +); diff --git a/tests/unit/experiments/test_regime_gate.py b/tests/unit/experiments/test_regime_gate.py new file mode 100644 index 0000000..43a8777 --- /dev/null +++ b/tests/unit/experiments/test_regime_gate.py @@ -0,0 +1,123 @@ +"""iv_regime_gate (plan 011): keep-or-zero a fire on a pre-computed session- +level regime score, plus the registered regime_threshold_v1 experiment running +end-to-end on an isolated store. Mirrors test_replay_run.py's rank_one_bullet +tests (the same keep-or-zero shape, keyed on session_date instead of rank).""" + +import os +from datetime import date, datetime, timezone + +from research import regime +from research.experiments import adjudicate, registry, replay +from research.experiments.replay import Fire, iv_regime_gate + +REPO = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "..")) +SESS = date(2026, 6, 11) + + +def _fire(session_date=SESS, baseline=None): + return Fire(fire_id="f1", symbol="AAA", session_date=session_date, fill_ts=None, + entry_price=10.0, stop_price=9.0, exit_price=None, baseline_r=baseline, + trigger_price=10.0) + + +# --------------------------------------------------------------------------- # +# iv_regime_gate (pure) +# --------------------------------------------------------------------------- # +def test_regime_gate_keeps_and_zeros(): + fire = _fire() + # favorable session -> keeps baseline + v, d = iv_regime_gate(fire, 0.5, {SESS: True}, "threshold_v1") + assert v == 0.5 + assert d["kind"] == "regime_gate" and d["favorable"] is True + + # unfavorable session -> zeroed + v, d = iv_regime_gate(fire, 0.5, {SESS: False}, "threshold_v1") + assert v == 0.0 and d["favorable"] is False + + # no score row for the session -> KEPT (fail-open), flagged in detail + v, d = iv_regime_gate(fire, 0.5, {}, "threshold_v1") + assert v == 0.5 and d["favorable"] is None and d["no_score"] is True + + +# --------------------------------------------------------------------------- # +# expR_delta sign — proves adjudication rewards removing LOSERS, not just +# removing trades (plan 011 discipline: "let the data pick the sign"). +# --------------------------------------------------------------------------- # +def test_regime_gate_expR_delta_sign(): + dates = [date(2026, 6, 10), date(2026, 6, 11), date(2026, 6, 12)] + + # zeroed sessions are net LOSERS -> removing them raises expR_delta (positive) + baselines = [-1.0, -0.5, 2.0] + favorable = {dates[0]: False, dates[1]: False, dates[2]: True} + runs = [] + for d, b in zip(dates, baselines): + v, det = iv_regime_gate(_fire(session_date=d), b, favorable, "threshold_v1") + runs.append({"baseline_r": b, "variant_r": v, "detail": det}) + _, metrics, _ = adjudicate.evaluate(runs, {"expR_delta_min": 0.0, "min_n": 1}, None) + assert metrics["expR_delta"] > 0, metrics + + # zeroed sessions are net WINNERS -> expR_delta negative + baselines2 = [1.0, 0.5, -2.0] + favorable2 = {dates[0]: False, dates[1]: False, dates[2]: True} + runs2 = [] + for d, b in zip(dates, baselines2): + v, det = iv_regime_gate(_fire(session_date=d), b, favorable2, "threshold_v1") + runs2.append({"baseline_r": b, "variant_r": v, "detail": det}) + _, metrics2, _ = adjudicate.evaluate(runs2, {"expR_delta_min": 0.0, "min_n": 1}, None) + assert metrics2["expR_delta"] < 0, metrics2 + + +# --------------------------------------------------------------------------- # +# registered experiment, end to end on an isolated store +# --------------------------------------------------------------------------- # +def _seed_daily(store, symbol, prev_close=8.0, sess_open=10.0, vol=1_000_000): + store.con.execute( + "INSERT INTO daily_bars (symbol, trade_date, open, close, volume) VALUES (?,?,?,?,?)", + [symbol, date(2026, 6, 10), prev_close - 0.5, prev_close, vol]) + store.con.execute( + "INSERT INTO daily_bars (symbol, trade_date, open, close, volume) VALUES (?,?,?,?,?)", + [symbol, SESS, sess_open, sess_open, vol]) + + +def _seed_minutes(store, symbol, rows, start_min=30): + for i, (o, h, low, c) in enumerate(rows): + ts = datetime(2026, 6, 11, 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 test_threshold_experiment_registers_and_runs(store): + _seed_daily(store, "AAA") + _seed_minutes(store, "AAA", + [(10 + i * 0.1, 10.2 + i * 0.1, 9.99 + i * 0.1, 10.1 + i * 0.1) for i in range(10)]) + store.con.execute( + "INSERT INTO setup_events (setup_id, symbol, session_date, setup_time, setup_name, " + "setup_version, entry_reference_price, invalidation_price, gap_pct, relative_volume) " + "VALUES (?,?,?,?,?,?,?,?,?,?)", + ["s-AAA-1", "AAA", SESS, datetime(2026, 6, 11, 13, 35), "orb", "v1", 10.0, 9.0, 0.12, 3.0]) + + exp_id = registry.register(store.con, os.path.join(REPO, "experiments", "regime_threshold_v1.json")) + assert exp_id == "regime_threshold_v1" + + regime.apply_schema(store.con) + store.con.execute( + "INSERT INTO regime_scores (model_version, session_date, p_profit, favorable, computed_at) " + "VALUES (?,?,?,?,?)", ["threshold_v1", SESS, 1.0, True, datetime.now(timezone.utc)]) + + n = replay.run(store) + assert n >= 1 + row = store.con.execute( + "SELECT baseline_r, variant_r, detail FROM experiment_runs " + "WHERE experiment_id='regime_threshold_v1' AND fire_id='setup:s-AAA-1'").fetchone() + assert row is not None + baseline_r, variant_r, detail = row + assert baseline_r is not None and variant_r == baseline_r # favorable -> kept + assert detail["favorable"] is True and detail["source"] == "setup_sim" + + out = adjudicate.adjudicate(store) + verdicts = {o["experiment_id"]: o["verdict"] for o in out} + assert "regime_threshold_v1" in verdicts + assert verdicts["regime_threshold_v1"] in {"PASS", "FAIL", "INCONCLUSIVE", "KILLED"} diff --git a/tests/unit/research/test_regime.py b/tests/unit/research/test_regime.py new file mode 100644 index 0000000..f5b8463 --- /dev/null +++ b/tests/unit/research/test_regime.py @@ -0,0 +1,168 @@ +"""research/regime.py — Step 0 probe, Step 1/2 walk-forward scorers, the +leakage guard, and an end-to-end build-features + score check. Every test runs +against an ISOLATED throwaway Postgres schema (EventStore ":memory:"), never +the production DB (mirrors tests/unit/experiments/conftest.py).""" + +from __future__ import annotations + +import copy +import random +from datetime import date, datetime, timedelta + +import pytest + +from research import regime +from storage.event_store import EventStore + + +@pytest.fixture +def store(): + s = EventStore(":memory:") + regime.apply_schema(s.con) + yield s + s.close() + + +# --------------------------------------------------------------------------- # +# Step 0 — probe +# --------------------------------------------------------------------------- # +def _insert_setup(store, symbol, sess_date, i): + 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}-{i}", symbol, sess_date, + datetime(sess_date.year, sess_date.month, sess_date.day, 9, 35), + "orb", "v1", 10.0, 9.0]) + + +def test_probe_stop_on_thin_sample(store): + for i in range(5): + _insert_setup(store, "AAA", date(2026, 1, 1) + timedelta(days=i), i) + info = regime.probe(store.con) + assert info["labelable_sessions"] == 5 + assert info["stop"] is True + + +def test_probe_no_stop_with_enough_sessions(store): + for i in range(160): + _insert_setup(store, "AAA", date(2026, 1, 1) + timedelta(days=i), i) + info = regime.probe(store.con) + assert info["labelable_sessions"] == 160 + assert info["stop"] is False + + +# --------------------------------------------------------------------------- # +# Synthetic feature/label series for the pure scorers (no DB needed) +# --------------------------------------------------------------------------- # +def _synthetic_rows(n=30, seed=0) -> list[dict]: + rnd = random.Random(seed) + d0 = date(2026, 1, 1) + rows = [] + for i in range(n): + tape = rnd.choice(list(regime.TAPE_ORDINAL.values())) + label_expr = rnd.uniform(-2, 2) + rows.append({ + "session_date": d0 + timedelta(days=i), + "breadth_gap": rnd.random(), + "breadth_rvol": rnd.uniform(0.5, 5), + "tape_regime": tape, + "spy_mom5": rnd.uniform(-0.05, 0.05), + "spy_mom20": rnd.uniform(-0.1, 0.1), + "spy_vol20": rnd.uniform(0.005, 0.03), + "bot_trailing_expr_10d": rnd.uniform(-0.5, 0.5), + "label_expr": label_expr, + "label_pos": label_expr > 0, + }) + return rows + + +# --------------------------------------------------------------------------- # +# THE leakage guard: shuffling labels of sessions >= D must not change +# favorable(D). Real assert, not a smoke test. +# --------------------------------------------------------------------------- # +def test_regime_score_is_walk_forward_threshold(): + rows = _synthetic_rows(30) + split = 15 + fav_before = [s["favorable"] for s in regime.score_threshold(rows)[:split]] + + mutated = copy.deepcopy(rows) + for r in mutated[split:]: + r["label_expr"] = -(r["label_expr"] or 0.0) + r["label_pos"] = not r["label_pos"] + r["tape_regime"] = -1.0 if r["tape_regime"] != -1.0 else 1.0 + + fav_after = [s["favorable"] for s in regime.score_threshold(mutated)[:split]] + assert fav_before == fav_after + + +def test_regime_score_is_walk_forward_meta(): + rows = _synthetic_rows(90) # >= MIN_TRAIN_SESSIONS so the actual fit path runs + split = 70 + p_before = [s["p_profit"] for s in regime.score_meta(rows)[:split]] + + mutated = copy.deepcopy(rows) + for r in mutated[split:]: + r["label_expr"] = -(r["label_expr"] or 0.0) + r["label_pos"] = not r["label_pos"] + + p_after = [s["p_profit"] for s in regime.score_meta(mutated)[:split]] + assert p_before == p_after + + +# --------------------------------------------------------------------------- # +# Feature/label separation +# --------------------------------------------------------------------------- # +def test_regime_features_no_label_leak(): + rows = _synthetic_rows(5) + X = regime._build_X(rows) + assert set(X.columns) == set(regime.FEATURE_COLS) + assert "label_expr" not in X.columns + assert "label_pos" not in X.columns + + +# --------------------------------------------------------------------------- # +# build-features + score, end to end on an isolated store +# --------------------------------------------------------------------------- # +def _seed_session(store, symbol, sess): + store.con.execute( + "INSERT INTO daily_bars (symbol, trade_date, open, close, volume) VALUES (?,?,?,?,?)", + [symbol, sess, 10.0, 10.0, 1_000_000]) + store.con.execute( + "INSERT INTO setup_events (setup_id, symbol, session_date, setup_time, setup_name, " + "setup_version, entry_reference_price, invalidation_price, gap_pct, relative_volume) " + "VALUES (?,?,?,?,?,?,?,?,?,?)", + [f"s-{symbol}-{sess}", symbol, sess, + datetime(sess.year, sess.month, sess.day, 13, 35), "orb", "v1", 10.0, 9.0, 0.12, 3.0]) + for i in range(6): + o, h, low, c = 10 + i * 0.1, 10.2 + i * 0.1, 9.99 + i * 0.1, 10.1 + i * 0.1 + ts = datetime(sess.year, sess.month, sess.day, 13, 35 + 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 test_build_features_and_score_threshold_end_to_end(store): + sess1, sess2 = date(2026, 6, 10), date(2026, 6, 11) + _seed_session(store, "AAA", sess1) + _seed_session(store, "AAA", sess2) + + n = regime.build_features(store.con) + assert n == 2 + rows = regime._feature_rows(store.con) + assert len(rows) == 2 + assert rows[0]["session_date"] == sess1 and rows[1]["session_date"] == sess2 + assert rows[0]["label_expr"] is not None + # trailing_expr strictly backward: sess2's trailing == sess1's OWN label (its + # only prior session), never sess2's own label + assert rows[1]["bot_trailing_expr_10d"] == rows[0]["label_expr"] + assert rows[0]["bot_trailing_expr_10d"] is None # no prior session at all + + n2 = regime.score(store.con, "threshold_v1") + assert n2 == 2 + scores = store.con.execute( + "SELECT session_date, favorable FROM regime_scores WHERE model_version='threshold_v1' " + "ORDER BY session_date").fetchall() + assert [r[0] for r in scores] == [sess1, sess2]