forked from animatedread/Warrior_EA
f8a15c7b50
- Implemented `catalog.py` to define instrument metadata and price scaling checks. - Developed `sqx.py` for decoding SQX tick .dat files, handling scaling and volume adjustments. - Created `sqxbars.py` to decode SQX bar files, ensuring price integrity against known ranges. - Introduced `volume_gate.py` for relative volume analysis on D1 bars, facilitating backtesting strategies.
35 lines
2.1 KiB
Python
35 lines
2.1 KiB
Python
"""
|
|
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")
|