momentum-system/research/conversion/signal_probe.py

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