conversion(009): committed go/no-go signal probe (Step 5 instrument)

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 <noreply@anthropic.com>

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.
+"""

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

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()