research/ml_selection_backtest.py

#!/usr/bin/env python3
"""research/ml_selection_backtest.py — "give it one day, can it pick winners?" backtest.

Answers the practical question: on a given day, if the ML model RANKS the universe and we
buy its top-N BULLISH picks, does the basket make money over the next 1 / 3 / 5 days —
net of costs, and does it beat just buying the whole market that day?

This is a stock-SELECTION / portfolio test (distinct from research/ml_backtest.py, which
measures per-prediction accuracy). It uses the trained 3D model to rank, then measures the
picks' REALIZED forward close-to-close returns straight from the OHLCV cache — including the
5-day horizon the model isn't trained on, so we can honestly see the 5-day outcome.

For each out-of-sample date:
  1. Predict every ticker with a row that day (batch), keep BULLISH picks.
  2. Rank by expected up-move (up_q50), tie-break by confidence; take top-N.
  3. Realized fwd return at 1/3/5 trading days for each pick, minus 0.30% round-trip cost.
  4. Basket return = equal-weight mean; baseline = equal-weight ALL stocks that day (market).

Usage:
    python research/ml_selection_backtest.py                 # top-10, all holdout dates
    python research/ml_selection_backtest.py --top 5 --step 2
    python research/ml_selection_backtest.py --date 2026-03-15   # inspect one day's picks
"""
from __future__ import annotations

import argparse
import os
import sys

import numpy as np
import pandas as pd

_PROJ_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if _PROJ_ROOT not in sys.path:
    sys.path.insert(0, _PROJ_ROOT)

from ml_predictor.features import FEATURE_COLUMNS  # noqa: E402
from ml_predictor.infer import MLPredictor  # noqa: E402
from ml_predictor.dataset import _cached_tickers, _load_ticker, DEFAULT_STEP  # noqa: E402
from research.strategy_validation_funnel import _sharpe, _max_drawdown  # noqa: E402

DEFAULT_CSV = os.path.join(_PROJ_ROOT, "ml_predictor", "training_data.csv")
OUT_CSV = os.path.join(os.path.dirname(os.path.abspath(__file__)), "ml_selection_results.csv")

ROUND_TRIP_COST_PCT = 0.30
HORIZONS = [1, 3, 5]
_CONF_RANK = {"HIGH": 0, "MEDIUM": 1, "LOW": 2}
_SELECTOR_TF = "3D"   # use the longest-horizon model to rank for a multi-day hold


def _passes_filters(r, filters: set) -> bool:
    """Strategy/quality gates (the docs' 'Filter' stage) applied to a BULLISH candidate.
    Each is an AND gate computed from the features already in the row."""
    def g(k, d=float("nan")):
        v = r.get(k, d)
        return float(v) if v is not None else d
    if "trend" in filters:      # confirmed uptrend: above EMA50 AND EMA200
        if not (g("price_vs_ema50") > 0 and g("price_vs_ema200") > 0):
            return False
    if "momentum" in filters:   # positive momentum: MACD>0 AND 10-day return>0
        if not (g("macd_hist") > 0 and g("return_10d") > 0):
            return False
    if "adx" in filters:        # trending market only
        if not (g("adx14") > 20):
            return False
    if "trigger" in filters:    # ≥1 bullish strategy trigger fired (S-signal confirmation)
        if sum(g(f"trigger_T{n}", 0) for n in range(1, 8)) < 1:
            return False
    if "lowvol" in filters:     # avoid extreme-volatility small-caps (noise)
        v = g("atr_pct")
        if not (v == v and v <= 4.0):
            return False
    if "notob" in filters:      # not overbought
        if not (g("rsi14") < 70):
            return False
    return True


def _load_close_series() -> dict:
    """Preload every cached ticker's close Series (for realized forward returns)."""
    out = {}
    for tk in _cached_tickers():
        loaded = _load_ticker(tk)
        if loaded is not None:
            out[tk] = loaded[0].dropna()  # close series
    return out


def _fwd_ret(close: pd.Series, date: pd.Timestamp, h: int) -> float:
    """Close-to-close % return h trading days after `date` (NaN if not enough future bars)."""
    try:
        idx = close.index.searchsorted(pd.Timestamp(date))
        if idx >= len(close):
            return float("nan")
        p0 = float(close.iloc[idx])
        j = idx + h
        if j >= len(close) or p0 <= 0:
            return float("nan")
        return (float(close.iloc[j]) / p0 - 1.0) * 100.0
    except Exception:
        return float("nan")


def run(csv_path: str = DEFAULT_CSV, top_n: int = 10, step: int = 1,
        one_date: str | None = None, rank_mode: str = "expmove",
        min_conf: str = "LOW", filters: set | None = None,
        six_filter: bool = False) -> pd.DataFrame:
    filters = filters or set()
    predictor = MLPredictor()
    if not predictor.available:
        raise SystemExit("ml_predictor model not loaded — run `python ml_predictor/train.py` first.")

    df = pd.read_csv(csv_path)
    df["date"] = pd.to_datetime(df["date"])
    holdout_start = predictor.manifest.get("holdout_start")
    # --six-filter validates the ML SELECTOR itself through the doc's funnel, which needs an
    # in-sample (IS) leg too — so evaluate the FULL date range and split at holdout_start.
    if six_filter:
        oos = df.copy()
    else:
        oos = df[df["date"] >= pd.to_datetime(holdout_start)].copy() if holdout_start else df
    print(f"  Model: rank by {_SELECTOR_TF} BULLISH · mode={rank_mode} · min_conf={min_conf} · "
          f"filters={sorted(filters) or 'none'} · top-{top_n} picks/day · cost {ROUND_TRIP_COST_PCT}% round-trip")
    print(f"  Out-of-sample rows: {len(oos):,} · loading close series for realized fwd returns…")
    closes = _load_close_series()

    dates = sorted(oos["date"].unique())
    if one_date:
        dates = [pd.Timestamp(one_date)]
    else:
        dates = dates[::step]

    pick_rows = []          # per-pick detail
    day_rows = []           # per-day basket vs market
    for d in dates:
        day = oos[oos["date"] == d]
        if len(day) < top_n:
            continue
        X = day[FEATURE_COLUMNS].to_numpy(dtype=float)
        q, proba_m, classes = predictor._raw_predict(_SELECTOR_TF, X)
        median_w = float(predictor.manifest.get("tf", {}).get(_SELECTOR_TF, {}).get("median_train_width", 1.5)) or 1.5
        cand = []
        for i, (_, r) in enumerate(day.iterrows()):
            row_q = {k: float(v[i]) for k, v in q.items()}
            pred = predictor._derive(row_q, proba_m[i], classes, _SELECTOR_TF, price=100.0,
                                     atr14=float(r["atr_pct"]) if np.isfinite(r["atr_pct"]) else None,
                                     median_w=median_w, live_price=None, today_high=None,
                                     news_score=0, anchor_close=100.0)
            if pred["direction"] != "BULLISH":
                continue
            crank = _CONF_RANK.get(pred["confidence"], 3)
            if crank > _CONF_RANK.get(min_conf, 2):     # confidence filter
                continue
            if filters and not _passes_filters(r, filters):   # strategy/quality gates
                continue
            atrp = float(r["atr_pct"]) if np.isfinite(r["atr_pct"]) else 2.0
            up50 = row_q["up50"]
            proba = proba_m[i]
            bull_p = float(proba[list(classes).index("BULLISH")]) if "BULLISH" in classes else 0.0
            cand.append({"ticker": r["ticker"], "up50": up50, "crank": crank, "conf": pred["confidence"],
                         "atr_pct": max(atrp, 0.1), "bull_p": bull_p})
        if not cand:
            day_rows.append({"date": d.strftime("%Y-%m-%d"), "n_picks": 0})
            continue
        # Ranking strategy (higher score = picked first):
        if rank_mode == "riskadj":       # expected move per unit volatility (Sharpe-like)
            keyfn = lambda c: (-(c["up50"] / c["atr_pct"]), c["crank"])
        elif rank_mode == "conf":        # confidence first (bull prob), then expected move
            keyfn = lambda c: (c["crank"], -c["bull_p"], -c["up50"])
        else:                            # "expmove" (default): raw expected up-move
            keyfn = lambda c: (-c["up50"], c["crank"])
        cand.sort(key=keyfn)
        picks = [(c["ticker"], c["up50"], c["crank"], c["conf"]) for c in cand[:top_n]]

        # Realized forward returns (net of cost) for picks and for the whole market that day.
        basket = {h: [] for h in HORIZONS}
        for tk, up50, _, conf in picks:
            cs = closes.get(tk)
            rets = {h: (_fwd_ret(cs, d, h) - ROUND_TRIP_COST_PCT) if cs is not None else float("nan")
                    for h in HORIZONS}
            for h in HORIZONS:
                if not np.isnan(rets[h]):
                    basket[h].append(rets[h])
            pick_rows.append({"date": d.strftime("%Y-%m-%d"), "ticker": tk, "exp_up_q50": round(up50, 2),
                              "confidence": conf, **{f"ret_{h}d_net": round(rets[h], 3) for h in HORIZONS}})
        market = {h: [] for h in HORIZONS}
        for tk in day["ticker"]:
            cs = closes.get(tk)
            for h in HORIZONS:
                v = _fwd_ret(cs, d, h) - ROUND_TRIP_COST_PCT if cs is not None else float("nan")
                if not np.isnan(v):
                    market[h].append(v)

        row = {"date": d.strftime("%Y-%m-%d"), "n_picks": len(picks)}
        for h in HORIZONS:
            row[f"basket_{h}d"] = float(np.mean(basket[h])) if basket[h] else float("nan")
            row[f"market_{h}d"] = float(np.mean(market[h])) if market[h] else float("nan")
        day_rows.append(row)

    picks_df = pd.DataFrame(pick_rows)
    days_df = pd.DataFrame(day_rows)
    picks_df.to_csv(OUT_CSV, index=False)

    if one_date:
        _one_day_report(one_date, picks_df, days_df)
    else:
        _summary(days_df, picks_df, top_n)
    if six_filter and not one_date:
        _six_filter_verdict(days_df, pd.to_datetime(holdout_start) if holdout_start else None)
    print(f"\n  ✓ Wrote per-pick detail → {OUT_CSV}")
    return days_df


def _six_filter_verdict(days_df: pd.DataFrame, split):
    """Run the '9,120-backtest' doc's 6-filter funnel on the ML SELECTION basket itself.
    Treats each decision day's top-N basket 3-day return as one trade; splits IS/OOS at the
    model's holdout_start. NOTE: the OOS leg is only as long as the manifest holdout window —
    if that is a handful of days the verdict is directional, not conclusive."""
    HOLD = 3
    d = days_df.copy()
    d = d[d["n_picks"] > 0]
    d["_dt"] = pd.to_datetime(d["date"])
    col = "basket_3d"
    if split is None:
        is_r = []
        oos_r = list(d[col].dropna())
    else:
        is_r = list(d[d["_dt"] < split][col].dropna())
        oos_r = list(d[d["_dt"] >= split][col].dropna())
    is_s, oos_s = _sharpe(is_r, HOLD), _sharpe(oos_r, HOLD)
    mdd = _max_drawdown(oos_r)
    n_oos = len(oos_r)
    checks = [
        ("[01] OOS Sharpe > 0.5", oos_s > 0.5, f"{oos_s:+.2f}"),
        ("[02] Max DD better than -35%", mdd > -35.0, f"{mdd:.1f}%"),
        ("[03] OOS Sharpe < 2.5 (not absurd)", oos_s < 2.5, f"{oos_s:+.2f}"),
        ("[04] OOS <= IS*1.3 + 0.5 (not overfit)", oos_s <= is_s * 1.3 + 0.5, f"OOS {oos_s:+.2f} / IS {is_s:+.2f}"),
        ("[05] At least 30 OOS trades", n_oos >= 30, f"{n_oos}"),
        ("[06] IS Sharpe > 0", is_s > 0, f"{is_s:+.2f}"),
    ]
    print("\n" + "=" * 78)
    print("  6-FILTER VALIDATION — is the ML top-N SELECTION strategy a real OOS edge?")
    print(f"  IS trades={len(is_r)}  OOS trades={n_oos}  (3-day basket return per decision day)")
    print("=" * 78)
    for label, ok, val in checks:
        print(f"  {'PASS' if ok else 'FAIL'}  {label:<42} {val}")
    verdict = "SURVIVES all 6 filters" if all(c[1] for c in checks) else "does NOT survive"
    print(f"\n  → The ML selection strategy {verdict}.")
    if n_oos < 30:
        print("  ⚠ OOS window is short (manifest holdout is small) — treat as directional only.")


def _summary(days_df: pd.DataFrame, picks_df: pd.DataFrame, top_n: int):
    active = days_df[days_df["n_picks"] > 0]
    print("\n" + "=" * 78)
    print(f"  STOCK-SELECTION BACKTEST — top-{top_n} BULLISH picks, equal-weight, held N days")
    print(f"  {len(active)} decision days · {len(picks_df):,} total picks · net of {ROUND_TRIP_COST_PCT}% cost")
    print("=" * 78)
    print(f"  {'Hold':<7}{'BasketAvg':>11}{'MarketAvg':>11}{'Edge':>9}{'PickWin%':>10}{'DaysBeatMkt':>13}{'BookRet':>9}")
    print("  " + "-" * 70)
    for h in HORIZONS:
        b = active[f"basket_{h}d"].dropna()
        m = active[f"market_{h}d"].dropna()
        pick_win = float((picks_df[f"ret_{h}d_net"] > 0).mean()) if len(picks_df) else float("nan")
        beat = float((active[f"basket_{h}d"] > active[f"market_{h}d"]).mean())
        # Equal-weight daily-rebalanced book return: compound each day's basket mean across days.
        book = 1.0
        for v in active[f"basket_{h}d"].dropna():
            book *= (1 + v / 100.0)
        book_ret = (book - 1) * 100.0
        print(f"  {str(h)+'d':<7}{b.mean():>+11.2f}{m.mean():>+11.2f}{(b.mean()-m.mean()):>+9.2f}"
              f"{pick_win:>10.0%}{beat:>13.0%}{book_ret:>+9.1f}")
    print("\n  BasketAvg/MarketAvg = mean per-pick vs per-stock net return over the hold.")
    print("  Edge = how much the picks beat buying the whole market. PickWin% = picks that closed green.")
    print("  DaysBeatMkt = fraction of decision days the basket beat the market.")
    print(f"  BookRet compounds each day's top-{top_n} basket across all decision days (rebalanced).")
    # Direct answer to the 5-day question
    b5 = active["basket_5d"].dropna()
    m5 = active["market_5d"].dropna()
    win5 = float((picks_df["ret_5d_net"] > 0).mean()) if len(picks_df) else float("nan")
    verdict = "YES" if (b5.mean() > 0 and b5.mean() > m5.mean()) else ("MARGINAL" if b5.mean() > 0 else "NO")
    print("\n  ── 5-DAY VERDICT ──")
    print(f"  Avg pick makes {b5.mean():+.2f}% over 5 days (net) vs market {m5.mean():+.2f}% · "
          f"{win5:.0%} of picks close green · edge {b5.mean()-m5.mean():+.2f}%  →  profitable? {verdict}")


def _one_day_report(date: str, picks_df: pd.DataFrame, days_df: pd.DataFrame):
    print("\n" + "=" * 78)
    print(f"  SINGLE-DAY PICKS — {date}")
    print("=" * 78)
    if picks_df.empty:
        print("  No BULLISH picks that day.")
        return
    cols = ["ticker", "confidence", "exp_up_q50", "ret_1d_net", "ret_3d_net", "ret_5d_net"]
    print(picks_df[cols].to_string(index=False))
    if not days_df.empty and days_df.iloc[0].get("n_picks", 0) > 0:
        r = days_df.iloc[0]
        print(f"\n  Basket avg — 1d {r['basket_1d']:+.2f}%  3d {r['basket_3d']:+.2f}%  5d {r['basket_5d']:+.2f}%  "
              f"(net) vs market 5d {r['market_5d']:+.2f}%")


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--csv", default=DEFAULT_CSV)
    ap.add_argument("--top", type=int, default=10)
    ap.add_argument("--step", type=int, default=1, help="sample every Nth decision day")
    ap.add_argument("--date", default=None, help="inspect a single date (YYYY-MM-DD)")
    ap.add_argument("--rank", default="expmove", choices=["expmove", "riskadj", "conf"],
                    help="ranking strategy for picks")
    ap.add_argument("--min-conf", default="LOW", choices=["LOW", "MEDIUM", "HIGH"],
                    help="only pick stocks at/above this confidence")
    ap.add_argument("--filters", default="", help="comma-separated quality gates: "
                    "trend,momentum,adx,trigger,lowvol,notob")
    ap.add_argument("--six-filter", action="store_true",
                    help="validate the ML selection basket through the doc's 6-filter funnel (IS vs OOS)")
    args = ap.parse_args()
    filters = {f.strip() for f in args.filters.split(",") if f.strip()}
    run(args.csv, top_n=args.top, step=args.step, one_date=args.date,
        rank_mode=args.rank, min_conf=args.min_conf, filters=filters,
        six_filter=args.six_filter)


if __name__ == "__main__":
    main()