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.
41 lines
2.2 KiB
Python
41 lines
2.2 KiB
Python
"""Confirmation of G2a (VRP>5) -- NO new thresholds, trial counter stays N=6."""
|
|
import sys, numpy as np, pandas as pd
|
|
sys.path.insert(0, ".")
|
|
import backtest as bt, macro_gates as mg
|
|
from scipy import stats
|
|
from run_screen import zscore_entries
|
|
|
|
spx = bt.load("SP500", "PERIOD_D1")
|
|
gt = mg.gate_table(pd.Series(spx["c"], index=pd.DatetimeIndex(spx["ts"]).normalize()))
|
|
G2a, G2b = mg.GATES["G2a"], mg.GATES["G2b"]
|
|
|
|
# D1: per-symbol split and Welch t of G2a vs G2b trade returns
|
|
a, b = mg.collect(G2a, gt), mg.collect(G2b, gt)
|
|
print("D1 G2a vs G2b trades:", len(a), len(b))
|
|
for s in mg.IDX:
|
|
x = [t["ret"] * 1e4 for t in a if t["symbol"] == s]; y = [t["ret"] * 1e4 for t in b if t["symbol"] == s]
|
|
print(f" {s:<7} G2a n={len(x):>3} {np.mean(x):7.1f}bp | G2b n={len(y):>3} {np.mean(y):7.1f}bp")
|
|
ra = [t["ret"] * 1e4 for t in a]; rb = [t["ret"] * 1e4 for t in b]
|
|
print(" Welch t(G2a-G2b) = %.2f p=%.4f" % stats.ttest_ind(ra, rb, equal_var=False))
|
|
print(" G2a alone vs 0: t=%.2f" % stats.ttest_1samp(ra, 0).statistic)
|
|
|
|
# H4 2021+: gate by PRIOR day's macro value (conservative; bar may be intraday)
|
|
def collect_h4(gate):
|
|
out = []
|
|
for s in mg.IDX:
|
|
d = bt.load(s, "PERIOD_H4"); e, xma = zscore_entries(d, 20, -1.5, 0)
|
|
if gate is not None:
|
|
day = pd.DatetimeIndex(d["ts"]).normalize() - pd.Timedelta(days=1)
|
|
ok = gate(gt).reindex(day, method="ffill").to_numpy() if False else gate(gt).reindex(gt.index).ffill()
|
|
prev = ok.shift(1).reindex(pd.DatetimeIndex(d["ts"]).normalize(), method="ffill").to_numpy()
|
|
e = e & np.where(pd.isna(prev), False, prev).astype(bool)
|
|
tr = bt.simulate(d, e, 1, xma, 10, 3.0); tr = bt.add_r(tr, d, 3.0)
|
|
for t in tr: t["symbol"] = s; t["exit_t"] = d["ts"][t["exit_i"]]
|
|
out += tr
|
|
return out
|
|
print("\nH4 2021+ (independent bars/trades)")
|
|
for name, g in (("BASE", None), ("G2a", G2a), ("G2b", G2b)):
|
|
tr = collect_h4(g)
|
|
for w, lo, hi in (("FULL", "2021-01-01", "2030-01-01"), ("2022", "2022-01-01", "2023-01-01"), ("2023+", "2023-01-01", "2030-01-01")):
|
|
s = mg.window_stats(tr, lo, hi)
|
|
if s: print(f" {name:<5}{w:<6} n={s['n']:>4} {s['pm']:>5.1f}/mo {s['bp']:>6.1f}bp R={s['R']:>6.3f} tot={s['tot']:>6.1%} DD={s['dd']:>5.1%} ret/DD={s['rdd']:>6.2f}")
|