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