okWarrior_EA/research/macro_gates2.py

"""
PRE-REGISTERED round 2 (2026-10-08), written before the run. Trial counter: N=7 after the
post-hoc H4 last-bar-of-day diagnostic of G2a (counted), +4 here -> N=11.
DIX/GEX: SqueezeMetrics free daily file (2011-05 ..), published after the close; used with
next-open fills, so causal. No thresholds fitted:
  G4a  DIX > its trailing-252-day median (prior 252 days only)   G4b  complement
  G5a  GEX > 0                                                   G5b  GEX <= 0
Survival (all): ret/DD beats ungated in 2011-16, 2017+, and 2022; cadence >=2/mo; >=40% trades kept.
(2008-09 is unavailable: DIX starts 2011.)
"""
import sys, numpy as np, pandas as pd
sys.path.insert(0, ".")
import macro_gates as mg, backtest as bt
dx = pd.read_csv("macro_data/DIX.csv", index_col=0, parse_dates=True)
spx = bt.load("SP500", "PERIOD_D1")
gt = mg.gate_table(pd.Series(spx["c"], index=pd.DatetimeIndex(spx["ts"]).normalize()))
gt = gt.join(dx[["dix", "gex"]], how="left")
gt["dix_med"] = gt.dix.rolling(252, min_periods=126).median().shift(1)
G = {"G4a": lambda g: g.dix > g.dix_med, "G4b": lambda g: g.dix <= g.dix_med,
     "G5a": lambda g: g.gex > 0, "G5b": lambda g: g.gex <= 0}
W = {"2011-16": ("2011-06-01", "2017-01-01"), "2017+": ("2017-01-01", "2030-01-01"),
     "2022": ("2022-01-01", "2023-01-01"), "FULL": ("2011-06-01", "2030-01-01")}
base = mg.collect()
runs = {"BASE": base, **{k: mg.collect(f, gt) for k, f in G.items()}}
res = {}
print(f"{'run':<6}{'window':<9}{'n':>5}{'/mo':>6}{'bp':>7}{'R':>7}{'tot':>8}{'DD':>7}{'ret/DD':>8}")
for k, tr in runs.items():
    for w, (lo, hi) in W.items():
        s = mg.window_stats(tr, lo, hi); res[(k, w)] = s
        print(f"{k:<6}{w:<9}" + (f"{s['n']:>5}{s['pm']:>6.1f}{s['bp']:>7.1f}{s['R']:>7.3f}{s['tot']:>8.1%}{s['dd']:>7.1%}{s['rdd']:>8.2f}" if s else "  <3"))
    print()
for k in G:
    ok = [bool(res[(k, w)] and np.nan_to_num(res[(k, w)]["rdd"], nan=-9) > np.nan_to_num(res[("BASE", w)]["rdd"], nan=-9)) for w in ("2011-16", "2017+", "2022")]
    f = res[(k, "FULL")]; keep = len(runs[k]) / len(base)
    print(k, ok, f"{f['pm']:.1f}/mo kept {keep:.0%}", "SURVIVES" if all(ok) and f["pm"] >= 2 and keep >= .4 else "REJECTED")