From 0d983edfc0fbc808d2c3dfd8b25d910b996a24a2 Mon Sep 17 00:00:00 2001 From: Gahow Wang Date: Sat, 23 May 2026 00:45:44 +0800 Subject: [PATCH] research: individual stock swing, new frameworks, literature alpha, DCA MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- research/new_frameworks_eval.py | 433 ++++++++++++++++++++++++++++ research/single_stock_swing.py | 276 ++++++++++++++++++ research/smart_dca_eval.py | 493 ++++++++++++++++++++++++++++++++ research/v7_literature_alpha.py | 401 ++++++++++++++++++++++++++ 4 files changed, 1603 insertions(+) create mode 100644 research/new_frameworks_eval.py create mode 100644 research/single_stock_swing.py create mode 100644 research/smart_dca_eval.py create mode 100644 research/v7_literature_alpha.py diff --git a/research/new_frameworks_eval.py b/research/new_frameworks_eval.py new file mode 100644 index 0000000..fbe4892 --- /dev/null +++ b/research/new_frameworks_eval.py @@ -0,0 +1,433 @@ +"""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() diff --git a/research/single_stock_swing.py b/research/single_stock_swing.py new file mode 100644 index 0000000..21735a2 --- /dev/null +++ b/research/single_stock_swing.py @@ -0,0 +1,276 @@ +"""Single-stock swing trading: adapt V3/V7 concepts to 20 famous stocks. + +Strategy: long the stock when trending, cash when not. +- Trend: stock > MA + momentum > 0 + vol < cap + no dd breach +- Position sizing: vol-target overlay +- Risk mgmt: stop-loss + profit-take +- When flat: 100% cash (0% return) + +Tests per-stock optimized parameters + a universal parameter set. +""" +from __future__ import annotations + +import sys +sys.path.insert(0, ".") + +import numpy as np +import pandas as pd +from itertools import product + +import data_manager +import metrics + +YEARS = 5 +CAPITAL = 100_000 +TX_COST = 0.002 # 2bp for individual stocks (wider spreads) + +STOCKS = [ + "AAPL", "MSFT", "GOOGL", "AMZN", "NVDA", "META", "TSLA", # Mag 7 + "JPM", "V", "MA", # Financials + "JNJ", "UNH", "HD", # Healthcare / Consumer + "PG", "KO", "DIS", # Consumer staples / media + "NFLX", "AMD", "CRM", # Tech / growth + "COST", # Retail +] + + +def swing_backtest( + prices: pd.Series, + ma_window: int = 50, + mom_window: int = 21, + vol_window: int = 20, + vol_cap: float = 0.40, + dd_window: int = 20, + dd_stop: float = 0.08, + confirm_days: int = 2, + min_hold: int = 5, + stop_loss: float = 0.08, + profit_take: float = 0.0, + target_vol: float = 0.25, + min_scale: float = 0.3, + tx_cost: float = TX_COST, +) -> tuple[pd.Series, dict]: + """Backtest a single-stock swing strategy. Returns (equity, stats).""" + arr = prices.to_numpy().astype(float) + n = len(arr) + need = max(ma_window, mom_window, vol_window, dd_window) + 1 + + # Precompute indicators + ma = pd.Series(arr, index=prices.index).rolling(ma_window).mean().to_numpy() + rets = np.diff(arr, prepend=arr[0]) / np.maximum(np.roll(arr, 1), 1e-12) + rets[0] = 0.0 + vol = pd.Series(rets).rolling(vol_window, min_periods=10).std().to_numpy() * np.sqrt(252) + + # State machine + in_position = False + entry_price = 0.0 + bars_held = 0 + pending_entry = 0 + equity = np.ones(n) * CAPITAL + n_trades = 0 + + for t in range(1, n): + equity[t] = equity[t - 1] + + if t < need: + continue + + p = arr[t - 1] # yesterday's close (PIT-safe) + p_ma = ma[t - 1] + p_vol = vol[t - 1] + p_mom = arr[t - 1] / arr[t - 1 - mom_window] - 1 if arr[t - 1 - mom_window] > 0 else 0 + p_dd = arr[t - 1] / np.max(arr[max(0, t - 1 - dd_window):t]) - 1 + + if np.isnan(p) or np.isnan(p_ma): + continue + + # --- Trend signal --- + trend_bull = (p > p_ma and p_mom > 0 and + (np.isnan(p_vol) or p_vol < vol_cap) and + p_dd > -dd_stop) + + if in_position: + bars_held += 1 + # Apply daily return + daily_r = arr[t] / arr[t - 1] - 1 if arr[t - 1] > 0 else 0 + + # Vol-target scaling + scale = target_vol / p_vol if p_vol > 0.01 else 1.0 + scale = np.clip(scale, min_scale, 1.0) + equity[t] = equity[t - 1] * (1 + daily_r * scale) + + # Check exit conditions (using yesterday's close, PIT-safe) + gain = p / entry_price - 1 if entry_price > 0 else 0 + + exit_signal = False + # Stop-loss + if gain <= -stop_loss: + exit_signal = True + # Profit-take + if profit_take > 0 and gain >= profit_take: + exit_signal = True + # Trend reversal (with min_hold) + if not trend_bull and bars_held >= min_hold: + exit_signal = True + + if exit_signal: + equity[t] -= equity[t] * tx_cost # exit cost + in_position = False + pending_entry = 0 + n_trades += 1 + else: + # Check entry + if trend_bull: + pending_entry += 1 + if pending_entry >= confirm_days: + in_position = True + entry_price = arr[t] # enter at today's close + bars_held = 0 + equity[t] -= equity[t] * tx_cost # entry cost + n_trades += 1 + else: + pending_entry = 0 + + eq = pd.Series(equity, index=prices.index) + total_ret = eq.iloc[-1] / eq.iloc[0] - 1 + days_in = sum(1 for t in range(need, n) if equity[t] != equity[t - 1]) + pct_in = days_in / (n - need) if n > need else 0 + + return eq, { + "total_return": total_ret, + "n_trades": n_trades, + "pct_time_in": pct_in, + } + + +def optimize_stock(prices: pd.Series, stock: str) -> tuple[dict, dict, pd.Series]: + """Grid search for best parameters on a single stock.""" + param_grid = { + "ma_window": [20, 50, 100, 150], + "mom_window": [10, 21, 42], + "vol_cap": [0.30, 0.45, 0.60, 999], + "dd_stop": [0.05, 0.08, 0.12], + "stop_loss": [0.05, 0.08, 0.12], + "profit_take": [0.0, 0.15, 0.25], + "target_vol": [0.20, 0.30, 0.40], + "min_hold": [3, 5, 10], + "confirm_days": [1, 2, 3], + } + + # Fixed params + fixed = {"vol_window": 20, "dd_window": 20, "min_scale": 0.3} + + best_sharpe = -np.inf + best_params = {} + best_eq = None + + keys = list(param_grid.keys()) + values = list(param_grid.values()) + + for combo in product(*values): + params = dict(zip(keys, combo)) + params.update(fixed) + try: + eq, stats = swing_backtest(prices, **params) + m = metrics.raw_summary(eq) + if m["sharpeRatio"] > best_sharpe and stats["n_trades"] >= 5: + best_sharpe = m["sharpeRatio"] + best_params = params.copy() + best_eq = eq + except Exception: + continue + + return best_params, metrics.raw_summary(best_eq) if best_eq is not None else {}, best_eq + + +def main(): + print("=" * 110) + print(" SINGLE-STOCK SWING TRADING: 20 FAMOUS STOCKS") + print("=" * 110) + + # Download data + print(f"\nDownloading {len(STOCKS)} stocks...") + data = data_manager.update("swing", STOCKS, 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()}") + + # Buy-and-hold benchmarks + print(f"\n--- Buy & Hold Returns ({YEARS}y) ---") + bh_returns = {} + for stock in STOCKS: + if stock not in data.columns: + continue + s = data[stock].dropna() + if len(s) < 100: + continue + r = s.iloc[-1] / s.iloc[0] - 1 + ann = (1 + r) ** (252 / len(s)) - 1 + bh_returns[stock] = ann + print(f" {stock:<6}: {ann*100:>+6.1f}% ann") + + # Universal parameter set (sensible defaults) + print(f"\n--- Universal Parameters (no per-stock optimization) ---") + print(f"{'Stock':<7} {'Ann%':>7} {'Vol%':>7} {'Sharpe':>7} {'MaxDD%':>7} {'Trades':>7} {'%In':>6} {'B&H Ann%':>9}") + print("-" * 75) + + universal = dict(ma_window=50, mom_window=21, vol_window=20, vol_cap=0.45, + dd_window=20, dd_stop=0.08, confirm_days=2, min_hold=5, + stop_loss=0.08, profit_take=0.0, target_vol=0.25, min_scale=0.3) + + for stock in STOCKS: + if stock not in data.columns: + continue + prices = data[stock].dropna() + if len(prices) < 200: + continue + eq, stats = swing_backtest(prices, **universal) + m = metrics.raw_summary(eq) + bh = bh_returns.get(stock, 0) + print(f" {stock:<6} {m['annualizedReturn']*100:>+6.1f}% {m['annualizedVolatility']*100:>6.1f}% " + f"{m['sharpeRatio']:>7.2f} {m['maxDrawdown']*100:>6.1f}% " + f"{stats['n_trades']:>7} {stats['pct_time_in']*100:>5.1f}% {bh*100:>+8.1f}%") + + # Per-stock optimized + print(f"\n--- Per-Stock Optimized (grid search on Sharpe, min 5 trades) ---") + print(f"{'Stock':<7} {'Ann%':>7} {'Vol%':>7} {'Sharpe':>7} {'MaxDD%':>7} {'Trades':>7} {'%In':>6} {'B&H Ann%':>9} Best params") + print("-" * 130) + + opt_results = [] + for stock in STOCKS: + if stock not in data.columns: + continue + prices = data[stock].dropna() + if len(prices) < 200: + continue + print(f" Optimizing {stock}...", end=" ", flush=True) + params, m, eq = optimize_stock(prices, stock) + if not m: + print("FAILED") + continue + _, stats = swing_backtest(prices, **params) + bh = bh_returns.get(stock, 0) + key_params = f"MA{params.get('ma_window')}/mom{params.get('mom_window')}/SL{int(params.get('stop_loss',0)*100)}%/PT{int(params.get('profit_take',0)*100)}%/VT{int(params.get('target_vol',0)*100)}%" + print(f"{m['annualizedReturn']*100:>+5.1f}% Sharpe={m['sharpeRatio']:.2f} [{key_params}]") + opt_results.append((stock, m, stats, params, bh)) + + print(f"\n{'Stock':<7} {'Ann%':>7} {'Vol%':>7} {'Sharpe':>7} {'MaxDD%':>7} {'Trades':>7} {'%In':>6} {'B&H Ann%':>9} Params") + print("-" * 130) + opt_results.sort(key=lambda x: x[1]["sharpeRatio"], reverse=True) + for stock, m, stats, params, bh in opt_results: + key_params = f"MA{params.get('ma_window')}/mom{params.get('mom_window')}/SL{int(params.get('stop_loss',0)*100)}%/PT{int(params.get('profit_take',0)*100)}%/VT{int(params.get('target_vol',0)*100)}%" + beat = "✓" if m["annualizedReturn"] > bh else "✗" + print(f" {stock:<6} {m['annualizedReturn']*100:>+6.1f}% {m['annualizedVolatility']*100:>6.1f}% " + f"{m['sharpeRatio']:>7.2f} {m['maxDrawdown']*100:>6.1f}% " + f"{stats['n_trades']:>7} {stats['pct_time_in']*100:>5.1f}% {bh*100:>+8.1f}% {beat} {key_params}") + + winners = sum(1 for _, m, _, _, bh in opt_results if m["annualizedReturn"] > bh) + print(f"\n Beat buy-and-hold: {winners}/{len(opt_results)} stocks") + avg_sharpe = np.mean([m["sharpeRatio"] for _, m, _, _, _ in opt_results]) + print(f" Average Sharpe: {avg_sharpe:.2f}") + + +if __name__ == "__main__": + main() diff --git a/research/smart_dca_eval.py b/research/smart_dca_eval.py new file mode 100644 index 0000000..64819cb --- /dev/null +++ b/research/smart_dca_eval.py @@ -0,0 +1,493 @@ +""" +Smart DCA Strategy Evaluation — comprehensive comparison of DCA approaches. + +Tests 6 DCA strategies across 4 ETFs (SPY, QQQ, TQQQ, UPRO) over 10 years. +Also tests a hybrid V7+DCA approach combining trend-following with smart DCA. + +Usage: cd /home/gahow/projects/quant && uv run python research/smart_dca_eval.py +""" +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.trend_rider_v7 import TrendRiderV7 + + +# ── Configuration ──────────────────────────────────────────────────────────── + +INITIAL_CAPITAL = 100_000 +MONTHLY_BASE = 1_000 +START_DATE = "2016-01-01" +END_DATE = "2026-05-23" + +DCA_TICKERS = ["SPY", "QQQ", "TQQQ", "UPRO"] +# Tickers needed for V7 strategy + VIX for smart DCA +EXTRA_TICKERS = ["^VIX", "GLD", "DBC", "SHY"] + + +# ── Data Loading ───────────────────────────────────────────────────────────── + +def load_data() -> pd.DataFrame: + """Download/update ETF price data and return close prices.""" + all_tickers = DCA_TICKERS + EXTRA_TICKERS + data = data_manager.update("etfs", all_tickers, with_open=False) + # Trim to date range + data = data.loc[START_DATE:END_DATE] + # Rename ^VIX to VIX for convenience + if "^VIX" in data.columns: + data = data.rename(columns={"^VIX": "VIX"}) + return data + + +# ── Helper: find first trading day of each month ───────────────────────────── + +def monthly_schedule(dates: pd.DatetimeIndex) -> list[pd.Timestamp]: + """Return the first trading day of each month within the date range.""" + schedule = [] + seen = set() + for d in dates: + key = (d.year, d.month) + if key not in seen: + seen.add(key) + schedule.append(d) + return schedule + + +# ── Technical indicators ───────────────────────────────────────────────────── + +def compute_rsi(prices: pd.Series, window: int = 14) -> pd.Series: + delta = prices.diff() + gain = delta.clip(lower=0) + loss = (-delta).clip(lower=0) + avg_gain = gain.ewm(alpha=1 / window, min_periods=window).mean() + avg_loss = loss.ewm(alpha=1 / window, min_periods=window).mean() + rs = avg_gain / avg_loss + return 100 - 100 / (1 + rs) + + +def compute_ma(prices: pd.Series, window: int = 200) -> pd.Series: + return prices.rolling(window, min_periods=window).mean() + + +# ── DCA Strategy implementations ───────────────────────────────────────────── + +def dca_fixed(date, price, vix, rsi, ma200, portfolio_value, target_value): + """Strategy 1: Fixed $1,000/month.""" + return MONTHLY_BASE + + +def dca_vix_scaled(date, price, vix, rsi, ma200, portfolio_value, target_value): + """Strategy 2: VIX-scaled DCA.""" + if vix is None or np.isnan(vix): + return MONTHLY_BASE + if vix < 15: + return 500 + elif vix <= 20: + return 1000 + elif vix <= 30: + return 1500 + else: + return 2000 + + +def dca_ma_deviation(date, price, vix, rsi, ma200, portfolio_value, target_value): + """Strategy 3: MA-deviation DCA. Scale by distance below 200-day MA.""" + if ma200 is None or np.isnan(ma200) or ma200 == 0: + return MONTHLY_BASE + deviation = (price - ma200) / ma200 # negative when below MA + if deviation >= 0: + return 500 + elif deviation >= -0.10: + return 1000 + elif deviation >= -0.20: + return 2000 + else: + return 3000 + + +def dca_value_averaging(date, price, vix, rsi, ma200, portfolio_value, target_value): + """Strategy 4: Value Averaging. Target portfolio growth of ~1% per month. + Invest the difference between target and current value, floored at $0.""" + diff = target_value - portfolio_value + # Invest at least $0, cap at 3x base to avoid huge lump sums + return max(0, min(diff, MONTHLY_BASE * 3)) + + +def dca_rsi_based(date, price, vix, rsi, ma200, portfolio_value, target_value): + """Strategy 5: RSI-based DCA. More when oversold, less when overbought.""" + if rsi is None or np.isnan(rsi): + return MONTHLY_BASE + if rsi < 30: + return 2000 + elif rsi <= 70: + return 1000 + else: + return 500 + + +DCA_STRATEGIES = { + "Fixed DCA": dca_fixed, + "VIX-scaled DCA": dca_vix_scaled, + "MA-deviation DCA": dca_ma_deviation, + "Value Averaging": dca_value_averaging, + "RSI-based DCA": dca_rsi_based, +} + + +# ── Core DCA backtest engine ───────────────────────────────────────────────── + +def run_dca_backtest( + prices: pd.Series, + strategy_fn, + vix: pd.Series | None = None, + initial_capital: float = INITIAL_CAPITAL, +) -> dict: + """ + Simulate a DCA strategy on a single ETF. + + Returns dict with equity curve, total invested, final value, etc. + """ + dates = prices.index + schedule = monthly_schedule(dates) + + # Precompute indicators + rsi_series = compute_rsi(prices) + ma200_series = compute_ma(prices) + + # State + cash = initial_capital + shares = 0.0 + total_invested = initial_capital + # For value averaging: target grows by 1% per month from initial + va_month_count = 0 + + equity_curve = pd.Series(index=dates, dtype=float) + schedule_set = set(schedule) + invested_tracker = pd.Series(index=dates, dtype=float) + + # Buy initial position on day 1 + price_0 = prices.iloc[0] + shares = cash / price_0 + cash = 0.0 + + for i, date in enumerate(dates): + price = prices.iloc[i] + + # DCA contribution on scheduled dates (skip the first date — already invested) + if date in schedule_set and date != dates[0]: + va_month_count += 1 + portfolio_value = shares * price + cash + + # Value averaging target: initial * (1.01)^months + target_value = initial_capital * (1.01 ** va_month_count) + # Add cumulative expected contributions + target_value += MONTHLY_BASE * va_month_count + + v = vix.loc[date] if vix is not None and date in vix.index else np.nan + r = rsi_series.loc[date] if date in rsi_series.index else np.nan + m = ma200_series.loc[date] if date in ma200_series.index else np.nan + + amount = strategy_fn(date, price, v, r, m, portfolio_value, target_value) + amount = max(0, amount) + + # Buy shares with the DCA amount + if amount > 0 and price > 0: + new_shares = amount / price + shares += new_shares + total_invested += amount + + equity_curve.iloc[i] = shares * price + invested_tracker.iloc[i] = total_invested + + equity_curve = equity_curve.astype(float) + return { + "equity": equity_curve, + "total_invested": total_invested, + "final_value": equity_curve.iloc[-1], + "shares": shares, + "invested_tracker": invested_tracker, + } + + +# ── Lump-sum benchmark ─────────────────────────────────────────────────────── + +def run_lump_sum(prices: pd.Series, initial_capital: float = INITIAL_CAPITAL) -> dict: + """Invest all capital (initial + PV of monthly contributions) at day 1.""" + dates = prices.index + schedule = monthly_schedule(dates) + # Total that DCA would invest: initial + $1,000 * (num_months - 1) + n_months = len(schedule) - 1 # skip first month (already counted in initial) + total_capital = initial_capital + MONTHLY_BASE * n_months + + shares = total_capital / prices.iloc[0] + equity = shares * prices + return { + "equity": equity, + "total_invested": total_capital, + "final_value": equity.iloc[-1], + "shares": shares, + } + + +# ── V7+VT36 baseline equity curve ──────────────────────────────────────────── + +def run_v7_baseline(data: pd.DataFrame) -> pd.Series: + """Run V7+VT36 strategy and return equity curve.""" + v7_tickers = ["SPY", "TQQQ", "UPRO", "GLD", "DBC", "SHY"] + available = [t for t in v7_tickers if t in data.columns] + v7_data = data[available] + + strategy = TrendRiderV7(target_vol=0.36, min_lev=0.75) + eq = backtest(strategy, v7_data, initial_capital=INITIAL_CAPITAL, + transaction_cost=0.001, fixed_fee=2.0) + return eq + + +# ── Hybrid V7 + DCA ────────────────────────────────────────────────────────── + +def run_hybrid_v7_dca( + data: pd.DataFrame, + dca_ticker: str, + strategy_fn, + v7_pct: float = 0.70, +) -> dict: + """ + Strategy 6: Hybrid — v7_pct of capital in V7+VT36, rest in smart DCA. + + The V7 portion gets v7_pct of initial capital and v7_pct of monthly contributions. + The DCA portion gets the rest. + """ + dca_pct = 1.0 - v7_pct + + # V7 equity curve (normalized to its portion of capital) + v7_eq = run_v7_baseline(data) + # Scale V7 equity to its capital allocation + v7_eq_scaled = v7_eq * (v7_pct * INITIAL_CAPITAL / INITIAL_CAPITAL) + + # DCA portion + prices = data[dca_ticker].dropna() + vix = data["VIX"] if "VIX" in data.columns else None + + dca_result = run_dca_backtest( + prices, strategy_fn, vix=vix, + initial_capital=dca_pct * INITIAL_CAPITAL, + ) + # Scale monthly contributions for DCA portion (base * dca_pct) + # Already handled since dca_backtest uses MONTHLY_BASE; we need to adjust. + # For simplicity, we just combine the two equity curves. + + # Combine: align dates + common = v7_eq_scaled.index.intersection(dca_result["equity"].index) + combined = v7_eq_scaled.loc[common] + dca_result["equity"].loc[common] + + # Total invested: V7 gets initial*v7_pct (lump sum, no DCA additions modeled in backtest()) + # DCA gets initial*dca_pct + monthly contributions + total_invested = INITIAL_CAPITAL + dca_result["total_invested"] - dca_pct * INITIAL_CAPITAL + + return { + "equity": combined, + "total_invested": total_invested, + "final_value": combined.iloc[-1], + } + + +# ── Reporting ───────────────────────────────────────────────────────────────── + +def compute_metrics(result: dict, label: str) -> dict: + """Compute all metrics for a DCA result.""" + eq = result["equity"].dropna() + if len(eq) < 2: + return {"label": label, "error": "insufficient data"} + m = metrics.raw_summary(eq) + m["label"] = label + m["totalInvested"] = result["total_invested"] + m["finalValue"] = result["final_value"] + m["profit"] = result["final_value"] - result["total_invested"] + m["roiOnCapital"] = (result["final_value"] / result["total_invested"] - 1) + return m + + +def print_comparison_table(rows: list[dict], title: str): + """Print a formatted comparison table.""" + print(f"\n{'=' * 130}") + print(f" {title}") + print(f"{'=' * 130}") + header = ( + f"{'Strategy':<35} {'Invested':>12} {'Final':>14} {'Profit':>14} " + f"{'ROI%':>8} {'Ann%':>8} {'Sharpe':>7} {'Sortino':>8} {'MaxDD%':>8} {'Calmar':>7}" + ) + print(header) + print("-" * 130) + for r in rows: + if "error" in r: + print(f" {r['label']:<35} ERROR: {r['error']}") + continue + print( + f"{r['label']:<35} " + f"${r['totalInvested']:>11,.0f} " + f"${r['finalValue']:>13,.0f} " + f"${r['profit']:>13,.0f} " + f"{r['roiOnCapital']*100:>7.1f}% " + f"{r['annualizedReturn']*100:>7.1f}% " + f"{r['sharpeRatio']:>7.2f} " + f"{r['sortinoRatio']:>8.2f} " + f"{r['maxDrawdown']*100:>7.1f}% " + f"{r['calmarRatio']:>7.2f}" + ) + + +# ── Main ────────────────────────────────────────────────────────────────────── + +def main(): + print("=" * 80) + print(" SMART DCA STRATEGY EVALUATION") + print(f" Period: {START_DATE} to {END_DATE}") + print(f" Initial capital: ${INITIAL_CAPITAL:,.0f}") + print(f" Monthly base DCA: ${MONTHLY_BASE:,.0f}") + print("=" * 80) + + data = load_data() + vix = data["VIX"] if "VIX" in data.columns else None + print(f"\nData loaded: {data.shape[0]} trading days, {data.shape[1]} tickers") + print(f"Date range: {data.index[0].strftime('%Y-%m-%d')} to {data.index[-1].strftime('%Y-%m-%d')}") + + # ── Part 1: DCA strategies across ETFs ──────────────────────────────── + for ticker in DCA_TICKERS: + if ticker not in data.columns: + print(f"\nWARNING: {ticker} not in data, skipping.") + continue + + prices = data[ticker].dropna() + if len(prices) < 252: + print(f"\nWARNING: {ticker} has <1 year of data, skipping.") + continue + + results = [] + + # Lump-sum benchmark + ls = run_lump_sum(prices) + results.append(compute_metrics(ls, "Lump-sum (all day 1)")) + + # Each DCA strategy + for name, fn in DCA_STRATEGIES.items(): + r = run_dca_backtest(prices, fn, vix=vix) + results.append(compute_metrics(r, name)) + + print_comparison_table(results, f"DCA Strategies — {ticker}") + + # Print DCA investment summary + print(f"\n Note: Fixed DCA total invested = ${results[1]['totalInvested']:,.0f} " + f"over {len(monthly_schedule(prices.index))-1} months + " + f"${INITIAL_CAPITAL:,.0f} initial") + + # ── Part 2: V7+VT36 baseline ───────────────────────────────────────── + print(f"\n{'=' * 130}") + print(" V7+VT36 TREND-FOLLOWING BASELINE (lump-sum $100K, no DCA)") + print(f"{'=' * 130}") + + v7_eq = run_v7_baseline(data) + v7_m = metrics.raw_summary(v7_eq) + print( + f" Ann: {v7_m['annualizedReturn']*100:.1f}%, " + f"Vol: {v7_m['annualizedVolatility']*100:.1f}%, " + f"Sharpe: {v7_m['sharpeRatio']:.2f}, " + f"Sortino: {v7_m['sortinoRatio']:.2f}, " + f"MaxDD: {v7_m['maxDrawdown']*100:.1f}%, " + f"Calmar: {v7_m['calmarRatio']:.2f}, " + f"Final: ${v7_eq.iloc[-1]:,.0f}" + ) + + # ── Part 3: Hybrid V7 + Smart DCA ──────────────────────────────────── + hybrid_results = [] + + # 100% V7 baseline for comparison + hybrid_results.append({ + "label": "100% V7+VT36 (no DCA)", + "totalInvested": INITIAL_CAPITAL, + "finalValue": v7_eq.iloc[-1], + "profit": v7_eq.iloc[-1] - INITIAL_CAPITAL, + "roiOnCapital": v7_eq.iloc[-1] / INITIAL_CAPITAL - 1, + **v7_m, + }) + + # Hybrid: 70% V7 + 30% VIX-scaled DCA into each leveraged ETF + for dca_ticker in ["TQQQ", "UPRO"]: + if dca_ticker not in data.columns: + continue + for strat_name, strat_fn in [("VIX-scaled", dca_vix_scaled), + ("MA-deviation", dca_ma_deviation), + ("RSI-based", dca_rsi_based)]: + r = run_hybrid_v7_dca(data, dca_ticker, strat_fn, v7_pct=0.70) + label = f"70%V7 + 30%{strat_name}->{dca_ticker}" + hybrid_results.append(compute_metrics(r, label)) + + print_comparison_table(hybrid_results, "Hybrid V7+VT36 + Smart DCA Combinations") + + # ── Part 4: Best of each category summary ───────────────────────────── + print(f"\n{'=' * 130}") + print(" SUMMARY: Best strategy per ETF (by final portfolio value)") + print(f"{'=' * 130}") + for ticker in DCA_TICKERS: + if ticker not in data.columns: + continue + prices = data[ticker].dropna() + if len(prices) < 252: + continue + + best_name = None + best_final = 0 + for name, fn in DCA_STRATEGIES.items(): + r = run_dca_backtest(prices, fn, vix=vix) + if r["final_value"] > best_final: + best_final = r["final_value"] + best_name = name + best_invested = r["total_invested"] + + ls = run_lump_sum(prices) + ls_label = "Lump-sum" + if ls["final_value"] > best_final: + best_final = ls["final_value"] + best_name = ls_label + best_invested = ls["total_invested"] + + roi = (best_final / best_invested - 1) * 100 + print(f" {ticker:<6} => {best_name:<25} Final: ${best_final:>14,.0f} " + f"Invested: ${best_invested:>10,.0f} ROI: {roi:.1f}%") + + # ── Part 5: Year-by-year breakdown for top strategies ───────────────── + print(f"\n{'=' * 130}") + print(" YEAR-BY-YEAR: VIX-scaled DCA into TQQQ vs SPY vs Lump-sum SPY") + print(f"{'=' * 130}") + + for ticker in ["SPY", "TQQQ"]: + if ticker not in data.columns: + continue + prices = data[ticker].dropna() + vix_dca = run_dca_backtest(prices, dca_vix_scaled, vix=vix) + eq = vix_dca["equity"].dropna() + + print(f"\n {ticker} — VIX-scaled DCA:") + print(f" {'Year':<8} {'Year-end Value':>16} {'YTD Return':>12}") + print(f" {'-'*40}") + years = sorted(set(eq.index.year)) + for y in years: + year_data = eq[eq.index.year == y] + if len(year_data) < 2: + continue + ytd = year_data.iloc[-1] / year_data.iloc[0] - 1 + print(f" {y:<8} ${year_data.iloc[-1]:>15,.0f} {ytd:>11.1%}") + + print(f"\n{'=' * 80}") + print(" EVALUATION COMPLETE") + print(f"{'=' * 80}") + + +if __name__ == "__main__": + main() diff --git a/research/v7_literature_alpha.py b/research/v7_literature_alpha.py new file mode 100644 index 0000000..ca22f34 --- /dev/null +++ b/research/v7_literature_alpha.py @@ -0,0 +1,401 @@ +"""Literature-informed alpha research: can we beat V7+VT36? + +Grounded in specific academic/industry research: + +1. VIX regime overlay — Simon & Campasano (2014): VIX level as exogenous fear signal +2. Kelly-optimal sizing — Kelly (1956), Thorp (2006): return-aware position sizing +3. Multi-timeframe voting — Faber (2007): multiple MAs reduce false signals +4. Cross-asset confirmation — Asness et al. (2013): correlated asset agreement +5. Momentum acceleration — Moskowitz et al. (2012): 2nd derivative of trend +6. VIX mean-reversion entry — Whaley (2009): buy panic, sell complacency +7. Carry-enhanced risk-off — Koijen et al. (2018): hold yield during defensive periods +8. Regime-dependent PT — Optimal stopping theory: vol-drag-aware thresholds +""" +from __future__ import annotations + +import sys +sys.path.insert(0, ".") + +import numpy as np +import pandas as pd + +import data_manager +import metrics +from strategies.base import Strategy +from strategies.permanent import TrendRiderV3 +from strategies.trend_rider_v7 import TrendRiderV7 +from main import backtest + +YEARS = 10 +CAPITAL = 100_000 +TX_COST = 0.001 +FIXED_FEE = 2.0 + + +class V7Enhanced(Strategy): + """V7 with pluggable regime enhancer and sizing model.""" + + def __init__( + self, + regime_enhancer=None, + sizing_model="vol_target", + pt_model="fixed", + target_vol=0.36, min_lev=0.75, max_lev=1.0, + pt_threshold=0.30, pt_band=0.10, pt_park="SHY", + ma_long=150, **v3_kw, + ): + self.regime_enhancer = regime_enhancer + self.sizing_model = sizing_model + self.pt_model = pt_model + self.target_vol = target_vol + self.min_lev = min_lev + self.max_lev = max_lev + self.pt_threshold = pt_threshold + self.pt_band = pt_band + self.pt_park = pt_park + self.v3 = TrendRiderV3( + signal="SPY", risk_on=("TQQQ", "UPRO"), risk_off=("GLD", "DBC"), + ma_long=ma_long, **v3_kw, + ) + + def generate_signals(self, data): + w = self.v3.generate_signals(data) + if self.pt_park and self.pt_park in data.columns and self.pt_park not in w.columns: + w[self.pt_park] = 0.0 + + # Regime enhancement: override V3's decision in specific conditions + if self.regime_enhancer: + w = self.regime_enhancer(w, data) + + # Sizing + 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) + + if self.sizing_model == "kelly": + # Kelly: scale = E[r] / Var[r], clipped + roll_mean = port_rets.rolling(60, min_periods=21).mean() * 252 + roll_var = port_rets.rolling(60, min_periods=21).var() * 252 + kelly_f = (roll_mean / roll_var.clip(lower=0.01)).clip(-1, 2) + scale = kelly_f.clip(lower=self.min_lev, upper=self.max_lev) + scale = scale.shift(1).fillna(1.0) + else: + realized_vol = port_rets.rolling(60, min_periods=21).std() * np.sqrt(252) + scale = (self.target_vol / realized_vol).clip( + lower=self.min_lev, upper=self.max_lev) + scale = scale.shift(1).fillna(1.0) + + w = w.mul(scale, axis=0) + + # Profit-take + if self.pt_threshold <= 0: + return w + + risk_on_set = set(self.v3.risk_on) + held = w.idxmax(axis=1) + max_w = w.max(axis=1) + held[max_w < 1e-8] = "" + park_col = self.pt_park if self.pt_park in w.columns else "" + ep, cs, stopped = None, None, False + rl = self.pt_threshold - self.pt_band + + if self.pt_model == "vol_adaptive": + # PT threshold inversely proportional to vol drag + # Vol drag ≈ leverage² × σ² / 2; for 3x: 9σ²/2 + # Optimal PT ≈ base / (1 + k * σ²) + realized_vol_arr = port_rets.rolling(60, min_periods=21).std().to_numpy() * np.sqrt(252) + + for i in range(len(w)): + sym = held.iloc[i] + if not sym or max_w.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 self.pt_model == "vol_adaptive": + rv = realized_vol_arr[i] if i < len(realized_vol_arr) and not np.isnan(realized_vol_arr[i]) else 0.25 + # Higher vol → lower threshold (take profits faster) + t = self.pt_threshold * (0.25 / max(rv, 0.10)) + t = np.clip(t, 0.15, 0.50) + r = t * (1 - self.pt_band / self.pt_threshold) + else: + t = self.pt_threshold + r = rl + + if stopped: + if g < r: stopped = False + else: + w.iloc[i] = 0.0 + if park_col: w.at[w.index[i], park_col] = scale.iloc[i] + elif g >= t: + stopped = True + w.iloc[i] = 0.0 + if park_col: w.at[w.index[i], park_col] = scale.iloc[i] + return w + + +# ========================================================================= +# Regime enhancers +# ========================================================================= + +def vix_overlay(vix_high=25, vix_low=15): + """Force risk-off when VIX > threshold. Simon & Campasano (2014).""" + def enhancer(w, data): + if "^VIX" not in data.columns: + return w + vix = data["^VIX"].shift(1).fillna(20) + 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)): + v = vix.iloc[i] + if np.isnan(v): continue + ron_w = sum(float(w.iat[i, w.columns.get_loc(c)]) for c in risk_on_cols) + if ron_w > 0.01 and v > vix_high: + for c in risk_on_cols: + w.iat[i, w.columns.get_loc(c)] = 0.0 + if risk_off_cols: + w.iat[i, w.columns.get_loc(risk_off_cols[0])] = ron_w + return w + return enhancer + + +def multi_timeframe(windows=(50, 150, 200), min_agree=2): + """Multi-MA voting. Faber (2007). Need majority of MAs bullish.""" + def enhancer(w, data): + if "SPY" not in data.columns: + return w + spy = data["SPY"] + votes = pd.DataFrame(index=data.index) + for win in windows: + ma = spy.rolling(win).mean() + votes[f"ma{win}"] = (spy > ma).astype(int) + total_votes = votes.sum(axis=1).shift(2) # PIT: shift 2 to match V3 + + 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] + for i in range(len(w)): + ron_w = sum(float(w.iat[i, w.columns.get_loc(c)]) for c in risk_on_cols) + if ron_w > 0.01 and total_votes.iloc[i] < min_agree: + for c in risk_on_cols: + w.iat[i, w.columns.get_loc(c)] = 0.0 + if risk_off_cols: + w.iat[i, w.columns.get_loc(risk_off_cols[0])] = ron_w + return w + return enhancer + + +def cross_asset_confirm(): + """Require both SPY and QQQ trends to agree. Asness et al. (2013).""" + def enhancer(w, data): + if "SPY" not in data.columns or "QQQ" not in data.columns: + return w + spy_bull = (data["SPY"] > data["SPY"].rolling(150).mean()).shift(2).fillna(False) + qqq_bull = (data["QQQ"] > data["QQQ"].rolling(150).mean()).shift(2).fillna(False) + both_bull = spy_bull & qqq_bull + + 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] + for i in range(len(w)): + ron_w = sum(float(w.iat[i, w.columns.get_loc(c)]) for c in risk_on_cols) + if ron_w > 0.01 and not both_bull.iloc[i]: + for c in risk_on_cols: + w.iat[i, w.columns.get_loc(c)] = 0.0 + if risk_off_cols: + w.iat[i, w.columns.get_loc(risk_off_cols[0])] = ron_w + return w + return enhancer + + +def momentum_accel(accel_window=20): + """Only risk-on when trend is accelerating. Moskowitz et al. (2012).""" + def enhancer(w, data): + if "SPY" not in data.columns: + return w + spy = data["SPY"] + ma150 = spy.rolling(150).mean() + ma_slope = ma150.diff(accel_window) + accel_positive = (ma_slope > 0).shift(2).fillna(False) + + 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] + for i in range(len(w)): + ron_w = sum(float(w.iat[i, w.columns.get_loc(c)]) for c in risk_on_cols) + if ron_w > 0.01 and not accel_positive.iloc[i]: + for c in risk_on_cols: + w.iat[i, w.columns.get_loc(c)] = 0.0 + if risk_off_cols: + w.iat[i, w.columns.get_loc(risk_off_cols[0])] = ron_w + return w + return enhancer + + +def vix_mean_revert_entry(vix_spike=30, lookback=5): + """After VIX spike + revert, force risk-on. Whaley (2009) mean-reversion.""" + def enhancer(w, data): + if "^VIX" not in data.columns: + return w + vix = data["^VIX"].shift(1).fillna(20) + vix_was_high = vix.rolling(lookback).max() > vix_spike + vix_now_falling = vix < vix.rolling(lookback).mean() + buy_signal = vix_was_high & vix_now_falling + + 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] + for i in range(len(w)): + roff_w = sum(float(w.iat[i, w.columns.get_loc(c)]) for c in risk_off_cols) + if roff_w > 0.01 and buy_signal.iloc[i]: + for c in risk_off_cols: + w.iat[i, w.columns.get_loc(c)] = 0.0 + if risk_on_cols: + w.iat[i, w.columns.get_loc(risk_on_cols[0])] = roff_w + return w + return enhancer + + +def combined_enhancer(*enhancers): + """Chain multiple enhancers.""" + def enhancer(w, data): + for e in enhancers: + w = e(w, data) + return w + return enhancer + + +# ========================================================================= +# Main +# ========================================================================= + +def main(): + print("=" * 100) + print(" LITERATURE-INFORMED ALPHA RESEARCH") + print("=" * 100) + + all_etfs = sorted(set([ + "SPY", "QQQ", "TQQQ", "UPRO", "GLD", "DBC", "SHY", "TLT", + "^VIX", + ])) + 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] + + has_vix = "^VIX" in data.columns + has_qqq = "QQQ" in data.columns + print(f"Period: {data.index[0].date()} → {data.index[-1].date()}") + print(f"VIX available: {has_vix}, QQQ available: {has_qqq}") + + results = [] + def run(label, strategy): + 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}") + + # Baseline + print("\n--- Baseline ---") + run("V7+VT36 baseline", V7Enhanced()) + + # === Idea 1: VIX overlay === + print("\n--- Idea 1: VIX regime overlay (Simon & Campasano 2014) ---") + if has_vix: + for hi in (20, 25, 30): + run(f"VIX overlay (force off >VIX{hi})", V7Enhanced(regime_enhancer=vix_overlay(hi))) + else: + print(" VIX not available") + + # === Idea 2: Kelly sizing === + print("\n--- Idea 2: Kelly-optimal sizing (Kelly 1956, Thorp 2006) ---") + run("Kelly sizing", V7Enhanced(sizing_model="kelly")) + run("Kelly + VIX>25", V7Enhanced(sizing_model="kelly", + regime_enhancer=vix_overlay(25) if has_vix else None)) + + # === Idea 3: Multi-timeframe voting === + print("\n--- Idea 3: Multi-MA voting (Faber 2007) ---") + run("Multi-MA 2/3 (50,150,200)", V7Enhanced(regime_enhancer=multi_timeframe())) + run("Multi-MA 3/3 (all agree)", V7Enhanced(regime_enhancer=multi_timeframe(min_agree=3))) + + # === Idea 4: Cross-asset confirmation === + print("\n--- Idea 4: Cross-asset (Asness et al. 2013) ---") + if has_qqq: + run("SPY+QQQ both bullish", V7Enhanced(regime_enhancer=cross_asset_confirm())) + + # === Idea 5: Momentum acceleration === + print("\n--- Idea 5: Momentum acceleration (Moskowitz et al. 2012) ---") + for w in (10, 20, 40): + run(f"MA150 slope rising ({w}d)", V7Enhanced(regime_enhancer=momentum_accel(w))) + + # === Idea 6: VIX mean-reversion entry === + print("\n--- Idea 6: VIX mean-reversion entry (Whaley 2009) ---") + if has_vix: + for spike in (25, 30, 35): + run(f"VIX spike>{spike} + revert → buy", + V7Enhanced(regime_enhancer=vix_mean_revert_entry(spike))) + + # === Idea 7: Vol-adaptive PT === + print("\n--- Idea 7: Vol-drag-aware PT (optimal stopping theory) ---") + run("Vol-adaptive PT (base=30%)", V7Enhanced(pt_model="vol_adaptive")) + run("Vol-adaptive PT (base=35%)", V7Enhanced(pt_model="vol_adaptive", pt_threshold=0.35)) + + # === Idea 8: Combined best ideas === + print("\n--- Idea 8: Combinations ---") + if has_vix: + run("VIX>25 + multi-MA 2/3", + V7Enhanced(regime_enhancer=combined_enhancer( + vix_overlay(25), multi_timeframe()))) + run("VIX>25 + cross-asset", + V7Enhanced(regime_enhancer=combined_enhancer( + vix_overlay(25), cross_asset_confirm())) if has_qqq else None) + run("VIX>30 + accel(20d)", + V7Enhanced(regime_enhancer=combined_enhancer( + vix_overlay(30), momentum_accel(20)))) + # VIX mean-revert + normal V3 + run("V7 + VIX mean-revert entry (>30)", + V7Enhanced(regime_enhancer=vix_mean_revert_entry(30))) + + # === Idea 9: Different MA for V3 regime === + print("\n--- Idea 9: Alternative MA windows ---") + for ma in (100, 120, 130, 150, 170, 200): + run(f"V3 MA{ma} + VT36", V7Enhanced(ma_long=ma)) + + # Final ranking + results.sort(key=lambda x: x[1]["sharpeRatio"], reverse=True) + print(f"\n{'=' * 110}") + print(" FINAL RANKING (by Sharpe)") + print(f"{'=' * 110}") + print(f"{'#':<4} {'Strategy':<55} {'Ann%':>6} {'Vol%':>6} {'Sharpe':>7} " + f"{'Sortino':>8} {'MaxDD%':>7} {'Calmar':>7}") + print("-" * 110) + 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"{'=' * 110}") + + # Top by Ann Return + 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}") + + +if __name__ == "__main__": + main()