From dbaeb701e2c30abea7e87090de5e926341a6d584 Mon Sep 17 00:00:00 2001 From: momentum dev Date: Tue, 14 Jul 2026 20:32:18 -0400 Subject: [PATCH] conversion(009): committed go/no-go signal probe (Step 5 instrument) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit research/conversion/{__init__,dataset,signal_probe}.py — the leakage-safe probe plan 009 Step 5 re-runs scoped to a live setup_version: - dataset.load_dataset: setup_events JOIN outcome_labels; label = reached_1r_before_minus_1r; features = trigger-time-knowable setup_events columns ONLY, dropping >70%-null and <2-distinct columns. HARD leakage rule — every outcome_labels column is blocklisted AND asserted absent from X, so a schema change can't leak the answer in. - signal_probe.run_probe: expanding-window CV over ~5 sequential session-date blocks (whole days, train strictly precedes test), HistGradientBoostingClassifier (max_iter=250, lr=0.05, max_depth=4, l2=1.0, categorical_features=indices), per-fold + mean ROC-AUC, top-decile +1R lift vs base, permutation importance on the last fold. Drops columns constant within a train slice (HistGBM binning). CLI: python3 -m research.conversion.signal_probe [--setup-version v1]. Tests: the critical leakage guard (zero outcome columns in features), AUC>0.5 on a synthetic signal + ~0.5 on pure noise, and a static live-path guard. All :memory:. No live-path change. Co-Authored-By: Claude Fable 5 --- research/conversion/__init__.py | 6 + research/conversion/dataset.py | 89 ++++++++++ research/conversion/signal_probe.py | 152 ++++++++++++++++++ tests/unit/conversion/__init__.py | 0 .../unit/conversion/test_conversion_probe.py | 112 +++++++++++++ 5 files changed, 359 insertions(+) create mode 100644 research/conversion/__init__.py create mode 100644 research/conversion/dataset.py create mode 100644 research/conversion/signal_probe.py create mode 100644 tests/unit/conversion/__init__.py create mode 100644 tests/unit/conversion/test_conversion_probe.py diff --git a/research/conversion/__init__.py b/research/conversion/__init__.py new file mode 100644 index 0000000..720d82c --- /dev/null +++ b/research/conversion/__init__.py @@ -0,0 +1,6 @@ +"""Conversion-signal probe (plans 008/009). + +Answers ONE question with a leakage-safe, date-blocked CV: do the trigger-time- +knowable features separate setups that convert (+1R before -1R) from those that +don't? A go/no-go instrument, not a live signal — nothing here gates trading. +""" diff --git a/research/conversion/dataset.py b/research/conversion/dataset.py new file mode 100644 index 0000000..83bf678 --- /dev/null +++ b/research/conversion/dataset.py @@ -0,0 +1,89 @@ +"""Assemble the leakage-safe conversion dataset. + +setup_events JOIN outcome_labels; label = reached_1r_before_minus_1r (int); +features = ONLY trigger-time-knowable setup_events columns. Drops columns that are +mostly null (>70%) or constant (<2 distinct) — both useless and the latter breaks +HistGBM's binning. + +HARD LEAKAGE RULE: no outcome_labels column may ever be a feature. The features +come from `se.*` alone and every outcome column is additionally blocklisted + +asserted, so a schema change can't silently leak the answer into the inputs. +""" + +from __future__ import annotations + +import pandas as pd + +LABEL = "reached_1r_before_minus_1r" + +# Everything in outcome_labels — derived from bars AFTER the decision point. None of +# these may be a feature. (Kept explicit so the leakage guard survives schema drift.) +OUTCOME_LABEL_COLUMNS = frozenset({ + "setup_id", "label_version", + "max_upside_next_5m", "max_upside_next_15m", "max_upside_next_60m", + "max_drawdown_next_5m", "max_drawdown_next_15m", "max_drawdown_next_60m", + "reached_1r_before_minus_1r", "reached_2r_before_minus_1r", + "held_vwap_5m", "held_vwap_15m", "trend_day_flag", "failed_breakout_flag", + "time_to_max_upside_minutes", "time_to_max_drawdown_minutes", +}) + +# setup_events identifiers / bookkeeping — knowable but not features. +ID_COLUMNS = frozenset({"setup_id", "symbol", "setup_time", "session_date"}) + +# String columns to model as categoricals (setup_version is constant once filtered +# to one version and gets dropped by the constant guard — harmless to list). +CATEGORICAL = ("setup_name", "setup_version", "tape_regime_at_trigger", + "relative_volume_source") + +NULL_FRAC_MAX = 0.70 + + +def load_dataset(con, setup_version: str | None = None, + null_frac_max: float = NULL_FRAC_MAX): + """Return (X, y, session_date, cat_cols). + + X: trigger-time-knowable setup_events features only (numeric coerced; categoricals + as pandas 'category'). y: int label. session_date: for date-blocked CV. + cat_cols: the categorical column names present in X. + """ + where, params = "", [] + if setup_version: + where, params = "WHERE se.setup_version = ?", [setup_version] + df = con.execute( + f"SELECT se.*, l.{LABEL} AS __label FROM setup_events se " + f"JOIN outcome_labels l ON l.setup_id = se.setup_id {where} " + "ORDER BY se.session_date, se.setup_time", params).df() + + df = df[df["__label"].notna()].reset_index(drop=True) + if df.empty: + raise ValueError("no labeled setups for this filter") + y = df["__label"].astype(int) + session_date = df["session_date"] + + drop = (ID_COLUMNS | OUTCOME_LABEL_COLUMNS | {"__label"}) + feat_cols = [c for c in df.columns if c not in drop] + X = df[feat_cols].copy() + + cat_cols = [c for c in CATEGORICAL if c in X.columns] + for c in X.columns: + if c not in cat_cols: + X[c] = pd.to_numeric(X[c], errors="coerce") # bool/Decimal/None -> float + + # drop mostly-null and constant columns (constant breaks HistGBM binning) + keep = [] + for c in X.columns: + if X[c].isna().mean() > null_frac_max: + continue + if X[c].nunique(dropna=True) < 2: + continue + keep.append(c) + X = X[keep] + cat_cols = [c for c in cat_cols if c in X.columns] + for c in cat_cols: + X[c] = X[c].astype("category") + + # leakage assertion — belt and suspenders on top of selecting from se.* only + leaked = set(X.columns) & OUTCOME_LABEL_COLUMNS + assert not leaked, f"LEAKAGE: outcome columns in features: {sorted(leaked)}" + + return X, y, session_date, cat_cols diff --git a/research/conversion/signal_probe.py b/research/conversion/signal_probe.py new file mode 100644 index 0000000..7be99e0 --- /dev/null +++ b/research/conversion/signal_probe.py @@ -0,0 +1,152 @@ +"""GO/NO-GO conversion-signal probe (plans 008/009). + +Leakage-safe, date-blocked CV over setup_events -> does anything trigger-time-knowable +separate +1R conversion? Expanding-window folds over ~5 sequential session-date blocks +(whole days, never shuffled: every train date precedes its test date). HistGBM per fold; +reports per-fold + mean ROC-AUC, top-decile +1R lift vs base, and permutation importance +on the last fold. + + python3 -m research.conversion.signal_probe [--setup-version v1] [--blocks N] + +Decision (plan 009): mean CV AUC >= ~0.55 with a consistent top-decile lift => signal +exists, proceed to model. Still ~0.50-0.52 fully instrumented => no ORB conversion signal; +redirect to the premarket setup. This prints the number; a human records the verdict. +""" + +from __future__ import annotations + +import argparse + +import numpy as np +from sklearn.ensemble import HistGradientBoostingClassifier +from sklearn.inspection import permutation_importance +from sklearn.metrics import roc_auc_score + +from research.conversion.dataset import load_dataset + +TOP_DECILE = 0.10 + + +def date_blocks(dates_sorted: list, n_blocks: int) -> list[list]: + """Split unique, ascending session dates into n_blocks sequential (time-ordered) + blocks of whole days.""" + n_blocks = max(2, min(n_blocks, len(dates_sorted))) + splits = np.array_split(np.array(dates_sorted, dtype=object), n_blocks) + return [list(s) for s in splits if len(s)] + + +def expanding_folds(blocks: list[list]): + """Yield (train_dates, test_dates): fold k trains on blocks[:k], tests on block k. + Leakage-safe — the test block is strictly later than every training day.""" + train: list = [] + for k in range(len(blocks) - 1): + train = train + blocks[k] + yield set(train), set(blocks[k + 1]) + + +def _cat_indices(columns: list, cat_cols: list) -> list[int]: + return [i for i, c in enumerate(columns) if c in cat_cols] + + +def _fit_fold(X_tr, y_tr, cat_cols, seed): + """Drop columns constant in THIS train slice (HistGBM binning needs >=2 distinct), + then fit. Returns (model, kept_columns).""" + keep = [c for c in X_tr.columns if X_tr[c].nunique(dropna=True) >= 2] + X_tr = X_tr[keep] + model = HistGradientBoostingClassifier( + max_iter=250, learning_rate=0.05, max_depth=4, l2_regularization=1.0, + categorical_features=_cat_indices(keep, cat_cols), random_state=seed) + model.fit(X_tr, y_tr) + return model, keep + + +def _top_decile_rate(y_true, proba, frac=TOP_DECILE): + n = len(y_true) + k = max(1, int(round(n * frac))) + order = np.argsort(proba)[::-1][:k] + return float(np.asarray(y_true)[order].mean()), k + + +def run_probe(con, setup_version: str | None = None, n_blocks: int = 5, seed: int = 0): + """Run the date-blocked CV. Returns a result dict; also fills per-fold detail.""" + X, y, session_date, cat_cols = load_dataset(con, setup_version=setup_version) + dates = sorted(session_date.unique()) + blocks = date_blocks(dates, n_blocks) + if len(blocks) < 2: + raise ValueError(f"need >=2 date blocks for CV, got {len(blocks)} " + f"from {len(dates)} session dates") + + folds, last = [], None + for i, (tr_dates, te_dates) in enumerate(expanding_folds(blocks)): + tr = session_date.isin(tr_dates).values + te = session_date.isin(te_dates).values + y_tr, y_te = y[tr], y[te] + if y_tr.nunique() < 2 or y_te.nunique() < 2: + folds.append({"fold": i + 1, "n_test": int(te.sum()), "auc": None, + "reason": "single-class train or test slice"}) + continue + model, keep = _fit_fold(X[tr], y_tr, cat_cols, seed) + proba = model.predict_proba(X[te][keep])[:, 1] + auc = roc_auc_score(y_te, proba) + dec_rate, k = _top_decile_rate(y_te.values, proba) + base = float(y_te.mean()) + folds.append({"fold": i + 1, "n_train": int(tr.sum()), "n_test": int(te.sum()), + "auc": float(auc), "base_rate": base, "top_decile_rate": dec_rate, + "top_decile_n": k, "lift": dec_rate - base}) + last = (model, X[te][keep], y_te, keep) + + aucs = [f["auc"] for f in folds if f.get("auc") is not None] + mean_auc = float(np.mean(aucs)) if aucs else None + + perm = None + if last is not None: + model, X_te, y_te, keep = last + r = permutation_importance(model, X_te, y_te, scoring="roc_auc", + n_repeats=5, random_state=seed) + perm = sorted(zip(keep, r.importances_mean), key=lambda t: t[1], reverse=True) + + return {"setup_version": setup_version, "n_rows": int(len(y)), "n_dates": len(dates), + "features": list(X.columns), "cat_cols": cat_cols, "folds": folds, + "mean_auc": mean_auc, "permutation_importance": perm} + + +def _print_report(res: dict) -> None: + sv = res["setup_version"] or "ALL" + print(f"=== conversion signal probe: setup_version={sv} " + f"({res['n_rows']} labeled setups, {res['n_dates']} session dates) ===") + print(f"features ({len(res['features'])}): {', '.join(res['features'])}") + print(f"categoricals: {', '.join(res['cat_cols']) or '(none)'}\n") + print("expanding-window folds (train precedes test):") + for f in res["folds"]: + if f.get("auc") is None: + print(f" fold {f['fold']}: n_test={f['n_test']:5} AUC= n/a ({f.get('reason','')})") + else: + print(f" fold {f['fold']}: n_train={f['n_train']:5} n_test={f['n_test']:5} " + f"AUC={f['auc']:.3f} top-decile +1R {f['top_decile_rate']*100:.0f}% " + f"vs base {f['base_rate']*100:.0f}% (lift {f['lift']*100:+.0f}pp, n={f['top_decile_n']})") + ma = res["mean_auc"] + print(f"\nMEAN CV AUC: {ma:.4f}" if ma is not None else "\nMEAN CV AUC: n/a (no scorable folds)") + if ma is not None: + verdict = ("SIGNAL (>=0.55) — proceed to model, shadow-first" if ma >= 0.55 + else "NOISE (~0.5) — no trigger-time conversion signal; redirect to premarket setup" + if ma <= 0.52 else "INCONCLUSIVE (0.52-0.55) — accrue more forward data, re-run") + print(f"GO/NO-GO hint: {verdict}") + if res["permutation_importance"]: + print("\npermutation importance (last fold, roc_auc):") + for name, imp in res["permutation_importance"][:12]: + print(f" {name:36} {imp:+.4f}") + + +def main() -> None: + from research.multi_schema import open_research_db + ap = argparse.ArgumentParser(description="conversion-signal go/no-go probe") + ap.add_argument("--setup-version", help="filter to one setup_version (e.g. v1, pm1)") + ap.add_argument("--blocks", type=int, default=5, help="CV date blocks (default 5)") + args = ap.parse_args() + con = open_research_db("market") + res = run_probe(con, setup_version=args.setup_version, n_blocks=args.blocks) + _print_report(res) + + +if __name__ == "__main__": + main() diff --git a/tests/unit/conversion/__init__.py b/tests/unit/conversion/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/conversion/test_conversion_probe.py b/tests/unit/conversion/test_conversion_probe.py new file mode 100644 index 0000000..e44084a --- /dev/null +++ b/tests/unit/conversion/test_conversion_probe.py @@ -0,0 +1,112 @@ +"""Conversion probe (plans 008/009): the critical leakage guard + signal/noise +sanity + a static live-path guard. Isolated :memory: schema, never prod.""" + +from datetime import date, datetime, timedelta + +import numpy as np +import pytest + +from research.conversion.dataset import OUTCOME_LABEL_COLUMNS, LABEL, load_dataset +from research.conversion.signal_probe import run_probe +from storage.db import get_connection + +_SE_COLS = ["setup_id", "symbol", "setup_time", "session_date", "setup_name", + "setup_version", "gap_pct", "relative_volume", "above_vwap_flag", + "session_minute_number", "tape_regime_at_trigger", + "spy_intraday_return_pct_at_trigger", "relative_volume_source"] + + +def _seed(con, rows): + """rows = list of dicts with feature values + 'label' + row index. Writes a + setup_event and matching outcome_label (with the OTHER outcome columns populated + too, so the leakage guard is tested against real values, not nulls).""" + for i, r in enumerate(rows): + d = r["session_date"] + sid = f"S{i}:{d}" + con.execute( + f"INSERT INTO setup_events ({', '.join(_SE_COLS)}) " + f"VALUES ({', '.join(['?'] * len(_SE_COLS))})", + [sid, "SYM", datetime(d.year, d.month, d.day, 9, 34), d, "orb_completion", + r.get("setup_version", "v1"), r["gap_pct"], r["relative_volume"], + bool(r.get("above_vwap", True)), 5, r.get("tape", "chop"), + r.get("spy_ret", 0.0), r.get("rvol_src", "avg_vol_20d")]) + con.execute( + "INSERT INTO outcome_labels (setup_id, label_version, " + "reached_1r_before_minus_1r, reached_2r_before_minus_1r, " + "max_upside_next_60m, trend_day_flag, failed_breakout_flag) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + [sid, "v1", bool(r["label"]), False, 0.42, True, False]) + + +def _dates(n_days, per_day): + d0 = date(2026, 6, 1) + return [d0 + timedelta(days=i) for i in range(n_days) for _ in range(per_day)] + + +def test_dataset_has_zero_outcome_columns(): + """THE leakage test: no outcome_labels column — least of all the label itself — + may appear in the feature matrix, even though the join carries them all.""" + con = get_connection(":memory:") + try: + rng = np.random.default_rng(0) + rows = [{"session_date": d, "gap_pct": float(rng.random()), + "relative_volume": float(rng.random() * 5), "label": int(rng.integers(2))} + for d in _dates(6, 20)] + _seed(con, rows) + X, y, sdate, cats = load_dataset(con, setup_version="v1") + assert set(X.columns).isdisjoint(OUTCOME_LABEL_COLUMNS) + assert LABEL not in X.columns + assert "max_upside_next_60m" not in X.columns # a populated outcome column + assert "trend_day_flag" not in X.columns and "failed_breakout_flag" not in X.columns + assert len(y) == len(rows) and set(y.unique()) <= {0, 1} + finally: + con.close() + + +def test_probe_recovers_signal(): + """A feature that genuinely predicts the label -> mean CV AUC well above 0.5.""" + con = get_connection(":memory:") + try: + rng = np.random.default_rng(7) + rows = [] + for d in _dates(10, 40): + gap = float(rng.random()) + label = int((gap + rng.normal(0, 0.15)) > 0.5) # gap_pct carries the signal + rows.append({"session_date": d, "gap_pct": gap, + "relative_volume": float(rng.random() * 5), "label": label}) + _seed(con, rows) + res = run_probe(con, setup_version="v1", n_blocks=5, seed=0) + assert res["mean_auc"] is not None and res["mean_auc"] > 0.62 + finally: + con.close() + + +def test_probe_flat_on_pure_noise(): + """Labels independent of every feature -> mean CV AUC ~0.5 (no false signal).""" + con = get_connection(":memory:") + try: + rng = np.random.default_rng(11) + rows = [{"session_date": d, "gap_pct": float(rng.random()), + "relative_volume": float(rng.random() * 5), "label": int(rng.integers(2))} + for d in _dates(10, 40)] + _seed(con, rows) + res = run_probe(con, setup_version="v1", n_blocks=5, seed=0) + assert res["mean_auc"] is not None and 0.40 <= res["mean_auc"] <= 0.60 + finally: + con.close() + + +def test_no_live_path_imports(): + """Static guard: the plan-009 research code never reaches into any live + gate/sizing/exit module (trading_execution / runtime / run_live_paper).""" + import pathlib + root = pathlib.Path(__file__).resolve().parents[3] + files = [root / "research" / "labeler.py", + root / "research" / "ingestion" / "market_context.py", + root / "research" / "conversion" / "dataset.py", + root / "research" / "conversion" / "signal_probe.py"] + forbidden = ("trading_execution", "run_live_paper", "import runtime", "from runtime") + for f in files: + src = f.read_text() + hits = [tok for tok in forbidden if tok in src] + assert not hits, f"{f.name} references a live path: {hits}"