54e3d50db8
Fused all ideas: QQQ core + foxhunt controllers as a ROTATOR (edge-decay-trust + resurrection + drawdown-CB) rotating QQQ<->crisis-alpha by regime. Pre-registered criterion: beat static GD on OOS Calmar. RESULT: full-sample fusion Calmar 0.41 > GD 0.34 (IS-flattered by 2008), but OOS (2018-2026) static GD 0.64 > fusion 0.58, even pure QQQ 0.59 > fusion -- GD strictly dominates OOS (higher CAGR + lower DD). Regime-rotation whipsaw + late-re-entry cost exceeds the protection benefit in bulls. Verdict: fusion fails OOS -> keep the simpler static GD. Simple beats fancy; only OOS separates them. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
102 lines
5.3 KiB
Python
102 lines
5.3 KiB
Python
#!/usr/bin/env python3
|
|
"""FUSION backtest: regime-adaptive QQQ <-> crisis-alpha rotation (the synthesis of all the ideas).
|
|
|
|
QQQ core whose weight is driven by edge-decay-TRUST (trailing risk-adjusted health, EMA-smoothed,
|
|
resurrection-capable) — NOT vol-normalized (so it keeps QQQ's CAGR in calm regimes). When QQQ's trust
|
|
decays (sustained downtrend/crisis) it ROTATES into the crisis-alpha sleeve (trend / diversified book,
|
|
which RISES in crises) instead of to cash; resurrects back to QQQ as it heals. Plus a light drawdown
|
|
circuit-breaker as a tail backstop. Honest test: vs QQQ / book / static GD, full + IS/OOS split +
|
|
turnover (the whipsaw cost). Verdict: fusion must beat static GD on CAGR-per-drawdown (Calmar), esp OOS.
|
|
"""
|
|
import datetime
|
|
import json
|
|
import math
|
|
import os
|
|
import sys
|
|
import urllib.request
|
|
|
|
import numpy as np
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from multistrat_paper import book_series # noqa: E402
|
|
|
|
|
|
def yh(s):
|
|
r = json.loads(urllib.request.urlopen(urllib.request.Request(
|
|
f"https://query1.finance.yahoo.com/v8/finance/chart/{s}?interval=1d&range=25y",
|
|
headers={"User-Agent": "Mozilla/5.0"}), timeout=30).read())["chart"]["result"][0]
|
|
a = r["indicators"].get("adjclose", [{}])[0].get("adjclose") or r["indicators"]["quote"][0]["close"]
|
|
return {datetime.datetime.utcfromtimestamp(t).strftime("%Y-%m-%d"): float(c) for t, c in zip(r["timestamp"], a) if c}
|
|
|
|
|
|
def edge_decay_trust(ret, win=126, ema=0.94):
|
|
"""trailing-Sharpe health -> theta in [0.1, 1], EMA-smoothed, resurrection-capable."""
|
|
T = len(ret); th = 1.0; out = np.ones(T)
|
|
for t in range(win, T):
|
|
seg = ret[t - win:t]; sr = seg.mean() / (seg.std() + 1e-9) * math.sqrt(252)
|
|
target = float(np.clip((sr + 0.5) / 1.0, 0.10, 1.0))
|
|
th = ema * th + (1 - ema) * target; out[t] = th
|
|
return out
|
|
|
|
|
|
def main():
|
|
syms = ["QQQ", "SPY", "IEF", "GLD", "DBC"]
|
|
data = {s: yh(s) for s in syms}
|
|
common = sorted(set.intersection(*[set(data[s]) for s in syms])); T = len(common)
|
|
yrs = (datetime.date.fromisoformat(common[-1]) - datetime.date.fromisoformat(common[0])).days / 365.25
|
|
P = {s: np.array([data[s][d] for d in common]) for s in syms}
|
|
R = {s: np.concatenate([[0], P[s][1:] / P[s][:-1] - 1]) for s in syms}
|
|
qqq = R["QQQ"]
|
|
R4 = np.column_stack([R[s] for s in ["SPY", "IEF", "GLD", "DBC"]]); P4 = np.column_stack([P[s] for s in ["SPY", "IEF", "GLD", "DBC"]])
|
|
# trend sleeve (12-1 TS-mom, equal-risk = managed-futures proxy) + diversified book
|
|
lc = np.log(P4); sg = np.zeros((T, 4)); sg[252:] = np.sign(lc[252:] - lc[:-252])
|
|
vol = np.full((T, 4), np.nan)
|
|
for t in range(63, T):
|
|
vol[t] = R4[t - 63:t].std(0)
|
|
iv = 1 / np.where(vol > 0, vol, np.nan); w = np.nan_to_num(sg * iv); g = np.abs(w).sum(1, keepdims=True); g[g == 0] = 1; w = w / g
|
|
trend = np.zeros(T); trend[1:] = np.sum(w[:-1] * R4[1:], axis=1)
|
|
book, _, _ = book_series(np.column_stack([R4, trend]))
|
|
|
|
# FUSION: w_qqq = trust(QQQ); rotate (1-w) into the crisis-alpha sleeve; + tail drawdown-CB
|
|
theta = edge_decay_trust(qqq)
|
|
def fusion(sleeve, dd_deadband=0.15):
|
|
wq = theta.copy(); ws = 1 - wq
|
|
r = np.zeros(T); eq = pk = 1.0; turn = 0.0
|
|
for t in range(252, T - 1):
|
|
base = wq[t] * qqq[t + 1] + ws[t] * sleeve[t + 1]
|
|
dd = eq / pk - 1
|
|
cb = float(np.clip(1 - 3 * max(0.0, -dd - dd_deadband), 0.5, 1.0)) # tail backstop only (>15% dd)
|
|
r[t + 1] = cb * base; eq *= (1 + r[t + 1]); pk = max(pk, eq)
|
|
turn += abs(wq[t] - wq[t - 1])
|
|
return r, turn / yrs # annualized one-way turnover of the QQQ leg
|
|
fus_trend, turn_t = fusion(trend)
|
|
fus_book, turn_b = fusion(book)
|
|
|
|
static_gd = 0.7 * qqq + 0.3 * book
|
|
|
|
def stats(r, lo=0, hi=None):
|
|
seg = r[lo:hi] if hi else r[lo:]; seg = np.nan_to_num(seg)
|
|
n = (seg != 0).sum(); y = n / 252 if n else 1
|
|
eq = np.cumprod(1 + seg); cagr = eq[-1] ** (1 / y) - 1 if y > 0 else 0
|
|
v = seg.std() * math.sqrt(252); dd = float((eq / np.maximum.accumulate(eq) - 1).min())
|
|
sh = (seg.mean() * 252 - 0.03) / v if v > 0 else 0
|
|
calmar = cagr / abs(dd) if dd < 0 else 0
|
|
return cagr, v, sh, dd, calmar
|
|
sp = int(0.6 * T)
|
|
print(f"FUSION BACKTEST — {common[0]}..{common[-1]} ({yrs:.0f}y), OOS split @ {common[sp]}")
|
|
print(f" {'strategy':>26} | seg | CAGR | maxDD | Sharpe* | Calmar | turn/yr")
|
|
cands = [("QQQ buy-hold", qqq, 0), ("book (diversified)", book, 0), ("static GD 70/30", static_gd, 0),
|
|
("FUSION QQQ<->trend", fus_trend, turn_t), ("FUSION QQQ<->book", fus_book, turn_b)]
|
|
for nm, r, turn in cands:
|
|
for seg, lab in [((0, None), "ALL"), ((0, sp), "IS "), ((sp, None), "OOS")]:
|
|
c, v, sh, dd, cal = stats(r, seg[0], seg[1])
|
|
tt = f"{turn:.1f}x" if (lab == "ALL" and turn) else ""
|
|
print(f" {nm if lab=='ALL' else '':>26} | {lab} | {100*c:>4.1f}% | {100*dd:>+4.0f}% | {sh:>+5.2f} | {cal:>5.2f} | {tt}")
|
|
print(" *Sharpe excess over financing. Calmar = CAGR/|maxDD| (higher=better frontier).")
|
|
print(" VERDICT: FUSION beats static GD on OOS Calmar = regime-rotation shifts the frontier out (worth it).")
|
|
print(" If not = whipsaw eats it; keep the simpler static GD.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|