Gahow Wang
0d983edfc0
Four research directions beyond V7+VT36: 1. single_stock_swing: 20 famous stocks (Mag 7 + others), per-stock optimized swing trading. High-vol growth stocks (AMD Sharpe 1.55, TSLA 1.54) work best, but overfitting risk is extreme — universal params only TSLA is viable. Not competitive with V7. 2. v7_literature_alpha: 9 academic directions (VIX overlay, Kelly sizing, multi-MA, cross-asset, momentum acceleration, VIX mean- reversion, vol-adaptive PT, combined). V3's regime engine already implicitly captures most literature signals. MA130 marginally better than MA150 (+0.02 Sharpe, within noise). 3. new_frameworks_eval: volatility trading (SVXY risk-off) and calendar effects (turn-of-month). SVXY and V7 regime structurally conflict — SVXY crashes exactly when V7 goes risk-off. Turn-of-month has decent Sharpe (1.30) but only 28% annual. Nothing beats V7. 4. smart_dca_eval: fixed/VIX-scaled/MA-deviation/value-averaging/RSI DCA into SPY/QQQ/TQQQ/UPRO + V7 hybrids. Smart DCA barely beats fixed DCA. Any DCA hybrid dilutes V7's alpha. DCA only useful for new monthly contributions that can't lump-sum into V7. Conclusion: V7+VT36 remains SOTA across all tested frameworks. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
434 lines
17 KiB
Python
434 lines
17 KiB
Python
"""New frameworks: Vol trading (#2) + Event-driven simplified (#4).
|
|
|
|
Framework 2: Volatility Trading
|
|
A. V7 risk-off → SVXY (harvest vol premium during normal risk-off)
|
|
B. Standalone SVXY mean-reversion (buy SVXY when VIX spikes + reverts)
|
|
C. V7 + SVXY risk-off with VIX gate (no SVXY when VIX > 30)
|
|
|
|
Framework 4: Event-Driven Simplified
|
|
D. Earnings calendar effect: avoid holding around earnings (high idio risk)
|
|
→ Not applicable to ETFs. Instead: monthly seasonality on leveraged ETFs
|
|
E. Turn-of-month effect (known anomaly: last 3 + first 3 days of month)
|
|
F. Holiday effect (market tends to rise before holidays)
|
|
G. Options expiration week effect (OpEx week has different dynamics)
|
|
|
|
Also: pure standalone strategies (not V7 modifications)
|
|
H. Pure SVXY with VIX regime gating
|
|
I. TQQQ buy-the-dip: buy TQQQ when RSI < 30, sell when RSI > 70
|
|
J. VIX term structure carry: long SVXY when VIX contango, cash when backwardation
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
sys.path.insert(0, ".")
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
import data_manager
|
|
import metrics
|
|
from main import backtest
|
|
from strategies.base import Strategy
|
|
from strategies.permanent import TrendRiderV3
|
|
from strategies.trend_rider_v7 import TrendRiderV7
|
|
|
|
YEARS = 10
|
|
CAPITAL = 100_000
|
|
TX_COST = 0.001
|
|
FIXED_FEE = 2.0
|
|
|
|
|
|
# =========================================================================
|
|
# Framework 2: Volatility Trading Strategies
|
|
# =========================================================================
|
|
|
|
class V7SvxyRiskOff(Strategy):
|
|
"""V7 with SVXY as risk-off instrument (gated by VIX level).
|
|
|
|
During risk-off:
|
|
- If VIX < vix_gate: hold SVXY (harvest vol premium)
|
|
- If VIX >= vix_gate: hold GLD/DBC (traditional safe haven)
|
|
"""
|
|
def __init__(self, vix_gate=30, svxy="SVXY", **v7_kw):
|
|
self.vix_gate = vix_gate
|
|
self.svxy = svxy
|
|
self.v7 = TrendRiderV7(**v7_kw)
|
|
|
|
def generate_signals(self, data):
|
|
w = self.v7.generate_signals(data)
|
|
svxy = self.svxy
|
|
if svxy not in data.columns or "^VIX" not in data.columns:
|
|
return w
|
|
if svxy not in w.columns:
|
|
w[svxy] = 0.0
|
|
|
|
vix = data["^VIX"].shift(2).fillna(20) # PIT: 2-day lag matching V3
|
|
roff_cols = [c for c in ["GLD", "DBC"] if c in w.columns]
|
|
|
|
for i in range(len(w)):
|
|
roff_w = sum(float(w.iat[i, w.columns.get_loc(c)]) for c in roff_cols)
|
|
if roff_w < 1e-8:
|
|
continue
|
|
v = vix.iloc[i]
|
|
if np.isnan(v) or v >= self.vix_gate:
|
|
continue
|
|
# Replace risk-off with SVXY
|
|
for c in roff_cols:
|
|
w.iat[i, w.columns.get_loc(c)] = 0.0
|
|
w.iat[i, w.columns.get_loc(svxy)] = roff_w
|
|
return w
|
|
|
|
|
|
class PureSvxyStrategy(Strategy):
|
|
"""Standalone SVXY with VIX regime gating.
|
|
|
|
Hold SVXY when VIX < upper_gate AND VIX > VIX_MA (mean-reverting down).
|
|
Cash otherwise. Vol-target overlay for sizing.
|
|
"""
|
|
def __init__(self, upper_gate=25, ma_window=20, target_vol=0.20,
|
|
min_lev=0.3, svxy="SVXY"):
|
|
self.upper_gate = upper_gate
|
|
self.ma_window = ma_window
|
|
self.target_vol = target_vol
|
|
self.min_lev = min_lev
|
|
self.svxy = svxy
|
|
|
|
def generate_signals(self, data):
|
|
cols = [c for c in [self.svxy, "^VIX", "SHY"] if c in data.columns]
|
|
w = pd.DataFrame(0.0, index=data.index, columns=cols + [self.svxy])
|
|
if self.svxy not in data.columns or "^VIX" not in data.columns:
|
|
return w
|
|
|
|
vix = data["^VIX"]
|
|
vix_ma = vix.rolling(self.ma_window).mean()
|
|
# Hold SVXY when: VIX < gate AND VIX is falling (below its MA)
|
|
hold_signal = ((vix < self.upper_gate) & (vix < vix_ma)).shift(2).fillna(False)
|
|
|
|
w[self.svxy] = hold_signal.astype(float)
|
|
w = w.fillna(0.0)
|
|
|
|
# Vol-target
|
|
if self.svxy in data.columns:
|
|
rets = data[self.svxy].pct_change(fill_method=None).fillna(0.0)
|
|
rv = rets.rolling(60, min_periods=21).std() * np.sqrt(252)
|
|
scale = (self.target_vol / rv).clip(lower=self.min_lev, upper=1.0).shift(1).fillna(1.0)
|
|
w = w.mul(scale, axis=0)
|
|
|
|
return w
|
|
|
|
|
|
class TqqqRsiStrategy(Strategy):
|
|
"""TQQQ buy-the-dip: buy when RSI < oversold, sell when RSI > overbought."""
|
|
def __init__(self, rsi_window=14, buy_level=30, sell_level=70,
|
|
target_vol=0.30, min_lev=0.5):
|
|
self.rsi_window = rsi_window
|
|
self.buy_level = buy_level
|
|
self.sell_level = sell_level
|
|
self.target_vol = target_vol
|
|
self.min_lev = min_lev
|
|
|
|
def generate_signals(self, data):
|
|
if "SPY" not in data.columns or "TQQQ" not in data.columns:
|
|
return pd.DataFrame(0.0, index=data.index, columns=data.columns)
|
|
|
|
# RSI on SPY (not TQQQ — SPY is less noisy)
|
|
spy = data["SPY"]
|
|
delta = spy.diff()
|
|
gain = delta.clip(lower=0)
|
|
loss = (-delta).clip(lower=0)
|
|
avg_gain = gain.rolling(self.rsi_window).mean()
|
|
avg_loss = loss.rolling(self.rsi_window).mean()
|
|
rs = avg_gain / avg_loss.clip(lower=1e-10)
|
|
rsi = 100 - (100 / (1 + rs))
|
|
rsi = rsi.shift(1) # PIT
|
|
|
|
cols = [c for c in data.columns]
|
|
w = pd.DataFrame(0.0, index=data.index, columns=cols)
|
|
|
|
in_position = False
|
|
for i in range(len(data)):
|
|
r = rsi.iloc[i]
|
|
if np.isnan(r):
|
|
continue
|
|
if not in_position and r < self.buy_level:
|
|
in_position = True
|
|
elif in_position and r > self.sell_level:
|
|
in_position = False
|
|
if in_position and "TQQQ" in w.columns:
|
|
w.iat[i, w.columns.get_loc("TQQQ")] = 1.0
|
|
|
|
# Vol-target
|
|
rets = data["TQQQ"].pct_change(fill_method=None).fillna(0.0) if "TQQQ" in data.columns else pd.Series(0.0, index=data.index)
|
|
rv = rets.rolling(60, min_periods=21).std() * np.sqrt(252)
|
|
scale = (self.target_vol / rv).clip(lower=self.min_lev, upper=1.0).shift(1).fillna(1.0)
|
|
w = w.mul(scale, axis=0)
|
|
return w
|
|
|
|
|
|
# =========================================================================
|
|
# Framework 4: Calendar/Seasonality Strategies
|
|
# =========================================================================
|
|
|
|
class TurnOfMonthStrategy(Strategy):
|
|
"""Exploit turn-of-month anomaly: hold TQQQ last N + first M trading days.
|
|
|
|
Lakonishok & Smidt (1988): stocks earn disproportionate returns
|
|
at month boundaries. Payroll flows, pension rebalancing.
|
|
"""
|
|
def __init__(self, days_before=3, days_after=3, target_vol=0.36, min_lev=0.75):
|
|
self.days_before = days_before
|
|
self.days_after = days_after
|
|
self.target_vol = target_vol
|
|
self.min_lev = min_lev
|
|
|
|
def generate_signals(self, data):
|
|
cols = [c for c in data.columns]
|
|
w = pd.DataFrame(0.0, index=data.index, columns=cols)
|
|
if "TQQQ" not in data.columns:
|
|
return w
|
|
|
|
# Identify turn-of-month windows
|
|
dates = data.index
|
|
months = dates.to_period("M")
|
|
|
|
for period in months.unique():
|
|
month_mask = months == period
|
|
month_dates = dates[month_mask]
|
|
if len(month_dates) < 5:
|
|
continue
|
|
# Last N days of month
|
|
for d in month_dates[-self.days_before:]:
|
|
w.at[d, "TQQQ"] = 1.0
|
|
# First M days of month
|
|
for d in month_dates[:self.days_after]:
|
|
w.at[d, "TQQQ"] = 1.0
|
|
|
|
# Shift for PIT
|
|
w = w.shift(1).fillna(0.0)
|
|
|
|
# Vol-target
|
|
if "TQQQ" in data.columns:
|
|
rets = data["TQQQ"].pct_change(fill_method=None).fillna(0.0)
|
|
port_rets = (w["TQQQ"] * rets)
|
|
rv = port_rets.rolling(60, min_periods=21).std() * np.sqrt(252)
|
|
scale = (self.target_vol / rv).clip(lower=self.min_lev, upper=1.0).shift(1).fillna(1.0)
|
|
w = w.mul(scale, axis=0)
|
|
return w
|
|
|
|
|
|
class V7TurnOfMonth(Strategy):
|
|
"""V7 but only enters risk-on during turn-of-month windows.
|
|
|
|
Combines V7 regime timing with monthly seasonality.
|
|
Outside the window: force to risk-off even if V7 says risk-on.
|
|
"""
|
|
def __init__(self, days_before=4, days_after=4, **v7_kw):
|
|
self.days_before = days_before
|
|
self.days_after = days_after
|
|
self.v7 = TrendRiderV7(**v7_kw)
|
|
|
|
def generate_signals(self, data):
|
|
w = self.v7.generate_signals(data)
|
|
|
|
dates = data.index
|
|
months = dates.to_period("M")
|
|
in_window = pd.Series(False, index=dates)
|
|
for period in months.unique():
|
|
month_dates = dates[months == period]
|
|
if len(month_dates) < 5:
|
|
continue
|
|
for d in month_dates[-self.days_before:]:
|
|
in_window[d] = True
|
|
for d in month_dates[:self.days_after]:
|
|
in_window[d] = True
|
|
|
|
risk_on_cols = [c for c in ["TQQQ", "UPRO"] if c in w.columns]
|
|
risk_off_cols = [c for c in ["GLD", "DBC"] if c in w.columns]
|
|
park = "SHY" if "SHY" in w.columns else ""
|
|
|
|
for i in range(len(w)):
|
|
if in_window.iloc[i]:
|
|
continue
|
|
ron_w = sum(float(w.iat[i, w.columns.get_loc(c)]) for c in risk_on_cols)
|
|
if ron_w > 0.01:
|
|
for c in risk_on_cols:
|
|
w.iat[i, w.columns.get_loc(c)] = 0.0
|
|
if park and park in w.columns:
|
|
w.iat[i, w.columns.get_loc(park)] = ron_w
|
|
return w
|
|
|
|
|
|
# =========================================================================
|
|
# Hybrid: V7 risk-on + Vol premium risk-off
|
|
# =========================================================================
|
|
|
|
class V7VolHybrid(Strategy):
|
|
"""Best of both worlds: V7 regime for risk-on, SVXY for risk-off.
|
|
|
|
Risk-on: TQQQ/UPRO (V7 momentum pick) — trend alpha
|
|
Risk-off: SVXY when VIX < gate, GLD when VIX >= gate — vol premium alpha
|
|
Vol-target + PT on risk-on only.
|
|
|
|
Two independent alpha sources:
|
|
1. Equity trend momentum (V7's regime timing)
|
|
2. Volatility risk premium (SVXY during calm periods)
|
|
"""
|
|
def __init__(self, vix_gate=28, target_vol=0.36, min_lev=0.75,
|
|
pt_threshold=0.30, pt_band=0.10):
|
|
self.vix_gate = vix_gate
|
|
self.target_vol = target_vol
|
|
self.min_lev = min_lev
|
|
self.pt_threshold = pt_threshold
|
|
self.pt_band = pt_band
|
|
self.v3 = TrendRiderV3(signal="SPY", risk_on=("TQQQ", "UPRO"),
|
|
risk_off=("GLD", "DBC"), ma_long=150)
|
|
|
|
def generate_signals(self, data):
|
|
w = self.v3.generate_signals(data)
|
|
for col in ["SHY", "SVXY"]:
|
|
if col in data.columns and col not in w.columns:
|
|
w[col] = 0.0
|
|
|
|
has_vix = "^VIX" in data.columns
|
|
has_svxy = "SVXY" in data.columns
|
|
if has_vix and has_svxy:
|
|
vix = data["^VIX"].shift(2).fillna(20)
|
|
roff_cols = [c for c in ["GLD", "DBC"] if c in w.columns]
|
|
for i in range(len(w)):
|
|
roff_w = sum(float(w.iat[i, w.columns.get_loc(c)]) for c in roff_cols)
|
|
if roff_w < 1e-8: continue
|
|
v = vix.iloc[i]
|
|
if np.isnan(v) or v >= self.vix_gate: continue
|
|
for c in roff_cols:
|
|
w.iat[i, w.columns.get_loc(c)] = 0.0
|
|
w.iat[i, w.columns.get_loc("SVXY")] = roff_w
|
|
|
|
# Vol-target
|
|
daily_ret = data.pct_change(fill_method=None).fillna(0.0)
|
|
common = w.columns.intersection(daily_ret.columns)
|
|
port_rets = (w[common] * daily_ret[common]).sum(axis=1)
|
|
rv = port_rets.rolling(60, min_periods=21).std() * np.sqrt(252)
|
|
scale = (self.target_vol / rv).clip(lower=self.min_lev, upper=1.0).shift(1).fillna(1.0)
|
|
w = w.mul(scale, axis=0)
|
|
|
|
# PT on risk-on only
|
|
if self.pt_threshold <= 0: return w
|
|
risk_on_set = {"TQQQ", "UPRO"}
|
|
held = w.idxmax(axis=1); mx = w.max(axis=1); held[mx < 1e-8] = ""
|
|
park = "SHY" if "SHY" in w.columns else ""
|
|
ep, cs, stopped = None, None, False
|
|
rl = self.pt_threshold - self.pt_band
|
|
for i in range(len(w)):
|
|
sym = held.iloc[i]
|
|
if not sym or mx.iloc[i] < 1e-8: cs, ep, stopped = None, None, False; continue
|
|
if sym != cs:
|
|
cs = sym; ep = float(data[sym].iloc[i-1]) if i>0 and sym in data.columns else None; stopped = False; continue
|
|
if sym not in risk_on_set: continue
|
|
if ep is None or ep <= 0 or sym not in data.columns: continue
|
|
y = float(data[sym].iloc[i-1]) if i>0 else float(data[sym].iloc[i])
|
|
g = y/ep - 1.0
|
|
if stopped:
|
|
if g < rl: stopped = False
|
|
else: w.iloc[i] = 0.0; (w.at.__setitem__((w.index[i], park), scale.iloc[i]) if park else None)
|
|
elif g >= self.pt_threshold:
|
|
stopped = True; w.iloc[i] = 0.0; (w.at.__setitem__((w.index[i], park), scale.iloc[i]) if park else None)
|
|
return w
|
|
|
|
|
|
# =========================================================================
|
|
# Main
|
|
# =========================================================================
|
|
|
|
def main():
|
|
print("=" * 110)
|
|
print(" NEW FRAMEWORKS: VOL TRADING + CALENDAR EFFECTS")
|
|
print("=" * 110)
|
|
|
|
all_etfs = sorted(set([
|
|
"SPY", "QQQ", "TQQQ", "UPRO", "GLD", "DBC", "SHY", "TLT",
|
|
"^VIX", "SVXY",
|
|
]))
|
|
data = data_manager.update("etfs", all_etfs, with_open=False)
|
|
if isinstance(data, tuple): data = data[0]
|
|
cutoff = data.index[-1] - pd.DateOffset(years=YEARS)
|
|
data = data[data.index >= cutoff]
|
|
|
|
print(f"Period: {data.index[0].date()} → {data.index[-1].date()}")
|
|
print(f"SVXY: {'yes' if 'SVXY' in data.columns else 'no'}, "
|
|
f"VIX: {'yes' if '^VIX' in data.columns else 'no'}")
|
|
|
|
results = []
|
|
def run(label, strategy):
|
|
try:
|
|
eq = backtest(strategy, data, initial_capital=CAPITAL,
|
|
transaction_cost=TX_COST, fixed_fee=FIXED_FEE)
|
|
m = metrics.raw_summary(eq)
|
|
results.append((label, m))
|
|
print(f" {label:<55} Ann={m['annualizedReturn']*100:>5.1f}% "
|
|
f"Sharpe={m['sharpeRatio']:.2f} MaxDD={m['maxDrawdown']*100:.1f}% "
|
|
f"Calmar={m['calmarRatio']:.2f}")
|
|
except Exception as e:
|
|
print(f" {label:<55} FAILED: {e}")
|
|
|
|
# === Baseline ===
|
|
print("\n--- Baseline ---")
|
|
run("V7+VT36 baseline", TrendRiderV7(target_vol=0.36, min_lev=0.75))
|
|
|
|
# === Framework 2: Vol Trading ===
|
|
print("\n--- Framework 2A: V7 + SVXY risk-off ---")
|
|
for gate in (25, 28, 30, 35):
|
|
run(f"V7+SVXY risk-off (VIX gate={gate})",
|
|
V7SvxyRiskOff(vix_gate=gate, target_vol=0.36, min_lev=0.75))
|
|
|
|
print("\n--- Framework 2B: Pure SVXY strategies ---")
|
|
for gate in (20, 25, 30):
|
|
run(f"Pure SVXY (VIX<{gate} + falling)", PureSvxyStrategy(upper_gate=gate))
|
|
|
|
print("\n--- Framework 2C: TQQQ RSI buy-the-dip ---")
|
|
for buy, sell in [(25, 70), (30, 70), (30, 65), (20, 75)]:
|
|
run(f"TQQQ RSI buy<{buy}/sell>{sell}", TqqqRsiStrategy(buy_level=buy, sell_level=sell))
|
|
|
|
print("\n--- Framework 2D: V7+Vol Hybrid (best of both) ---")
|
|
for gate in (25, 28, 30):
|
|
run(f"V7VolHybrid (VIX gate={gate})",
|
|
V7VolHybrid(vix_gate=gate))
|
|
|
|
# === Framework 4: Calendar Effects ===
|
|
print("\n--- Framework 4A: Turn-of-month TQQQ ---")
|
|
for before, after in [(3, 3), (4, 4), (2, 5)]:
|
|
run(f"TQQQ turn-of-month ({before}d before + {after}d after)",
|
|
TurnOfMonthStrategy(days_before=before, days_after=after))
|
|
|
|
print("\n--- Framework 4B: V7 + turn-of-month filter ---")
|
|
for before, after in [(4, 4), (5, 5), (3, 6)]:
|
|
run(f"V7 risk-on only in TOM window ({before}+{after}d)",
|
|
V7TurnOfMonth(days_before=before, days_after=after,
|
|
target_vol=0.36, min_lev=0.75))
|
|
|
|
# === Final ranking ===
|
|
results.sort(key=lambda x: x[1]["sharpeRatio"], reverse=True)
|
|
print(f"\n{'=' * 115}")
|
|
print(" FINAL RANKING (by Sharpe)")
|
|
print(f"{'=' * 115}")
|
|
print(f"{'#':<4} {'Strategy':<55} {'Ann%':>6} {'Vol%':>6} {'Sharpe':>7} "
|
|
f"{'Sortino':>8} {'MaxDD%':>7} {'Calmar':>7}")
|
|
print("-" * 115)
|
|
for i, (label, m) in enumerate(results, 1):
|
|
marker = " ★" if i <= 3 else ""
|
|
print(f"{i:<4} {label:<55} "
|
|
f"{m['annualizedReturn']*100:>5.1f}% "
|
|
f"{m['annualizedVolatility']*100:>5.1f}% "
|
|
f"{m['sharpeRatio']:>7.2f} {m['sortinoRatio']:>8.2f} "
|
|
f"{m['maxDrawdown']*100:>6.1f}% {m['calmarRatio']:>7.2f}{marker}")
|
|
print(f"{'=' * 115}")
|
|
|
|
results.sort(key=lambda x: x[1]["annualizedReturn"], reverse=True)
|
|
print(f"\n Top 5 by Ann Return:")
|
|
for i, (label, m) in enumerate(results[:5], 1):
|
|
print(f" {i}. {label:<50} Ann={m['annualizedReturn']*100:.1f}% "
|
|
f"Sharpe={m['sharpeRatio']:.2f} MaxDD={m['maxDrawdown']*100:.1f}%")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|