Warrior_EA/research/cond_scan.py

"""
PRE-REGISTERED conditional-direction scan (2026-10-08): "given vol regime x recent move x weekday, where does price go?"
Universe: SP500/NAS100/US30/DAX40 D1, pooled. Features known at the close of day t:
  VRP bucket {<=0, 0-5, >5}   (VIX - RV10 of SP500, same as macro_gates)
  ret1 bucket {<=-1%, mid, >=+1%} (that index's own prior-day return)
  ret5 bucket {<=-2%, mid, >=+2%} (5-day return)
  weekday {Mon..Fri}            => 135 cells.   Outcome: open(t+1) -> close(t+3), minus real round-trip spread.
Walk-forward: for test year Y = 2013..2026 select on data < Y: cells with n>=40 and |t| >= 2.5 on the
(excess over train mean) outcome, top 5 by |t|, trade each in the sign of its train excess in year Y.
Statistic: mean net bp per selected-cell trade over all OOS years. NULL: the identical procedure with the
cell label of each day randomly permuted within each index (500 permutations) -> p = share of null >= observed.
Pass: observed stat > 0 and permutation p <= 0.05. One config, no re-runs with other numbers.
"""
import sys, numpy as np, pandas as pd
sys.path.insert(0, ".")
import backtest as bt, macro_gates as mg
H = 3
def build():
    spx = bt.load("SP500", "PERIOD_D1")
    gt = mg.gate_table(pd.Series(spx["c"], index=pd.DatetimeIndex(spx["ts"]).normalize()))
    rows = []
    for s in mg.IDX:
        d = bt.load(s, "PERIOD_D1"); ts = pd.DatetimeIndex(d["ts"]).normalize(); c, o = d["c"], d["o"]; n = len(c)
        df = pd.DataFrame({"sym": s, "date": ts, "c": c}, index=range(n))
        df["r1"] = pd.Series(c).pct_change(); df["r5"] = pd.Series(c).pct_change(5)
        df["vrp"] = gt.vrp.reindex(ts).to_numpy(); df["dow"] = ts.dayofweek
        fwd = np.full(n, np.nan); 
        for i in range(n - H - 1):
            fwd[i] = (c[i + H + 1] - o[i + 1]) / o[i + 1] - d["cost"][i + 1] / o[i + 1]   # buy next open, sell close of t+4? (H+1 bars later)
        df["fwd"] = fwd * 1e4; df["cost"] = d["cost"] / np.r_[o[1:], o[-1]] * 1e4
        rows.append(df)
    X = pd.concat(rows, ignore_index=True).dropna(subset=["r1", "r5", "vrp", "fwd"])
    X["cell"] = (pd.cut(X.vrp, [-1e9, 0, 5, 1e9], labels=False) * 45 + pd.cut(X.r1, [-1, -.01, .01, 1], labels=False) * 15
                 + pd.cut(X.r5, [-1, -.02, .02, 1], labels=False) * 5 + X.dow).astype(int)
    X["year"] = X.date.dt.year
    return X

def procedure(X, cell, years=range(2013, 2027), K=5, MINN=40, TH=2.5):
    pnl = []
    for Y in years:
        tr, te = X.year < Y, X.year == Y
        g = X[tr].assign(cell=cell[tr]).groupby("cell").fwd
        m, sd, nn = g.mean(), g.std(), g.count(); mu = X[tr].fwd.mean()
        t = (m - mu) / (sd / np.sqrt(nn)); t = t[(nn >= MINN) & (t.abs() >= TH)]
        pick = t.reindex(t.abs().sort_values(ascending=False).index)[:K]
        for cl, tv in pick.items():
            sel = X[te & (cell == cl)]
            if len(sel): pnl += list(np.sign(tv) * (sel.fwd + sel.cost) - sel.cost)   # direction flips the gross, cost always paid
    return np.array(pnl)

if __name__ == "__main__":
    X = build(); cell = X.cell.copy()
    obs = procedure(X, cell)
    print(f"cells {X.cell.nunique()}  rows {len(X)}  OOS selected trades {len(obs)}  mean net {obs.mean():.1f} bp  hit {(obs>0).mean():.1%}")
    rng = np.random.default_rng(0); null = []
    for k in range(500):
        sh = X.groupby("sym").cell.transform(lambda v: rng.permutation(v.to_numpy()))
        p = procedure(X, sh); null.append(p.mean() if len(p) else np.nan)
    null = np.array(null); null = null[~np.isnan(null)]
    print(f"null mean {null.mean():.1f} sd {null.std():.1f}  p(null>=obs) = {(null >= obs.mean()).mean():.3f}")
    print("VERDICT:", "PASS" if obs.mean() > 0 and (null >= obs.mean()).mean() <= .05 else "REJECTED")