diff --git a/strategies/adaptive_blend.py b/strategies/adaptive_blend.py new file mode 100644 index 0000000..07a09b5 --- /dev/null +++ b/strategies/adaptive_blend.py @@ -0,0 +1,190 @@ +""" +PIT-Compliant Adaptive Multi-Strategy Blend — Sharpe 1.5+ target. + +Combines three uncorrelated alpha sources with adaptive (trailing-Sharpe) +weighting and a vol-target overlay: + + Leg 1: TrendRiderV5 (ETF regime/leverage timing) — inherently PIT + Leg 2: PIT Stock Momentum (cross-sectional 12-1 mom, masked universe) + Leg 3: Cross-Asset Time-Series Momentum (6m, top 2, inv-vol) — inherently PIT + +Blending: weights proportional to trailing 126d rolling Sharpe (clipped ≥ 0). +Vol-target: scale combined exposure so realized 20d vol ≈ 22%. + +Backtest (2017-06 to 2026-05, PIT-compliant): + Full: Sharpe 1.69, CAGR 33.6%, MaxDD -20.6%, Calmar 1.63 + IS: Sharpe 1.36 (2017-2022) + OOS: Sharpe 2.21 (2023-2026) + Bootstrap P(Sharpe > 1.5) = 72% + +All components are PIT-safe: + - V5: ETF-only (SPY/UPRO/TQQQ/GLD/DBC) + - Stock: uses universe_history.mask_prices() for S&P 500 membership + - XA: ETF-only (SPY/GLD/TLT/IEF/DBC) + - Adaptive weights: trailing 126d (no future info) + - Vol-target: 1-day lagged realized vol +""" +from __future__ import annotations + +import numpy as np +import pandas as pd + +from strategies.cross_asset_momentum import CrossAssetMomentum +from strategies.trend_rider_v5 import TrendRiderV5 + + +class AdaptiveBlend: + """ + Adaptive multi-strategy blend with vol-target overlay. + + Usage: + blend = AdaptiveBlend() + daily_returns = blend.run(etf_panel, pit_masked_prices) + """ + + def __init__( + self, + # V5 params + v5_promote_thresholds=(0.35, 0.55), + v5_demote_thresholds=(0.25, 0.45), + # Stock params + stock_top_n: int = 12, + stock_rebal_freq: int = 42, + stock_mom_blend: float = 1.0, + stock_asym_vol: bool = True, + stock_asym_vol_floor: float = 0.50, + # XA params + xa_lookback: int = 126, + xa_top_k: int = 2, + xa_rebal_freq: int = 21, + # Blend params + adaptive_lookback: int = 126, + vol_target: float = 0.22, + vol_window: int = 20, + # Backtest range + start: str = "2017-06-01", + ): + self.v5 = TrendRiderV5( + promote_thresholds=v5_promote_thresholds, + demote_thresholds=v5_demote_thresholds, + ) + self.stock_params = dict( + top_n=stock_top_n, rebal_freq=stock_rebal_freq, + mom_blend=stock_mom_blend, asym_vol=stock_asym_vol, + asym_vol_floor=stock_asym_vol_floor, dd_dampen=False, + ) + self.xa = CrossAssetMomentum( + lookback=xa_lookback, top_k=xa_top_k, + rebal_freq=xa_rebal_freq, vol_scale=True, + ) + self.adaptive_lookback = adaptive_lookback + self.vol_target = vol_target + self.vol_window = vol_window + self.start = start + + def run( + self, + etf_panel: pd.DataFrame, + pit_masked: pd.DataFrame, + transaction_cost: float = 0.001, + ) -> pd.Series: + """ + Execute the full blend and return daily returns. + + Parameters + ---------- + etf_panel : price panel including ETFs (SPY, UPRO, TQQQ, GLD, DBC, TLT, IEF, etc.) + pit_masked : S&P 500 prices masked by membership (NaN outside membership) + transaction_cost : one-way transaction cost (default 10bps) + + Returns + ------- + pd.Series of daily returns (after vol-target scaling) + """ + from research.pit_optimization import PITEnsemble + from research.trend_rider_robustness import portfolio_returns + + # --- Leg 1: V5 --- + w_v5 = self.v5.generate_signals(etf_panel) + rets_v5 = portfolio_returns(w_v5, etf_panel, transaction_cost=transaction_cost) + rets_v5 = rets_v5.loc[self.start:] + + # --- Leg 2: PIT Stock Momentum --- + strat_stock = PITEnsemble(**self.stock_params) + w_stock = strat_stock.generate_signals(pit_masked) + rets_stock = (w_stock * pit_masked.pct_change(fill_method=None).fillna(0.0)).sum(axis=1) + rets_stock = rets_stock.loc[self.start:] + + # --- Leg 3: Cross-Asset Momentum --- + w_xa = self.xa.generate_signals(etf_panel) + rets_xa = portfolio_returns(w_xa, etf_panel, transaction_cost=transaction_cost) + rets_xa = rets_xa.loc[self.start:] + + # --- Adaptive blending --- + idx = rets_v5.index.intersection(rets_stock.index).intersection(rets_xa.index) + df = pd.DataFrame({ + "v5": rets_v5.loc[idx], + "stock": rets_stock.loc[idx], + "xa": rets_xa.loc[idx], + }).fillna(0.0) + + roll_mu = df.rolling(self.adaptive_lookback).mean() + roll_std = df.rolling(self.adaptive_lookback).std() + roll_sharpe = (roll_mu / roll_std * np.sqrt(252)).clip(lower=0) + w_sum = roll_sharpe.sum(axis=1).replace(0, 1) + adaptive_weights = roll_sharpe.div(w_sum, axis=0).fillna(1.0 / 3) + + combined_rets = (df * adaptive_weights).sum(axis=1) + + # --- Vol-target overlay --- + realized = combined_rets.rolling(self.vol_window).std(ddof=1) * np.sqrt(252) + realized = realized.shift(1).fillna(self.vol_target) + scale = (self.vol_target / realized.replace(0.0, np.nan)).clip(upper=1.0).fillna(1.0) + final_rets = combined_rets * scale + + return final_rets + + def run_with_diagnostics( + self, + etf_panel: pd.DataFrame, + pit_masked: pd.DataFrame, + transaction_cost: float = 0.001, + ) -> dict: + """Run and return diagnostics (individual rets, weights, combined).""" + from research.pit_optimization import PITEnsemble + from research.trend_rider_robustness import portfolio_returns + + w_v5 = self.v5.generate_signals(etf_panel) + rets_v5 = portfolio_returns(w_v5, etf_panel, transaction_cost=transaction_cost).loc[self.start:] + + strat_stock = PITEnsemble(**self.stock_params) + w_stock = strat_stock.generate_signals(pit_masked) + rets_stock = (w_stock * pit_masked.pct_change(fill_method=None).fillna(0.0)).sum(axis=1).loc[self.start:] + + w_xa = self.xa.generate_signals(etf_panel) + rets_xa = portfolio_returns(w_xa, etf_panel, transaction_cost=transaction_cost).loc[self.start:] + + idx = rets_v5.index.intersection(rets_stock.index).intersection(rets_xa.index) + df = pd.DataFrame({"v5": rets_v5.loc[idx], "stock": rets_stock.loc[idx], "xa": rets_xa.loc[idx]}).fillna(0) + + roll_mu = df.rolling(self.adaptive_lookback).mean() + roll_std = df.rolling(self.adaptive_lookback).std() + roll_sharpe = (roll_mu / roll_std * np.sqrt(252)).clip(lower=0) + w_sum = roll_sharpe.sum(axis=1).replace(0, 1) + adaptive_weights = roll_sharpe.div(w_sum, axis=0).fillna(1.0 / 3) + + combined_rets = (df * adaptive_weights).sum(axis=1) + realized = combined_rets.rolling(self.vol_window).std(ddof=1) * np.sqrt(252) + realized = realized.shift(1).fillna(self.vol_target) + scale = (self.vol_target / realized.replace(0.0, np.nan)).clip(upper=1.0).fillna(1.0) + final_rets = combined_rets * scale + + return { + "rets_v5": rets_v5, + "rets_stock": rets_stock, + "rets_xa": rets_xa, + "adaptive_weights": adaptive_weights, + "combined_rets": combined_rets, + "vol_scale": scale, + "final_rets": final_rets, + } diff --git a/strategies/composite_alpha.py b/strategies/composite_alpha.py new file mode 100644 index 0000000..f59db6d --- /dev/null +++ b/strategies/composite_alpha.py @@ -0,0 +1,133 @@ +""" +Composite Alpha Strategy. + +Combines the strongest alpha factors discovered in research: +1. Recovery (63d) - strongest single IC +2. Intermediate momentum (7m) - strong trend signal +3. Quality (consistency + low drawdown) - filters lottery tickets +4. MA200 trend confirmation - only stocks above their MA200 + +With: +- Inverse-vol weighting for risk parity among selected stocks +- SPY MA200 market regime gate (reduce exposure in bear markets) +- Biweekly rebalancing (compromise between signal freshness and turnover) +""" + +import numpy as np +import pandas as pd +from strategies.base import Strategy + + +class CompositeAlphaStrategy(Strategy): + """ + Multi-factor alpha composite with regime gating. + """ + + def __init__( + self, + tickers: list[str] | None = None, + benchmark: str = "SPY", + recovery_window: int = 63, + intermediate_period: int = 147, + skip: int = 21, + quality_window: int = 252, + vol_window: int = 60, + rebal_freq: int = 10, + top_n: int = 20, + regime_gate: bool = True, + regime_ma: int = 200, + ): + self.tickers = tickers + self.benchmark = benchmark + self.recovery_window = recovery_window + self.intermediate_period = intermediate_period + self.skip = skip + self.quality_window = quality_window + self.vol_window = vol_window + self.rebal_freq = rebal_freq + self.top_n = top_n + self.regime_gate = regime_gate + self.regime_ma = regime_ma + + def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame: + # Separate benchmark from stocks if needed + if self.tickers is not None: + stock_data = data[[t for t in self.tickers if t in data.columns]] + else: + stock_cols = [c for c in data.columns if c != self.benchmark] + stock_data = data[stock_cols] + + # --- Factor 1: Recovery --- + recovery = stock_data / stock_data.rolling( + self.recovery_window, min_periods=self.recovery_window + ).min() - 1 + + # --- Factor 2: Intermediate momentum (7-month, skip 1 month) --- + int_mom = stock_data.shift(self.skip).pct_change(self.intermediate_period - self.skip) + + # --- Factor 3: Quality composite --- + # Consistency: fraction of positive 21-day returns + monthly_ret = stock_data.pct_change(21) + consistency = (monthly_ret > 0).astype(float).rolling( + self.quality_window, min_periods=self.quality_window // 2 + ).mean() + + # Up-volume proxy: sum of positive daily returns over 20 days + daily_ret = stock_data.pct_change() + up_vol_proxy = daily_ret.where(daily_ret > 0, 0).rolling(20, min_periods=15).sum() + + # --- Factor 4: Above MA200 (per-stock trend filter) --- + ma200 = stock_data.rolling(200, min_periods=200).mean() + above_ma = (stock_data > ma200) + + # --- Cross-sectional ranks --- + rec_rank = recovery.rank(axis=1, pct=True, na_option="keep") + mom_rank = int_mom.rank(axis=1, pct=True, na_option="keep") + con_rank = consistency.rank(axis=1, pct=True, na_option="keep") + upv_rank = up_vol_proxy.rank(axis=1, pct=True, na_option="keep") + + # Composite: recovery 30%, int_momentum 30%, consistency 20%, up_volume 20% + composite = (0.30 * rec_rank + 0.30 * mom_rank + + 0.20 * con_rank + 0.20 * upv_rank) + + # Apply per-stock MA200 filter: must be in uptrend + composite = composite.where(above_ma, np.nan) + + # --- Select top_n --- + rank = composite.rank(axis=1, ascending=False, na_option="bottom") + n_valid = composite.notna().sum(axis=1) + enough = n_valid >= min(self.top_n, 5) + top_mask = (rank <= self.top_n) & enough.values.reshape(-1, 1) + + # --- Inverse-vol weighting --- + vol = daily_ret.rolling(self.vol_window, min_periods=30).std().replace(0, np.nan) + inv_vol = (1.0 / vol).where(top_mask, 0.0) + row_sums = inv_vol.sum(axis=1).replace(0, np.nan) + signals = inv_vol.div(row_sums, axis=0).fillna(0.0) + + # --- Market regime gate (SPY > MA200) --- + if self.regime_gate and self.benchmark in data.columns: + spy = data[self.benchmark] + spy_ma = spy.rolling(self.regime_ma, min_periods=self.regime_ma).mean() + market_bull = (spy > spy_ma).astype(float) + # Partial scaling: 100% when bullish, 30% when bearish (don't go fully to cash) + regime_scale = market_bull * 0.7 + 0.3 + signals = signals.mul(regime_scale, axis=0) + + # --- Biweekly rebalance --- + warmup = max(self.quality_window, 200, self.intermediate_period) + self.skip + rebal_mask = pd.Series(False, index=data.index) + rebal_indices = list(range(warmup, len(data), self.rebal_freq)) + rebal_mask.iloc[rebal_indices] = True + + signals[~rebal_mask] = np.nan + signals = signals.ffill().fillna(0.0) + signals.iloc[:warmup] = 0.0 + + # Align to full data columns (in case benchmark is in data) + full_signals = pd.DataFrame(0.0, index=data.index, columns=data.columns) + for col in signals.columns: + if col in full_signals.columns: + full_signals[col] = signals[col] + + return full_signals.shift(1).fillna(0.0) diff --git a/strategies/cross_asset_momentum.py b/strategies/cross_asset_momentum.py new file mode 100644 index 0000000..cb4c2b8 --- /dev/null +++ b/strategies/cross_asset_momentum.py @@ -0,0 +1,93 @@ +""" +Cross-asset time-series momentum strategy (ETF-only, inherently PIT). + +Alpha source: Moskowitz, Ooi, Pedersen (2012) — assets with positive +12-month returns continue to trend. Earns during equity crises when +bonds/gold trend up while stocks trend down. + +Universe: SPY, GLD, TLT, IEF, DBC (broad, liquid ETFs) +Signal: 12-month total return; go long top K assets with positive momentum +Rebalance: monthly (21 trading days) +If no asset has positive 12m return → 100% cash (SHY proxy = 0 weights) +""" + +import numpy as np +import pandas as pd + + +class CrossAssetMomentum: + """Time-series momentum across major asset classes.""" + + UNIVERSE = ["SPY", "GLD", "TLT", "IEF", "DBC"] + + def __init__( + self, + lookback: int = 252, + top_k: int = 3, + rebal_freq: int = 21, + vol_scale: bool = True, + vol_window: int = 63, + ): + self.lookback = lookback + self.top_k = top_k + self.rebal_freq = rebal_freq + self.vol_scale = vol_scale + self.vol_window = vol_window + + def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame: + """ + Parameters + ---------- + data : DataFrame with columns including UNIVERSE ETFs (prices). + + Returns + ------- + DataFrame of weights aligned to data.index and data.columns. + """ + # Restrict to our universe (ignore missing gracefully) + available = [t for t in self.UNIVERSE if t in data.columns] + prices = data[available] + + # 12-month return signal (shifted 1 day for execution lag) + mom = prices.pct_change(self.lookback).shift(1) + + # Inverse-vol for position sizing (optional) + if self.vol_scale: + daily_ret = prices.pct_change() + vol = daily_ret.rolling(self.vol_window).std() * np.sqrt(252) + inv_vol = (1.0 / vol).replace([np.inf, -np.inf], np.nan) + else: + inv_vol = None + + weights = pd.DataFrame(0.0, index=data.index, columns=data.columns) + n = len(prices.index) + last_w = pd.Series(0.0, index=available) + + for i in range(self.lookback + 1, n, self.rebal_freq): + row_mom = mom.iloc[i] + # Only go long assets with positive momentum + positive = row_mom[row_mom > 0].sort_values(ascending=False) + + if positive.empty: + last_w = pd.Series(0.0, index=available) + else: + selected = positive.head(self.top_k).index.tolist() + + if self.vol_scale and inv_vol is not None: + iv = inv_vol.iloc[i][selected] + if iv.sum() > 0: + w = iv / iv.sum() + else: + w = pd.Series(1.0 / len(selected), index=selected) + else: + w = pd.Series(1.0 / len(selected), index=selected) + + last_w = pd.Series(0.0, index=available) + last_w[selected] = w.values + + # Hold until next rebalance + end_i = min(i + self.rebal_freq, n) + for col in available: + weights.iloc[i:end_i, weights.columns.get_loc(col)] = last_w[col] + + return weights diff --git a/strategies/enhanced_recovery_momentum.py b/strategies/enhanced_recovery_momentum.py new file mode 100644 index 0000000..8644259 --- /dev/null +++ b/strategies/enhanced_recovery_momentum.py @@ -0,0 +1,99 @@ +""" +Enhanced Recovery Momentum Strategy. + +Improvements over base RecoveryMomentumStrategy: +1. Inverse-volatility weighting (allocate more to lower-vol winners → better risk-adjusted) +2. Monthly rebalancing (controls turnover) +3. Momentum filter gate: only pick stocks with positive 6-month momentum + (avoids "dead cat bounces" — recovery without underlying trend) +4. Volatility regime scaling: reduce exposure when market vol is elevated +""" + +import numpy as np +import pandas as pd +from strategies.base import Strategy + + +class EnhancedRecoveryMomentumStrategy(Strategy): + """ + Recovery + momentum with inverse-vol weighting and regime awareness. + """ + + def __init__( + self, + recovery_window: int = 63, + mom_lookback: int = 252, + mom_skip: int = 21, + intermediate_mom: int = 126, + vol_window: int = 60, + rebal_freq: int = 21, + top_n: int = 20, + regime_scale: bool = True, + ): + self.recovery_window = recovery_window + self.mom_lookback = mom_lookback + self.mom_skip = mom_skip + self.intermediate_mom = intermediate_mom + self.vol_window = vol_window + self.rebal_freq = rebal_freq + self.top_n = top_n + self.regime_scale = regime_scale + + def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame: + # Factor 1: Recovery — price / rolling min + recovery = data / data.rolling(self.recovery_window, min_periods=self.recovery_window).min() - 1 + + # Factor 2: 12-1 month momentum + momentum = data.shift(self.mom_skip).pct_change(self.mom_lookback - self.mom_skip) + + # Factor 3: Intermediate momentum (6m) as a filter gate + # Only consider stocks with positive intermediate trend + intermediate = data.shift(self.mom_skip).pct_change(self.intermediate_mom - self.mom_skip) + trend_gate = intermediate > 0 + + # Cross-sectional percentile ranks + rec_rank = recovery.rank(axis=1, pct=True, na_option="keep") + mom_rank = momentum.rank(axis=1, pct=True, na_option="keep") + + # Composite score (50/50), gated by intermediate trend + composite = 0.5 * rec_rank + 0.5 * mom_rank + composite = composite.where(trend_gate, np.nan) + + # Select top_n + rank = composite.rank(axis=1, ascending=False, na_option="bottom") + n_valid = composite.notna().sum(axis=1) + enough = n_valid >= self.top_n + top_mask = (rank <= self.top_n) & enough.values.reshape(-1, 1) + + # Inverse-vol weighting among selected stocks + returns = data.pct_change() + vol = returns.rolling(self.vol_window, min_periods=30).std().replace(0, np.nan) + inv_vol = (1.0 / vol).where(top_mask, 0.0) + + row_sums = inv_vol.sum(axis=1).replace(0, np.nan) + signals = inv_vol.div(row_sums, axis=0).fillna(0.0) + + # Regime scaling: reduce exposure when market vol is high + if self.regime_scale: + market_vol = vol.mean(axis=1) + vol_median = market_vol.rolling(252, min_periods=126).median() + vol_ratio = (vol_median / market_vol).clip(0.5, 1.2) + signals = signals.mul(vol_ratio, axis=0) + # Re-normalize: cap at 1.0 + row_totals = signals.sum(axis=1) + overflow = row_totals > 1.0 + signals.loc[overflow] = signals.loc[overflow].div( + row_totals[overflow], axis=0 + ) + + # Monthly rebalance: keep only rebal-day signals, forward-fill + warmup = max(self.mom_lookback, self.recovery_window, self.vol_window + 252) + rebal_mask = pd.Series(False, index=data.index) + rebal_indices = list(range(warmup, len(data), self.rebal_freq)) + rebal_mask.iloc[rebal_indices] = True + + signals[~rebal_mask] = np.nan + signals = signals.ffill().fillna(0.0) + signals.iloc[:warmup] = 0.0 + + return signals.shift(1).fillna(0.0) diff --git a/strategies/ensemble_alpha.py b/strategies/ensemble_alpha.py new file mode 100644 index 0000000..cced1a9 --- /dev/null +++ b/strategies/ensemble_alpha.py @@ -0,0 +1,367 @@ +""" +Round 3: Signal-level ensemble of the two best strategies. + +Key insight from R1/R2: +- FactorCombo rec_mfilt+deep_upvol: CAGR 34.6%, MaxDD -33.9%, Calmar 1.02 +- Recovery+Mom Top20: CAGR 34.5%, MaxDD -37.7%, Calmar 0.91 +- Inv-vol weighting HURTS recovery signals (they're high-vol by nature) +- More factors = more noise for this alpha source +- Monthly rebalancing is optimal + +New approach: +1. Ensemble the two best SIGNALS (not strategies) at the rank level + → diversifies stock picks while preserving signal strength +2. Equal weighting (proven better for recovery-type signals) +3. Tail-risk protection: only scale down in EXTREME drawdown regimes + (>15% drawdown from peak), not regular vol spikes +4. Test whether a 126-day recovery (deeper) adds signal vs 63-day +""" + +import numpy as np +import pandas as pd +from strategies.base import Strategy + + +def _rank(df): + return df.rank(axis=1, pct=True, na_option="keep") + + +class EnsembleAlphaStrategy(Strategy): + """ + Ensemble of the two strongest signals with tail-risk protection. + """ + + def __init__( + self, + rebal_freq: int = 21, + top_n: int = 20, + tail_protection: bool = True, + tail_threshold: float = -0.15, # drawdown level to trigger protection + tail_scale: float = 0.5, # how much to reduce in tail event + ): + self.rebal_freq = rebal_freq + self.top_n = top_n + self.tail_protection = tail_protection + self.tail_threshold = tail_threshold + self.tail_scale = tail_scale + + def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame: + p = data + + # === Signal A: rec_mfilt + deep_upvol (from FactorCombo) === + rec_126 = p / p.rolling(126, min_periods=126).min() - 1 + mom_filter = p.shift(21).pct_change(105) + rec_mfilt = rec_126.where(mom_filter > 0, np.nan) + rec_mfilt_r = _rank(rec_mfilt) + + ret = p.pct_change() + up_vol = ret.where(ret > 0, 0).rolling(20, min_periods=15).sum() + deep_upvol = _rank(rec_126) * _rank(up_vol) + deep_upvol_r = _rank(deep_upvol) + + signal_a = 0.5 * rec_mfilt_r + 0.5 * deep_upvol_r + + # === Signal B: Recovery 63d + 12-1 momentum (from RecoveryMom) === + rec_63 = p / p.rolling(63, min_periods=63).min() - 1 + mom_12_1 = p.shift(21).pct_change(231) + + rec_63_r = _rank(rec_63) + mom_r = _rank(mom_12_1) + + signal_b = 0.5 * rec_63_r + 0.5 * mom_r + + # === Ensemble: average of both signals === + ensemble = 0.5 * signal_a + 0.5 * signal_b + + # === Select top_n === + rank = ensemble.rank(axis=1, ascending=False, na_option="bottom") + n_valid = ensemble.notna().sum(axis=1) + enough = n_valid >= self.top_n + top_mask = (rank <= self.top_n) & enough.values.reshape(-1, 1) + + # Equal weight (proven better for recovery signals) + raw = top_mask.astype(float) + row_sums = raw.sum(axis=1).replace(0, np.nan) + signals = raw.div(row_sums, axis=0).fillna(0.0) + + # === Tail-risk protection === + if self.tail_protection: + # Portfolio equity proxy: equal-weight market return + mkt_ret = ret.mean(axis=1) + mkt_eq = (1 + mkt_ret).cumprod() + mkt_dd = mkt_eq / mkt_eq.cummax() - 1 + in_tail = mkt_dd < self.tail_threshold + scale = pd.Series(1.0, index=data.index) + scale[in_tail] = self.tail_scale + signals = signals.mul(scale, axis=0) + + # === Monthly rebalance === + warmup = 252 + rebal_mask = pd.Series(False, index=data.index) + rebal_indices = list(range(warmup, len(data), self.rebal_freq)) + rebal_mask.iloc[rebal_indices] = True + + signals[~rebal_mask] = np.nan + signals = signals.ffill().fillna(0.0) + signals.iloc[:warmup] = 0.0 + + return signals.shift(1).fillna(0.0) + + +class EnhancedFactorComboStrategy(Strategy): + """ + FactorCombo signal enhanced with: + 1. Additional momentum confirmation (12-1 momentum rank as tiebreaker) + 2. Concentration in top conviction names (top_n=15 instead of 20) + 3. Optional tail protection + """ + + def __init__( + self, + rebal_freq: int = 21, + top_n: int = 15, + mom_boost: float = 0.2, # weight given to additional momentum signal + tail_protection: bool = False, + ): + self.rebal_freq = rebal_freq + self.top_n = top_n + self.mom_boost = mom_boost + self.tail_protection = tail_protection + + def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame: + p = data + + # Core signal: rec_mfilt + deep_upvol + rec_126 = p / p.rolling(126, min_periods=126).min() - 1 + mom_filter = p.shift(21).pct_change(105) + rec_mfilt = rec_126.where(mom_filter > 0, np.nan) + rec_mfilt_r = _rank(rec_mfilt) + + ret = p.pct_change() + up_vol = ret.where(ret > 0, 0).rolling(20, min_periods=15).sum() + deep_upvol = _rank(rec_126) * _rank(up_vol) + deep_upvol_r = _rank(deep_upvol) + + base_signal = 0.5 * rec_mfilt_r + 0.5 * deep_upvol_r + + # Momentum boost: 12-1 month return rank + mom_12_1 = p.shift(21).pct_change(231) + mom_r = _rank(mom_12_1) + + # Combined: base + momentum tiebreaker + signal = (1 - self.mom_boost) * base_signal + self.mom_boost * mom_r + + # Select top_n + rank = signal.rank(axis=1, ascending=False, na_option="bottom") + n_valid = signal.notna().sum(axis=1) + enough = n_valid >= self.top_n + top_mask = (rank <= self.top_n) & enough.values.reshape(-1, 1) + + # Equal weight + raw = top_mask.astype(float) + row_sums = raw.sum(axis=1).replace(0, np.nan) + signals = raw.div(row_sums, axis=0).fillna(0.0) + + # Tail protection + if self.tail_protection: + mkt_ret = ret.mean(axis=1) + mkt_eq = (1 + mkt_ret).cumprod() + mkt_dd = mkt_eq / mkt_eq.cummax() - 1 + in_tail = mkt_dd < -0.15 + scale = pd.Series(1.0, index=data.index) + scale[in_tail] = 0.5 + signals = signals.mul(scale, axis=0) + + # Monthly rebalance + warmup = 252 + rebal_mask = pd.Series(False, index=data.index) + rebal_indices = list(range(warmup, len(data), self.rebal_freq)) + rebal_mask.iloc[rebal_indices] = True + + signals[~rebal_mask] = np.nan + signals = signals.ffill().fillna(0.0) + signals.iloc[:warmup] = 0.0 + + return signals.shift(1).fillna(0.0) + + +class RiskManagedEnsembleStrategy(Strategy): + """ + EnsembleAlpha with market-aware drawdown risk management. + + Key insight: Using the strategy's OWN drawdown to scale down creates + a negative feedback loop (cut → miss rebound → deeper DD → cut more). + Instead, use MARKET drawdown as the systemic risk signal: + - Market crash → reduce exposure (systemic risk) + - Strategy underperforms but market is fine → stay invested (alpha issue, not risk) + + Mechanisms: + 1. Market DD dampener: scales down proportionally to equal-weight market drawdown. + Only fires during systemic stress. Recovers as market recovers. + 2. Vol spike guard: when 10-day portfolio vol > 90th percentile of history, + reduce to vol_spike_floor. Catches acute crises. + + Both use lagged (T-1) estimates → PIT-safe. + Parameter choices justified by market microstructure (not optimized): + - dd_denom=0.20 → at 20% market crash, exposure reduced to floor + - dd_floor=0.40 → never go below 40% (still participate in recovery) + - vol_spike_floor=0.50 → during vol spikes, halve exposure + """ + + def __init__( + self, + top_n: int = 10, + dd_floor: float = 0.40, + dd_denom: float = 0.20, + vol_spike_guard: bool = True, + vol_spike_window: int = 10, + vol_spike_lookback: int = 252, + vol_spike_floor: float = 0.50, + ): + self.ensemble = EnsembleAlphaStrategy(top_n=top_n, tail_protection=False) + self.dd_floor = dd_floor + self.dd_denom = dd_denom + self.vol_spike_guard = vol_spike_guard + self.vol_spike_window = vol_spike_window + self.vol_spike_lookback = vol_spike_lookback + self.vol_spike_floor = vol_spike_floor + + def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame: + # Step 1: Get raw signals from the ensemble (already shifted by 1) + raw = self.ensemble.generate_signals(data) + + # Step 2: Compute MARKET returns (equal-weight of all stocks) + daily_rets = data.pct_change().fillna(0.0) + mkt_rets = daily_rets.mean(axis=1) + + # Step 3: Market drawdown dampener + mkt_eq = (1 + mkt_rets).cumprod() + mkt_dd = mkt_eq / mkt_eq.cummax() - 1 # always ≤ 0 + # Linear: at DD=0 → 1.0, at DD=-dd_denom → dd_floor + dd_scale = (1.0 + mkt_dd / self.dd_denom).clip(lower=self.dd_floor, upper=1.0) + dd_scale_lagged = dd_scale.shift(1).fillna(1.0) # PIT + + # Step 4: Vol spike guard (uses portfolio's own vol for specificity) + if self.vol_spike_guard: + port_rets = (raw * daily_rets).sum(axis=1) + short_vol = port_rets.rolling(self.vol_spike_window, min_periods=5).std() * np.sqrt(252) + vol_90th = short_vol.rolling(self.vol_spike_lookback, min_periods=126).quantile(0.90) + in_spike = short_vol > vol_90th + vol_scale = pd.Series(1.0, index=data.index) + vol_scale[in_spike] = self.vol_spike_floor + vol_scale_lagged = vol_scale.shift(1).fillna(1.0) # PIT + else: + vol_scale_lagged = 1.0 + + # Step 5: Combined scaling + final_scale = dd_scale_lagged * vol_scale_lagged + return raw.mul(final_scale, axis=0) + + +class SharpeBoostedEnsembleStrategy(Strategy): + """ + Optimized ensemble targeting Sharpe >1.5 while maintaining high CAGR. + + Key improvements over EnsembleAlphaStrategy: + 1. Bimonthly rebalance (42d): recovery signals have 126-day lookback, + monthly rebal causes unnecessary turnover. Let winners run. + 2. Slightly wider basket (top_n=12): diversifies idiosyncratic risk + without diluting alpha (sweet spot between 10-15). + 3. Asymmetric vol scaling: only de-risk in high-vol NEGATIVE return + regimes (high-vol + positive = good, don't cut). + 4. Light market-DD dampener: only fires in severe systemic stress + (dd_denom=0.35 → need 35% market crash to reach floor). + + PIT compliance: + - All signal lookbacks use .shift(21) or rolling windows (no current-day data) + - Asymmetric vol uses .shift(1) on scale + - DD dampener uses .shift(1) on mkt_dd + - Final signals use .shift(1) for execution lag + + Parameter count: 4 meaningful (rebal_freq, top_n, asym_vol_floor, dd_denom) + All have economic justification, not optimized on in-sample. + """ + + def __init__( + self, + top_n: int = 12, + rebal_freq: int = 42, + asym_vol_floor: float = 0.50, + dd_floor: float = 0.70, + dd_denom: float = 0.35, + ): + self.top_n = top_n + self.rebal_freq = rebal_freq + self.asym_vol_floor = asym_vol_floor + self.dd_floor = dd_floor + self.dd_denom = dd_denom + + def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame: + p = data + ret = p.pct_change() + + # === Signal A: rec_mfilt + deep_upvol === + rec_126 = p / p.rolling(126, min_periods=126).min() - 1 + mom_filter = p.shift(21).pct_change(105) + rec_mfilt = rec_126.where(mom_filter > 0, np.nan) + rec_mfilt_r = _rank(rec_mfilt) + + up_vol = ret.where(ret > 0, 0).rolling(20, min_periods=15).sum() + deep_upvol = _rank(rec_126) * _rank(up_vol) + deep_upvol_r = _rank(deep_upvol) + signal_a = 0.5 * rec_mfilt_r + 0.5 * deep_upvol_r + + # === Signal B: Recovery 63d + 12-1 momentum === + rec_63 = p / p.rolling(63, min_periods=63).min() - 1 + mom_12_1 = p.shift(21).pct_change(231) + rec_63_r = _rank(rec_63) + mom_r = _rank(mom_12_1) + signal_b = 0.5 * rec_63_r + 0.5 * mom_r + + # === Ensemble: equal-weight average of both signals === + ensemble = 0.5 * signal_a + 0.5 * signal_b + + # === Select top_n === + rank = ensemble.rank(axis=1, ascending=False, na_option="bottom") + n_valid = ensemble.notna().sum(axis=1) + enough = n_valid >= self.top_n + top_mask = (rank <= self.top_n) & enough.values.reshape(-1, 1) + + raw = top_mask.astype(float) + row_sums = raw.sum(axis=1).replace(0, np.nan) + signals = raw.div(row_sums, axis=0).fillna(0.0) + + # === Bimonthly rebalance (42 trading days) === + warmup = 252 + rebal_mask = pd.Series(False, index=data.index) + rebal_indices = list(range(warmup, len(data), self.rebal_freq)) + rebal_mask.iloc[rebal_indices] = True + signals[~rebal_mask] = np.nan + signals = signals.ffill().fillna(0.0) + signals.iloc[:warmup] = 0.0 + signals = signals.shift(1).fillna(0.0) # PIT: 1-day execution lag + + # === Asymmetric vol scaling === + # Only reduce exposure when vol is high AND returns are negative + # High vol + positive returns = riding a trend, don't cut + daily_rets = data.pct_change().fillna(0.0) + port_rets = (signals * daily_rets).sum(axis=1) + short_vol = port_rets.rolling(20, min_periods=10).std() * np.sqrt(252) + vol_median = short_vol.rolling(252, min_periods=126).median() + recent_ret = port_rets.rolling(20, min_periods=10).sum() + high_vol_neg = (short_vol > vol_median * 1.5) & (recent_ret < 0) + asym_scale = pd.Series(1.0, index=data.index) + asym_scale[high_vol_neg] = self.asym_vol_floor + signals = signals.mul(asym_scale.shift(1).fillna(1.0), axis=0) # PIT + + # === Light market-DD dampener === + # Uses market (not strategy) drawdown to avoid negative feedback loop + mkt_rets = daily_rets.mean(axis=1) + mkt_eq = (1 + mkt_rets).cumprod() + mkt_dd = mkt_eq / mkt_eq.cummax() - 1 + dd_scale = (1.0 + mkt_dd / self.dd_denom).clip( + lower=self.dd_floor, upper=1.0 + ) + signals = signals.mul(dd_scale.shift(1).fillna(1.0), axis=0) # PIT + + return signals diff --git a/strategies/hybrid_alpha.py b/strategies/hybrid_alpha.py new file mode 100644 index 0000000..44d1c10 --- /dev/null +++ b/strategies/hybrid_alpha.py @@ -0,0 +1,184 @@ +""" +Hybrid Alpha Strategy - Round 2 iteration. + +Takes the best elements from the top performers: +1. FactorCombo's rec_mfilt + deep_upvol signal (strongest alpha) +2. Inverse-vol weighting (better risk-adjusted from ImprovedMomQuality) +3. Light regime awareness (partial scale-down, not binary) +4. Monthly rebalancing + +Also tests: +- Recovery + quality blend without MA200 filter +- Wider top_n for diversification +""" + +import numpy as np +import pandas as pd +from strategies.base import Strategy + + +def _rank(df): + return df.rank(axis=1, pct=True, na_option="keep") + + +class HybridAlphaStrategy(Strategy): + """ + Combines FactorCombo's best signal with risk-parity weighting. + """ + + def __init__( + self, + rebal_freq: int = 21, + top_n: int = 20, + vol_window: int = 60, + use_invvol: bool = True, + regime_dampen: float = 0.5, # scale factor in bear regime (1.0 = no regime) + ): + self.rebal_freq = rebal_freq + self.top_n = top_n + self.vol_window = vol_window + self.use_invvol = use_invvol + self.regime_dampen = regime_dampen + + def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame: + p = data + + # --- Signal: rec_mfilt + deep x upvol (from FactorCombo) --- + # Recovery momentum-filtered + rec = p / p.rolling(126, min_periods=126).min() - 1 + mom = p.shift(21).pct_change(105) + rec_mfilt = rec.where(mom > 0, np.nan) + rec_mfilt_r = _rank(rec_mfilt) + + # Deep recovery x up-volume + rec_126 = p / p.rolling(126, min_periods=126).min() - 1 + ret = p.pct_change() + up_vol = ret.where(ret > 0, 0).rolling(20, min_periods=15).sum() + deep_upvol = _rank(rec_126) * _rank(up_vol) + deep_upvol_r = _rank(deep_upvol) + + # Combined signal + signal = 0.5 * rec_mfilt_r + 0.5 * deep_upvol_r + + # --- Select top_n --- + rank = signal.rank(axis=1, ascending=False, na_option="bottom") + n_valid = signal.notna().sum(axis=1) + enough = n_valid >= self.top_n + top_mask = (rank <= self.top_n) & enough.values.reshape(-1, 1) + + if self.use_invvol: + # Inverse-vol weighting + vol = ret.rolling(self.vol_window, min_periods=30).std().replace(0, np.nan) + inv_vol = (1.0 / vol).where(top_mask, 0.0) + row_sums = inv_vol.sum(axis=1).replace(0, np.nan) + signals = inv_vol.div(row_sums, axis=0).fillna(0.0) + else: + # Equal weight + raw = top_mask.astype(float) + row_sums = raw.sum(axis=1).replace(0, np.nan) + signals = raw.div(row_sums, axis=0).fillna(0.0) + + # --- Light regime dampening --- + if self.regime_dampen < 1.0: + # Use market-wide vol regime instead of MA200 (more responsive) + market_vol = ret.mean(axis=1).rolling(20).std() * np.sqrt(252) + vol_90th = market_vol.rolling(252, min_periods=126).quantile(0.90) + high_vol = market_vol > vol_90th + regime_scale = pd.Series(1.0, index=data.index) + regime_scale[high_vol] = self.regime_dampen + signals = signals.mul(regime_scale, axis=0) + + # --- Monthly rebalance --- + warmup = 252 + rebal_mask = pd.Series(False, index=data.index) + rebal_indices = list(range(warmup, len(data), self.rebal_freq)) + rebal_mask.iloc[rebal_indices] = True + + signals[~rebal_mask] = np.nan + signals = signals.ffill().fillna(0.0) + signals.iloc[:warmup] = 0.0 + + return signals.shift(1).fillna(0.0) + + +class RecoveryQualityBlendStrategy(Strategy): + """ + Blends recovery, momentum, and quality without strict MA200 filter. + Uses intermediate momentum as a soft signal (not hard gate). + """ + + def __init__( + self, + recovery_window: int = 63, + mom_lookback: int = 252, + mom_skip: int = 21, + quality_window: int = 252, + vol_window: int = 60, + rebal_freq: int = 21, + top_n: int = 20, + ): + self.recovery_window = recovery_window + self.mom_lookback = mom_lookback + self.mom_skip = mom_skip + self.quality_window = quality_window + self.vol_window = vol_window + self.rebal_freq = rebal_freq + self.top_n = top_n + + def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame: + p = data + + # Recovery + recovery = p / p.rolling(self.recovery_window, min_periods=self.recovery_window).min() - 1 + + # 12-1 month momentum + momentum = p.shift(self.mom_skip).pct_change(self.mom_lookback - self.mom_skip) + + # Intermediate momentum (7m) + int_mom = p.shift(self.mom_skip).pct_change(147 - self.mom_skip) + + # Quality: consistency + monthly_ret = p.pct_change(21) + consistency = (monthly_ret > 0).astype(float).rolling( + self.quality_window, min_periods=self.quality_window // 2 + ).mean() + + # Up-volume proxy + ret = p.pct_change() + up_vol = ret.where(ret > 0, 0).rolling(20, min_periods=15).sum() + + # Cross-sectional ranks + rec_rank = _rank(recovery) + mom_rank = _rank(momentum) + int_mom_rank = _rank(int_mom) + con_rank = _rank(consistency) + upv_rank = _rank(up_vol) + + # Composite: weighted blend of all factors + # Recovery 25%, momentum 25%, intermediate momentum 20%, quality 15%, up_vol 15% + composite = (0.25 * rec_rank + 0.25 * mom_rank + 0.20 * int_mom_rank + + 0.15 * con_rank + 0.15 * upv_rank) + + # Select top_n + rank = composite.rank(axis=1, ascending=False, na_option="bottom") + n_valid = composite.notna().sum(axis=1) + enough = n_valid >= self.top_n + top_mask = (rank <= self.top_n) & enough.values.reshape(-1, 1) + + # Inverse-vol weighting + vol = ret.rolling(self.vol_window, min_periods=30).std().replace(0, np.nan) + inv_vol = (1.0 / vol).where(top_mask, 0.0) + row_sums = inv_vol.sum(axis=1).replace(0, np.nan) + signals = inv_vol.div(row_sums, axis=0).fillna(0.0) + + # Monthly rebalance + warmup = max(self.mom_lookback, self.quality_window, self.recovery_window) + self.mom_skip + rebal_mask = pd.Series(False, index=data.index) + rebal_indices = list(range(warmup, len(data), self.rebal_freq)) + rebal_mask.iloc[rebal_indices] = True + + signals[~rebal_mask] = np.nan + signals = signals.ffill().fillna(0.0) + signals.iloc[:warmup] = 0.0 + + return signals.shift(1).fillna(0.0) diff --git a/strategies/improved_momentum_quality.py b/strategies/improved_momentum_quality.py new file mode 100644 index 0000000..aa7a71e --- /dev/null +++ b/strategies/improved_momentum_quality.py @@ -0,0 +1,94 @@ +""" +Improved Momentum Quality Strategy. + +Improvements over base MomentumQualityStrategy: +1. Monthly rebalancing (original rebalances daily → high turnover) +2. Added recovery factor (strong predictor per IC analysis) +3. Replaced expensive .apply() consistency calc with vectorized version +4. Inverse-vol weighting instead of equal-weight +5. NaN handling fixed throughout +""" + +import numpy as np +import pandas as pd +from strategies.base import Strategy + + +class ImprovedMomentumQualityStrategy(Strategy): + """ + Momentum + quality + recovery with monthly rebal and inv-vol weighting. + """ + + def __init__( + self, + momentum_period: int = 252, + skip: int = 21, + quality_window: int = 252, + recovery_window: int = 63, + vol_window: int = 60, + rebal_freq: int = 21, + top_n: int = 20, + ): + self.momentum_period = momentum_period + self.skip = skip + self.quality_window = quality_window + self.recovery_window = recovery_window + self.vol_window = vol_window + self.rebal_freq = rebal_freq + self.top_n = top_n + + def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame: + # --- Momentum factor --- + momentum = data.shift(self.skip).pct_change(self.momentum_period - self.skip) + + # --- Quality: return consistency (vectorized) --- + # Fraction of positive 21-day returns over rolling window + monthly_ret = data.pct_change(21) + positive_indicator = (monthly_ret > 0).astype(float) + consistency = positive_indicator.rolling( + self.quality_window, min_periods=self.quality_window // 2 + ).mean() + + # --- Quality: inverse max drawdown --- + rolling_max = data.rolling(self.quality_window, min_periods=self.quality_window // 2).max() + drawdown = data / rolling_max - 1 + worst_dd = drawdown.rolling(self.quality_window, min_periods=self.quality_window // 2).min() + inv_dd = -worst_dd # higher = smaller drawdown = better + + # --- Recovery factor --- + recovery = data / data.rolling(self.recovery_window, min_periods=self.recovery_window).min() - 1 + + # --- Cross-sectional ranking --- + mom_rank = momentum.rank(axis=1, pct=True, na_option="keep") + con_rank = consistency.rank(axis=1, pct=True, na_option="keep") + dd_rank = inv_dd.rank(axis=1, pct=True, na_option="keep") + rec_rank = recovery.rank(axis=1, pct=True, na_option="keep") + + # Composite: momentum 35%, recovery 25%, consistency 20%, drawdown 20% + scores = (0.35 * mom_rank + 0.25 * rec_rank + + 0.20 * con_rank + 0.20 * dd_rank) + + # --- Select top_n --- + n_valid = scores.notna().sum(axis=1) + enough = n_valid >= self.top_n + score_rank = scores.rank(axis=1, ascending=False, na_option="bottom") + top_mask = (score_rank <= self.top_n) & enough.values.reshape(-1, 1) + + # --- Inverse-vol weighting --- + returns = data.pct_change() + vol = returns.rolling(self.vol_window, min_periods=30).std().replace(0, np.nan) + inv_vol = (1.0 / vol).where(top_mask, 0.0) + row_sums = inv_vol.sum(axis=1).replace(0, np.nan) + signals = inv_vol.div(row_sums, axis=0).fillna(0.0) + + # --- Monthly rebalance --- + warmup = max(self.momentum_period, self.quality_window, self.recovery_window) + self.skip + rebal_mask = pd.Series(False, index=data.index) + rebal_indices = list(range(warmup, len(data), self.rebal_freq)) + rebal_mask.iloc[rebal_indices] = True + + signals[~rebal_mask] = np.nan + signals = signals.ffill().fillna(0.0) + signals.iloc[:warmup] = 0.0 + + return signals.shift(1).fillna(0.0) diff --git a/strategies/long_hedged.py b/strategies/long_hedged.py new file mode 100644 index 0000000..23241c2 --- /dev/null +++ b/strategies/long_hedged.py @@ -0,0 +1,124 @@ +"""Market-hedged long-only stock portfolio. + +Architecture +------------ +Long: top-N stock portfolio (factor-selected, inv-vol weighted). +Short hedge: SPY (or SH ETF) at hedge_ratio × long_gross. + +This isolates cross-sectional stock-selection alpha while removing the +broad-market beta. Unlike L/S of individual stocks, the short leg is on +the index — so: + * no meme-stock blowups on short side (GME/AMC type events) + * borrow cost on SPY is ≈ 5-15 bps annualized (very cheap) + * no short-dividend pass-through issue (pay SPY div, but that's offset + by long-side dividends roughly) + +Because the long leg is monthly-rebalanced and the short hedge is fixed +at -1.0 × long_gross, total turnover is dominated by the long leg — +similar to V6 long-only. + +Output is PIT-safe via terminal `.shift(1)`. +""" +from __future__ import annotations + +import numpy as np +import pandas as pd + +from strategies.base import Strategy +from strategies.factor_combo import SIGNAL_REGISTRY + + +class LongHedgedStock(Strategy): + """Long-only stock momentum hedged with SPY short.""" + + def __init__( + self, + signal_name: str = "rec_mfilt+deep_upvol", + top_n: int = 15, + rebal_freq: int = 21, + hedge_symbol: str = "SPY", + hedge_ratio: float = 1.0, + long_gross: float = 1.0, + invvol_window: int = 60, + invvol_floor: float = 0.10, + invvol_cap: float = 0.20, + stock_universe: list[str] | None = None, + # Regime gate: zero out positions when regime_signal < its MA(ma_window) + regime_gate: bool = False, + regime_signal: str = "SPY", + ma_window: int = 200, + ) -> None: + if signal_name not in SIGNAL_REGISTRY: + raise ValueError(f"Unknown signal: {signal_name}") + self.signal_name = signal_name + self.signal_func = SIGNAL_REGISTRY[signal_name] + self.top_n = top_n + self.rebal_freq = rebal_freq + self.hedge_symbol = hedge_symbol + self.hedge_ratio = hedge_ratio + self.long_gross = long_gross + self.invvol_window = invvol_window + self.invvol_floor = invvol_floor + self.invvol_cap = invvol_cap + self.stock_universe = stock_universe + self.regime_gate = regime_gate + self.regime_signal = regime_signal + self.ma_window = ma_window + + def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame: + if self.hedge_symbol not in data.columns: + raise ValueError(f"hedge_symbol {self.hedge_symbol!r} missing from panel") + universe = self.stock_universe or [ + c for c in data.columns if c != self.hedge_symbol + ] + universe = [c for c in universe if c in data.columns] + + stock_panel = data[universe] + sig = self.signal_func(stock_panel) + rank = sig.rank(axis=1, ascending=False, na_option="bottom") + n_valid = sig.notna().sum(axis=1) + enough = n_valid >= self.top_n + top_mask = (rank <= self.top_n) & enough.values.reshape(-1, 1) + + # Inv-vol weighting within selection + rets = stock_panel.pct_change(fill_method=None) + vol = rets.rolling(self.invvol_window, + min_periods=self.invvol_window // 2).std() * np.sqrt(252) + vol_clipped = vol.clip(lower=self.invvol_floor, upper=self.invvol_cap) + invvol = (1.0 / vol_clipped).where(top_mask, 0.0) + row_sums = invvol.sum(axis=1).replace(0, np.nan) + long_w = invvol.div(row_sums, axis=0).fillna(0.0) * self.long_gross + + # Monthly rebalance + warmup = 252 + rebal_mask = pd.Series(False, index=data.index) + rebal_idx = list(range(warmup, len(data), self.rebal_freq)) + rebal_mask.iloc[rebal_idx] = True + long_w[~rebal_mask] = np.nan + long_w = long_w.ffill().fillna(0.0) + long_w.iloc[:warmup] = 0.0 + + # Build full weights frame: longs in stocks, short in SPY + out = pd.DataFrame(0.0, index=data.index, columns=data.columns) + for c in long_w.columns: + if c in out.columns: + out[c] = long_w[c] + + # Short hedge: only when long leg is active (gross > 0) + long_gross_now = long_w.abs().sum(axis=1) + active = (long_gross_now > 0) + out[self.hedge_symbol] = -self.hedge_ratio * long_gross_now * active.astype(float) + + # Regime gate: zero everything when regime signal is in bear regime. + # Avoids the negative-carry case where long stocks tank with SPY but + # the hedge can't fully offset (since long has higher beta). + if self.regime_gate and self.regime_signal in data.columns: + regime_px = data[self.regime_signal] + ma = regime_px.rolling(self.ma_window).mean() + risk_on = (regime_px > ma).astype(float).fillna(0.0) + out = out.mul(risk_on, axis=0) + + return out.shift(1).fillna(0.0) + + +__all__ = ["LongHedgedStock"] diff --git a/strategies/ls_momentum.py b/strategies/ls_momentum.py new file mode 100644 index 0000000..e16be41 --- /dev/null +++ b/strategies/ls_momentum.py @@ -0,0 +1,251 @@ +"""Industry-neutral long/short momentum on the S&P 500. + +Strategy +-------- +At each rebalance date (default: monthly): + 1. Compute 12-1 month momentum for every stock in the panel. + 2. Group stocks by GICS sector. + 3. Within each sector, rank by momentum. + 4. Long the top `long_pct` (default 20%) of each sector. + 5. Short the bottom `short_pct` (default 20%) of each sector. + 6. Equal-weight within long-leg and short-leg, scaled so gross long = 1.0 + and gross short = 1.0 → 200% gross exposure, ~0 net (β ≈ 0). + +The β-neutrality comes from sector-level matching: each sector contributes +both long and short positions in equal $-amounts, so sector and (mostly) +market exposures cancel out. + +Output +------ +A weights DataFrame with positive (long) and negative (short) entries. +PIT-safe via terminal `.shift(1)`. + +Costs +----- +Realistic backtest of L/S requires three additional costs not present in +long-only: + * borrow fee on the short leg (handled by the eval script, not here) + * higher slippage per turnover (this strategy churns more than V5) + * dividend payment on shorts (small for SP500 ~ 1.5% × |short_w|) +The strategy reports raw weights; the eval script applies costs. +""" +from __future__ import annotations + +import os +import urllib.request +import io +import json + +import numpy as np +import pandas as pd + +from strategies.base import Strategy + + +SECTOR_CACHE = "data/us_sectors.csv" +WIKIPEDIA_SP500_URL = "https://en.wikipedia.org/wiki/List_of_S%26P_500_companies" + + +def fetch_sp500_sectors(force: bool = False) -> pd.DataFrame: + """Return a DataFrame indexed by ticker with GICS sector / sub-industry. + + Cached at data/us_sectors.csv. Wikipedia is the canonical source for + current S&P 500 sector membership; for backtest purposes we use today's + sector — sector membership is stable enough year-over-year that this + introduces minimal lookahead bias for an industry-neutral strategy. + """ + if not force and os.path.exists(SECTOR_CACHE): + df = pd.read_csv(SECTOR_CACHE, index_col=0) + if "GICS Sector" in df.columns and len(df) > 100: + return df + + print("--- Fetching S&P 500 GICS sectors from Wikipedia ---") + headers = {"User-Agent": "Mozilla/5.0 (quant-backtest)"} + req = urllib.request.Request(WIKIPEDIA_SP500_URL, headers=headers) + with urllib.request.urlopen(req) as resp: + html = resp.read().decode("utf-8") + tables = pd.read_html(io.StringIO(html)) + df = tables[0] + df = df.rename(columns={"Symbol": "ticker"}) + df["ticker"] = df["ticker"].str.replace(".", "-", regex=False) + df = df.set_index("ticker") + keep = [c for c in df.columns if c in ("GICS Sector", "GICS Sub-Industry", + "Security")] + df = df[keep] + os.makedirs(os.path.dirname(SECTOR_CACHE), exist_ok=True) + df.to_csv(SECTOR_CACHE) + print(f"--- Cached {len(df)} sector mappings to {SECTOR_CACHE} ---") + return df + + +def _signal_mom_12_1(prices: pd.DataFrame) -> pd.DataFrame: + """12-1 month cross-sectional momentum (highest = long).""" + return prices.shift(21).pct_change(231) + + +def _signal_reversal_1m(prices: pd.DataFrame) -> pd.DataFrame: + """1-month reversal: highest 21-day return → SHORT (so we negate).""" + return -prices.pct_change(21) + + +def _signal_reversal_5d(prices: pd.DataFrame) -> pd.DataFrame: + """Short-term 5-day reversal.""" + return -prices.pct_change(5) + + +def _signal_recovery_63(prices: pd.DataFrame) -> pd.DataFrame: + """Recovery factor: price / 63d low (V-shape continuation, long-only-friendly).""" + return prices / prices.rolling(63, min_periods=63).min() - 1 + + +def _signal_low_vol(prices: pd.DataFrame) -> pd.DataFrame: + """Low-vol: invert 60-day realized vol so low vol → high signal.""" + rets = prices.pct_change(fill_method=None) + vol = rets.rolling(60, min_periods=40).std() * np.sqrt(252) + return -vol + + +def _signal_quality_mom(prices: pd.DataFrame) -> pd.DataFrame: + """Composite: 12-1 mom + consistency (% positive days over 252d) + low-vol. + + Combines a positive long-side selection (mom × consistency) and avoids the + fragile far-tail of pure momentum by inverse-vol weighting. + """ + mom = prices.shift(21).pct_change(231) + rets = prices.pct_change(fill_method=None) + pos_days = (rets > 0).rolling(252, min_periods=126).mean() + vol = rets.rolling(60, min_periods=40).std() * np.sqrt(252) + + mom_r = mom.rank(axis=1, pct=True, na_option="keep") + cons_r = pos_days.rank(axis=1, pct=True, na_option="keep") + inv_vol_r = (-vol).rank(axis=1, pct=True, na_option="keep") + return 0.4 * mom_r + 0.3 * cons_r + 0.3 * inv_vol_r + + +def _signal_mom_x_lowvol(prices: pd.DataFrame) -> pd.DataFrame: + """Momentum filtered by low-vol — long winners, short LOW-vol losers. + + Reduces meme-stock blowups on the short leg by avoiding high-vol losers. + """ + mom = prices.shift(21).pct_change(231) + rets = prices.pct_change(fill_method=None) + vol = rets.rolling(60, min_periods=40).std() * np.sqrt(252) + mom_r = mom.rank(axis=1, pct=True, na_option="keep") + inv_vol_r = (-vol).rank(axis=1, pct=True, na_option="keep") + return 0.5 * mom_r + 0.5 * inv_vol_r + + +SIGNAL_REGISTRY = { + "mom_12_1": _signal_mom_12_1, + "reversal_1m": _signal_reversal_1m, + "reversal_5d": _signal_reversal_5d, + "recovery_63": _signal_recovery_63, + "low_vol": _signal_low_vol, + "quality_mom": _signal_quality_mom, + "mom_x_lowvol": _signal_mom_x_lowvol, +} + + +class IndustryNeutralLSMomentum(Strategy): + """Industry-neutral long/short portfolio with selectable signal.""" + + def __init__( + self, + rebal_freq: int = 21, + mom_lookback: int = 252, + mom_skip: int = 21, + long_pct: float = 0.20, + short_pct: float = 0.20, + min_sector_size: int = 5, + sector_map: pd.Series | None = None, + gross_long: float = 1.0, + gross_short: float = 1.0, + signal_name: str = "mom_12_1", + ) -> None: + self.rebal_freq = rebal_freq + self.mom_lookback = mom_lookback + self.mom_skip = mom_skip + self.long_pct = long_pct + self.short_pct = short_pct + self.min_sector_size = min_sector_size + self.sector_map = sector_map + self.gross_long = gross_long + self.gross_short = gross_short + if signal_name not in SIGNAL_REGISTRY: + raise ValueError(f"Unknown signal: {signal_name}") + self.signal_name = signal_name + self.signal_func = SIGNAL_REGISTRY[signal_name] + + def _resolve_sector_map(self, columns: list[str]) -> pd.Series: + if self.sector_map is not None: + return self.sector_map.reindex(columns) + df = fetch_sp500_sectors() + s = df["GICS Sector"] + return s.reindex(columns) + + def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame: + cols = list(data.columns) + sectors = self._resolve_sector_map(cols) + mom = self.signal_func(data) + + weights = pd.DataFrame(0.0, index=data.index, columns=cols) + warmup = self.mom_lookback + 5 + + # Pre-compute which rows are rebal days + rebal_idx = list(range(warmup, len(data), self.rebal_freq)) + rebal_set = set(rebal_idx) + + # Group columns by sector + sector_to_cols: dict[str, list[str]] = {} + for c in cols: + s = sectors.get(c) + if pd.isna(s): + continue + sector_to_cols.setdefault(s, []).append(c) + + for t in rebal_idx: + row_mom = mom.iloc[t] + longs: dict[str, float] = {} + shorts: dict[str, float] = {} + + for sector, members in sector_to_cols.items(): + ms = row_mom.reindex(members).dropna() + if len(ms) < self.min_sector_size: + continue + n_long = max(1, int(round(len(ms) * self.long_pct))) + n_short = max(1, int(round(len(ms) * self.short_pct))) + ranked = ms.sort_values(ascending=False) + long_picks = ranked.head(n_long).index + short_picks = ranked.tail(n_short).index + for sym in long_picks: + longs[sym] = longs.get(sym, 0.0) + 1.0 + for sym in short_picks: + shorts[sym] = shorts.get(sym, 0.0) - 1.0 + + if not longs or not shorts: + continue + # Equal-weight within long leg and short leg + n_l = sum(longs.values()) + n_s = -sum(shorts.values()) + for sym in longs: + longs[sym] = self.gross_long * longs[sym] / n_l + for sym in shorts: + shorts[sym] = self.gross_short * shorts[sym] / n_s + + for sym, w in longs.items(): + weights.iat[t, cols.index(sym)] = w + for sym, w in shorts.items(): + weights.iat[t, cols.index(sym)] = w + + # Forward-fill between rebal dates + non_rebal_mask = pd.Series(True, index=data.index) + for i in rebal_idx: + non_rebal_mask.iat[i] = False + weights[non_rebal_mask.values] = np.nan + weights = weights.ffill().fillna(0.0) + weights.iloc[:warmup] = 0.0 + + return weights.shift(1).fillna(0.0) + + +__all__ = ["IndustryNeutralLSMomentum", "fetch_sp500_sectors"] diff --git a/strategies/permanent.py b/strategies/permanent.py new file mode 100644 index 0000000..0345416 --- /dev/null +++ b/strategies/permanent.py @@ -0,0 +1,762 @@ +"""Permanent Portfolio family — ported from usmart-quant TAA strategies. + +Three strategies, all operating on a small ETF universe (SPY, TQQQ, UPRO, +GLD, DBC, TLT, SHY). Each `generate_signals(data)` returns a weights +DataFrame already 1-day lagged (PIT-safe), columns must be a subset of +``data.columns``. + +* :class:`PermanentOverlay` — Browne's 25/25/25/25 with Faber MA200 + overlay on the stock slot. Bullish → TQQQ; bearish → cash. Source: + ``usmart-quant/strategies/taa_permanent_overlay.py``. +* :class:`TrendRiderV3` — risk-on/risk-off basket with momentum-ranked + pick, MA200 + vol/dd/peak gates, regime-min-hold + confirm + cooloff. + Source: ``usmart-quant/strategies/taa_trend_rider_v3.py``. +* :class:`PermanentV4` — improved Permanent. Stock slot picks the + momentum leader from (TQQQ, UPRO); bond slot rotates to SHY when TLT + is below its own MA200 (avoids 2022-style bond crashes); inflation + slot picks from (GLD, DBC). All four slots stay 25% — the same + diversification floor, but each slot self-rotates to its strongest + member. +""" +from __future__ import annotations + +import numpy as np +import pandas as pd + +from strategies.base import Strategy + + +# Universe of ETFs the strategies trade. The runner ensures these are +# present as columns in the price DataFrame. +ETF_UNIVERSE = ["SPY", "TQQQ", "UPRO", "GLD", "DBC", "TLT", "SHY"] + +TREND_RIDER_V4_UNIVERSE = [ + "SPY", "QQQ", + "SSO", "QLD", "UPRO", "TQQQ", + "SHY", "IEF", "TLT", + "GLD", "DBC", +] + +# Global expansion: USD-listed leveraged ETFs giving HK/China exposure. +# YINN — 3x FTSE China 50 (mostly HK-listed: Tencent, Meituan, Alibaba HK ADR) +# CHAU — 3x CSI 300 A-shares (mainland blue-chips traded SH/SZ) +# Both trade in USD so they compose cleanly with TQQQ/UPRO. Full Yahoo +# history: YINN since 2010, CHAU since 2015-04. +GLOBAL_ETF_UNIVERSE = ETF_UNIVERSE + ["YINN", "CHAU"] + +# HK-listed leveraged ETFs. Pure HK exposure (no proxy through ADRs): +# 7200.HK — HSI 2x (since 2017-03) +# 7500.HK — HSTECH 2x (since 2019-05) +# Note these trade in HKD; risk-off basket stays USD (GLD, DBC). Because +# HKD is pegged to USD (7.75–7.85), the FX drift over the test period is +# < 1% — acceptable as quasi-USD for this evaluation. +HK_ETF_UNIVERSE = ETF_UNIVERSE + ["7200.HK", "7500.HK"] + + +def _empty_weights(data: pd.DataFrame, cols: list[str]) -> pd.DataFrame: + return pd.DataFrame(0.0, index=data.index, columns=cols) + + +class PermanentOverlay(Strategy): + """Permanent Portfolio with Faber MA200 overlay on stock slot. + + 25% stock + 25% bonds + 25% gold + 25% cash. Stock slot holds TQQQ + when SPY > MA200 (PIT-lagged), else SHY (cash). Monthly rebalance. + """ + + def __init__( + self, + ma_window: int = 200, + rebal_every: int = 21, + signal: str = "SPY", + stock_on: str = "TQQQ", + stock_off: str = "SHY", + bonds: str = "TLT", + gold: str = "GLD", + cash: str = "SHY", + ) -> None: + self.ma_window = ma_window + self.rebal_every = rebal_every + self.signal = signal + self.stock_on = stock_on + self.stock_off = stock_off + self.bonds = bonds + self.gold = gold + self.cash = cash + + def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame: + cols = list(set([self.signal, self.stock_on, self.stock_off, + self.bonds, self.gold, self.cash])) + cols = [c for c in cols if c in data.columns] + w = pd.DataFrame(np.nan, index=data.index, columns=cols) + + spy = data[self.signal] + ma = spy.rolling(self.ma_window).mean() + bull = (spy > ma) + + for i, dt in enumerate(data.index): + if i < self.ma_window: + continue + if (i - self.ma_window) % self.rebal_every != 0: + continue + row = {c: 0.0 for c in cols} + if bull.iloc[i]: + row[self.stock_on] = row.get(self.stock_on, 0.0) + 0.25 + row[self.bonds] = row.get(self.bonds, 0.0) + 0.25 + row[self.gold] = row.get(self.gold, 0.0) + 0.25 + row[self.cash] = row.get(self.cash, 0.0) + 0.25 + else: + # Stock slot collapses into cash → effective 50% cash + row[self.bonds] = row.get(self.bonds, 0.0) + 0.25 + row[self.gold] = row.get(self.gold, 0.0) + 0.25 + row[self.cash] = row.get(self.cash, 0.0) + 0.50 + for s, ww in row.items(): + if s in w.columns: + w.at[dt, s] = ww + + # Forward-fill across non-rebal days (NaNs); fill warmup with 0. + w = w.ffill().fillna(0.0) + return w.shift(1).fillna(0.0) + + +class PermanentV4(Strategy): + """Improved Permanent — Faber filters on stock + bond + commodity basket. + + Slots (25% each): + stock: SPY > MA200 → max-momentum of (TQQQ, UPRO); else SHY + bond: TLT > MA200(TLT) → TLT; else SHY + gold: max-momentum of (GLD, DBC) over 63 days + cash: SHY (fixed) + + Three targeted upgrades over PermanentOverlay (which only filters + the stock slot): + + 1. Bond slot Faber filter solves 2022 (TLT −29% kills static + Permanent's bond sleeve). Vanilla PermanentOverlay was −20.7% + in 2022; adding the bond filter alone halves that. + 2. Stock slot picks momentum leader of (TQQQ, UPRO) — UPRO + substitutes when S&P leads QQQ (e.g. 2022 tech-led pullback). + 3. Inflation slot rotates between GLD and DBC. GLD captures + deflation/stagflation (2020); DBC captures commodity-driven + inflation (2022). Picking the leader avoids GLD's 2022 flat + year while still owning gold when it leads. + + Rebalance every 21 days. PIT-safe via terminal .shift(1). + """ + + def __init__( + self, + ma_window: int = 200, + mom_lookback: int = 63, + rebal_every: int = 21, + regime_signal: str = "SPY", + stock_basket: tuple[str, ...] = ("TQQQ", "UPRO"), + gold_basket: tuple[str, ...] = ("GLD", "DBC"), + bond: str = "TLT", + cash: str = "SHY", + ) -> None: + self.ma_window = ma_window + self.mom_lookback = mom_lookback + self.rebal_every = rebal_every + self.regime_signal = regime_signal + self.stock_basket = stock_basket + self.gold_basket = gold_basket + self.bond = bond + self.cash = cash + + def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame: + cols = list({self.regime_signal, *self.stock_basket, *self.gold_basket, + self.bond, self.cash}) + cols = [c for c in cols if c in data.columns] + w = pd.DataFrame(np.nan, index=data.index, columns=cols) + + spy = data[self.regime_signal] + spy_bull = spy > spy.rolling(self.ma_window).mean() + tlt_bull = data[self.bond] > data[self.bond].rolling(self.ma_window).mean() + mom = data.pct_change(self.mom_lookback) + + warmup = max(self.ma_window, self.mom_lookback) + for i, dt in enumerate(data.index): + if i < warmup: + continue + if (i - warmup) % self.rebal_every != 0: + continue + slots: dict[str, float] = {c: 0.0 for c in cols} + + # Stock slot + if spy_bull.iloc[i]: + pick, best = None, -np.inf + for s in self.stock_basket: + r = mom.at[dt, s] if s in mom.columns else np.nan + if pd.notna(r) and r > best: + best, pick = r, s + if pick is None: + pick = self.cash + else: + pick = self.cash + slots[pick] += 0.25 + + # Bond slot + slots[self.bond if tlt_bull.iloc[i] else self.cash] += 0.25 + + # Gold/commodity slot — basket leader by momentum (no MA filter: + # commodities are valuable diversifier even when not trending up) + pick, best = None, -np.inf + for s in self.gold_basket: + r = mom.at[dt, s] if s in mom.columns else np.nan + if pd.notna(r) and r > best: + best, pick = r, s + if pick is None: + pick = self.cash + slots[pick] += 0.25 + + slots[self.cash] += 0.25 + + for s, ww in slots.items(): + if s in w.columns: + w.at[dt, s] = ww + + w = w.ffill().fillna(0.0) + return w.shift(1).fillna(0.0) + + +class TrendRiderV3(Strategy): + """Risk-on / risk-off basket with momentum-ranked pick + regime gates. + + Faithful port of ``taa_trend_rider_v3.py`` with vol/MA/dd/peak + hysteresis, min-hold, confirm-days, entry stop-loss, and cooloff. + + Output is a single 100% allocation to whichever basket member is the + momentum leader at the current regime. PIT-safe (1-day signal lag). + """ + + DEFAULT_RISK_ON = ("TQQQ", "UPRO") + DEFAULT_RISK_OFF = ("GLD", "DBC") + + def __init__( + self, + signal: str = "SPY", + risk_on: tuple[str, ...] = DEFAULT_RISK_ON, + risk_off: tuple[str, ...] = DEFAULT_RISK_OFF, + ma_long: int = 200, + ma_short: int = 50, + vol_window: int = 20, + vol_enter: float = 0.14, + vol_exit: float = 0.20, + dd_window: int = 40, + dd_stop: float = 0.05, + peak_window: int = 20, + peak_enter: float = 0.02, + peak_exit: float = 0.05, + regime_min_hold: int = 15, + instrument_min_hold: int = 30, + confirm_days: int = 3, + stop_loss_pct: float = 0.15, + cooloff_days: int = 20, + mom_lookback: int = 63, + ) -> None: + self.signal = signal + self.risk_on = risk_on + self.risk_off = risk_off + self.ma_long = ma_long + self.ma_short = ma_short + self.vol_window = vol_window + self.vol_enter = vol_enter + self.vol_exit = vol_exit + self.dd_window = dd_window + self.dd_stop = dd_stop + self.peak_window = peak_window + self.peak_enter = peak_enter + self.peak_exit = peak_exit + self.regime_min_hold = regime_min_hold + self.instrument_min_hold = instrument_min_hold + self.confirm_days = confirm_days + self.stop_loss_pct = stop_loss_pct + self.cooloff_days = cooloff_days + self.mom_lookback = mom_lookback + + @staticmethod + def _above_ma(closes: np.ndarray, w: int) -> bool: + return closes.size >= w and float(closes[-1]) > float(closes[-w:].mean()) + + @staticmethod + def _vol(closes: np.ndarray, w: int) -> float: + if closes.size < w + 1: + return float("nan") + rets = np.diff(closes[-w - 1:]) / np.maximum(closes[-w - 1:-1], 1e-12) + return float(rets.std(ddof=1) * np.sqrt(252)) + + @staticmethod + def _total_return(closes: np.ndarray, w: int) -> float: + if closes.size < w + 1 or closes[-w - 1] <= 0: + return float("nan") + return float(closes[-1] / closes[-w - 1] - 1.0) + + def _desired_regime(self, closes: np.ndarray, current: str | None) -> str: + window_dd = closes[-self.dd_window:] + if closes[-1] / window_dd.max() - 1.0 <= -self.dd_stop: + return "risk_off" + if not self._above_ma(closes, self.ma_long): + return "risk_off" + v = self._vol(closes, self.vol_window) + if v != v: + v = 1.0 + peak_ratio = closes[-1] / closes[-self.peak_window:].max() + if current == "risk_on": + if (self._above_ma(closes, self.ma_short) + and v < self.vol_exit + and peak_ratio >= 1.0 - self.peak_exit): + return "risk_on" + return "risk_off" + if (self._above_ma(closes, self.ma_short) + and v < self.vol_enter + and peak_ratio >= 1.0 - self.peak_enter): + return "risk_on" + return "risk_off" + + def _pick_top(self, prices_t: np.ndarray, basket_idx: list[int], + closes_per_sym: dict[int, np.ndarray]) -> int | None: + best_i, best_r = None, -np.inf + for ix in basket_idx: + closes = closes_per_sym[ix] + r = self._total_return(closes, self.mom_lookback) + if r != r: + continue + if r > best_r: + best_r, best_i = r, ix + return best_i + + def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame: + cols = list({self.signal, *self.risk_on, *self.risk_off}) + cols = [c for c in cols if c in data.columns] + sym_to_ix = {s: ix for ix, s in enumerate(cols)} + w = _empty_weights(data, cols) + + if self.signal not in sym_to_ix: + return w.shift(1).fillna(0.0) + + sig_arr = data[self.signal].to_numpy() + # Per-symbol close arrays (for momentum pick) + sym_arrays = {sym_to_ix[s]: data[s].to_numpy() for s in cols} + + ron_idx = [sym_to_ix[s] for s in self.risk_on if s in sym_to_ix] + roff_idx = [sym_to_ix[s] for s in self.risk_off if s in sym_to_ix] + + need = max(self.ma_long, self.vol_window + 1, self.dd_window, + self.peak_window, self.mom_lookback + 1) + 1 + + current_regime: str | None = None + bars_in_regime = 0 + pending_regime: str | None = None + pending_count = 0 + current_sym: int | None = None + bars_in_sym = 0 + sym_entry_close: float | None = None + cooloff_remaining = 0 + + for t in range(len(data)): + if t < need: + continue + # Signal uses prices through t-1 (PIT lag) + sig_closes = sig_arr[: t] + if np.isnan(sig_closes[-1]): + continue + + desired = self._desired_regime(sig_closes, current_regime) + emergency = (sig_closes[-1] / sig_closes[-self.dd_window:].max() - 1.0) <= -self.dd_stop + + # Slice per-symbol closes through t-1 + cps = {ix: arr[:t] for ix, arr in sym_arrays.items()} + cur_close = float(sig_arr[t - 1]) if not np.isnan(sig_arr[t - 1]) else None + # ^ used only for stop-loss reference computation below + + def assign_one(sym_ix: int) -> None: + nonlocal current_sym, bars_in_sym, sym_entry_close + current_sym = sym_ix + bars_in_sym = 0 + # Entry "fill" reference is today's close (but recorded at decision) + p = float(sym_arrays[sym_ix][t]) if t < sym_arrays[sym_ix].size else float("nan") + sym_entry_close = p if not np.isnan(p) else float(sym_arrays[sym_ix][t - 1]) + + # First placement + if current_regime is None: + basket = ron_idx if desired == "risk_on" else roff_idx + pick = self._pick_top(None, basket, cps) + if pick is None: + continue + current_regime = desired + bars_in_regime = 0 + assign_one(pick) + w.iat[t, pick] = 1.0 + continue + + bars_in_regime += 1 + bars_in_sym += 1 + if cooloff_remaining > 0: + cooloff_remaining -= 1 + + in_on = current_regime == "risk_on" + sym_yclose = (float(sym_arrays[current_sym][t - 1]) + if current_sym is not None and not np.isnan(sym_arrays[current_sym][t - 1]) + else None) + + # Stop-loss + if (in_on and sym_yclose is not None and sym_entry_close + and sym_yclose / sym_entry_close - 1.0 <= -self.stop_loss_pct): + pick = self._pick_top(None, roff_idx, cps) + if pick is not None: + current_regime = "risk_off" + bars_in_regime = 0 + assign_one(pick) + pending_regime = None + pending_count = 0 + cooloff_remaining = self.cooloff_days + w.iat[t, pick] = 1.0 + continue + + # Emergency dd stop + if emergency and current_regime != "risk_off": + pick = self._pick_top(None, roff_idx, cps) + if pick is not None: + current_regime = "risk_off" + bars_in_regime = 0 + assign_one(pick) + pending_regime = None + pending_count = 0 + w.iat[t, pick] = 1.0 + continue + + # Regime change with confirm + min-hold + cooloff + if desired != current_regime: + if current_regime == "risk_off" and cooloff_remaining > 0: + pending_regime = None + pending_count = 0 + elif bars_in_regime < self.regime_min_hold: + pending_regime = None + pending_count = 0 + else: + if desired != pending_regime: + pending_regime = desired + pending_count = 1 + else: + pending_count += 1 + if pending_count >= self.confirm_days: + basket = ron_idx if desired == "risk_on" else roff_idx + pick = self._pick_top(None, basket, cps) + if pick is None: + pick = current_sym + current_regime = desired + bars_in_regime = 0 + assign_one(pick) + pending_regime = None + pending_count = 0 + w.iat[t, pick] = 1.0 + continue + # Hold prior allocation + if current_sym is not None: + w.iat[t, current_sym] = 1.0 + continue + + # Same regime — possibly rotate within basket + pending_regime = None + pending_count = 0 + basket = ron_idx if current_regime == "risk_on" else roff_idx + top = self._pick_top(None, basket, cps) + if top is None or top == current_sym: + if current_sym is not None: + w.iat[t, current_sym] = 1.0 + continue + if bars_in_sym < self.instrument_min_hold: + if current_sym is not None: + w.iat[t, current_sym] = 1.0 + continue + assign_one(top) + w.iat[t, top] = 1.0 + + return w.shift(1).fillna(0.0) + + +class TrendRiderV4(Strategy): + """Diversified TrendRider portfolio allocator. + + V3 is a single-instrument state machine. V4 keeps the same broad regime + idea, but allocates across sleeves: core equity, capped leveraged equity, + defensive bonds/cash, and inflation hedges. It is still PIT-safe through a + terminal ``shift(1)``. + """ + + def __init__( + self, + signal: str = "SPY", + core_equity: tuple[str, ...] = ("SPY", "QQQ"), + leveraged_equity: tuple[str, ...] = ("SSO", "QLD", "UPRO", "TQQQ"), + defensive: tuple[str, ...] = ("SHY", "IEF", "TLT"), + inflation: tuple[str, ...] = ("GLD", "DBC"), + ma_long: int = 200, + ma_short: int = 50, + vol_window: int = 20, + vol_enter: float = 0.14, + vol_exit: float = 0.20, + dd_window: int = 40, + dd_stop: float = 0.05, + peak_window: int = 20, + peak_enter: float = 0.02, + peak_exit: float = 0.05, + regime_min_hold: int = 15, + confirm_days: int = 3, + mom_lookback: int = 63, + rebal_every: int = 21, + max_single_weight: float = 0.45, + max_leveraged_weight: float = 0.90, + risk_on_targets: tuple[float, float, float, float] = (0.10, 0.85, 0.00, 0.05), + risk_off_targets: tuple[float, float, float, float] = (0.30, 0.00, 0.50, 0.20), + ) -> None: + self.signal = signal + self.core_equity = core_equity + self.leveraged_equity = leveraged_equity + self.defensive = defensive + self.inflation = inflation + self.ma_long = ma_long + self.ma_short = ma_short + self.vol_window = vol_window + self.vol_enter = vol_enter + self.vol_exit = vol_exit + self.dd_window = dd_window + self.dd_stop = dd_stop + self.peak_window = peak_window + self.peak_enter = peak_enter + self.peak_exit = peak_exit + self.regime_min_hold = regime_min_hold + self.confirm_days = confirm_days + self.mom_lookback = mom_lookback + self.rebal_every = rebal_every + self.max_single_weight = max_single_weight + self.max_leveraged_weight = max_leveraged_weight + self.risk_on_targets = risk_on_targets + self.risk_off_targets = risk_off_targets + + def _desired_regime(self, closes: np.ndarray, current: str | None) -> str: + return TrendRiderV3( + signal=self.signal, + ma_long=self.ma_long, + ma_short=self.ma_short, + vol_window=self.vol_window, + vol_enter=self.vol_enter, + vol_exit=self.vol_exit, + dd_window=self.dd_window, + dd_stop=self.dd_stop, + peak_window=self.peak_window, + peak_enter=self.peak_enter, + peak_exit=self.peak_exit, + )._desired_regime(closes, current) + + def _sleeve_weights( + self, + amount: float, + basket: tuple[str, ...], + cols: list[str], + mom_row: pd.Series, + vol_row: pd.Series, + top_n: int, + require_positive: bool = False, + ) -> dict[str, float]: + if amount <= 0: + return {} + candidates = [] + for sym in basket: + if sym not in cols or sym not in mom_row.index: + continue + mom = float(mom_row.get(sym, np.nan)) + if not np.isfinite(mom): + continue + if require_positive and mom <= 0: + continue + vol = float(vol_row.get(sym, np.nan)) + if not np.isfinite(vol) or vol <= 0: + vol = 0.20 + candidates.append((sym, mom, max(vol, 0.05))) + if not candidates: + return {} + + candidates.sort(key=lambda item: item[1], reverse=True) + selected = candidates[:max(1, top_n)] + inv_vol = np.array([1.0 / item[2] for item in selected], dtype=float) + inv_vol = inv_vol / inv_vol.sum() + return {sym: float(amount * weight) for (sym, _, _), weight in zip(selected, inv_vol)} + + def _redistribute(self, row: dict[str, float], excess: float, + preferred: list[str]) -> float: + remaining = excess + for sym in preferred: + if remaining <= 1e-12: + break + if sym not in row: + continue + spare = max(self.max_single_weight - row.get(sym, 0.0), 0.0) + add = min(spare, remaining) + row[sym] = row.get(sym, 0.0) + add + remaining -= add + return remaining + + def _apply_caps(self, row: dict[str, float], cols: list[str]) -> dict[str, float]: + row = {sym: float(weight) for sym, weight in row.items() if sym in cols and weight > 1e-12} + for sym in cols: + row.setdefault(sym, 0.0) + + leveraged = [sym for sym in self.leveraged_equity if sym in row] + lev_total = sum(row[sym] for sym in leveraged) + excess = 0.0 + if lev_total > self.max_leveraged_weight and lev_total > 0: + scale = self.max_leveraged_weight / lev_total + for sym in leveraged: + old = row[sym] + row[sym] = old * scale + excess += old - row[sym] + + preferred = [*self.defensive, *self.inflation, *self.core_equity] + if excess > 1e-12: + excess = self._redistribute(row, excess, preferred) + + for _ in range(len(row) + 1): + over = [sym for sym, weight in row.items() if weight > self.max_single_weight] + if not over: + break + for sym in over: + excess += row[sym] - self.max_single_weight + row[sym] = self.max_single_weight + excess = self._redistribute(row, excess, preferred) + if excess <= 1e-12: + break + + if excess > 1e-12: + receivers = [sym for sym in row if row[sym] < self.max_single_weight - 1e-12] + spare = sum(self.max_single_weight - row[sym] for sym in receivers) + if spare > 0: + for sym in receivers: + add = excess * (self.max_single_weight - row[sym]) / spare + row[sym] += add + excess = 0.0 + + total = sum(row.values()) + if total > 0: + row = {sym: weight / total for sym, weight in row.items()} + return {sym: weight for sym, weight in row.items() if weight > 1e-10} + + def _allocate(self, regime: str, cols: list[str], + mom_row: pd.Series, vol_row: pd.Series) -> dict[str, float]: + if regime == "risk_on": + core, leveraged, defensive, inflation = self.risk_on_targets + sleeve_targets = { + "core": core, + "leveraged": leveraged, + "defensive": defensive, + "inflation": inflation, + } + else: + core, leveraged, defensive, inflation = self.risk_off_targets + sleeve_targets = { + "core": core, + "leveraged": leveraged, + "defensive": defensive, + "inflation": inflation, + } + + row: dict[str, float] = {sym: 0.0 for sym in cols} + sleeves = [ + (sleeve_targets["core"], self.core_equity, 2, False), + (sleeve_targets["leveraged"], self.leveraged_equity, 2, True), + (sleeve_targets["defensive"], self.defensive, 2, False), + (sleeve_targets["inflation"], self.inflation, 2, False), + ] + unallocated = 0.0 + for amount, basket, top_n, require_positive in sleeves: + alloc = self._sleeve_weights(amount, basket, cols, mom_row, vol_row, top_n, require_positive) + if not alloc: + unallocated += amount + continue + for sym, weight in alloc.items(): + row[sym] += weight + + if unallocated > 0: + fallback = next((sym for sym in self.defensive if sym in cols), None) + if fallback is not None: + row[fallback] += unallocated + + return self._apply_caps(row, cols) + + def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame: + cols = list({ + self.signal, + *self.core_equity, + *self.leveraged_equity, + *self.defensive, + *self.inflation, + }) + cols = [c for c in cols if c in data.columns] + w = pd.DataFrame(np.nan, index=data.index, columns=cols) + + if self.signal not in data.columns: + return _empty_weights(data, cols).shift(1).fillna(0.0) + + signal_arr = data[self.signal].to_numpy() + returns = data[cols].pct_change(fill_method=None) + momentum = data[cols].pct_change(self.mom_lookback, fill_method=None) + vol = returns.rolling(self.vol_window).std() * np.sqrt(252) + need = max(self.ma_long, self.vol_window + 1, self.dd_window, + self.peak_window, self.mom_lookback + 1) + + current_regime: str | None = None + bars_in_regime = 0 + pending_regime: str | None = None + pending_count = 0 + + for i, dt in enumerate(data.index): + if i < need: + continue + closes = signal_arr[: i + 1] + if np.isnan(closes[-1]): + continue + + desired = self._desired_regime(closes, current_regime) + regime_changed = False + if current_regime is None: + current_regime = desired + bars_in_regime = 0 + regime_changed = True + else: + bars_in_regime += 1 + if desired != current_regime: + if bars_in_regime >= self.regime_min_hold: + if desired != pending_regime: + pending_regime = desired + pending_count = 1 + else: + pending_count += 1 + if pending_count >= self.confirm_days: + current_regime = desired + bars_in_regime = 0 + pending_regime = None + pending_count = 0 + regime_changed = True + else: + pending_regime = None + pending_count = 0 + else: + pending_regime = None + pending_count = 0 + + if not regime_changed and (i - need) % self.rebal_every != 0: + continue + + row = self._allocate( + current_regime, + cols, + momentum.iloc[i], + vol.iloc[i], + ) + w.loc[dt, cols] = 0.0 + for sym, weight in row.items(): + w.at[dt, sym] = weight + + w = w.ffill().fillna(0.0) + return w.shift(1).fillna(0.0) diff --git a/strategies/trend_rider_v5.py b/strategies/trend_rider_v5.py new file mode 100644 index 0000000..91d565a --- /dev/null +++ b/strategies/trend_rider_v5.py @@ -0,0 +1,372 @@ +"""TrendRiderV5 — V3 with conviction-gated leverage tier modulation. + +Design rationale +---------------- +V3 picks one of {TQQQ, UPRO, GLD, DBC} and rides it 100%. Its 75 regime +switches over 11 years are the *correct* edge — we don't disturb them. + +V5 layers a small post-processor: at each rebalance event V3 produces, V5 +inspects the prevailing conviction and decides what fraction of the equity +sleeve is held in the 3× ETF vs its 1× counterpart. The state is a discrete +*leverage tier* in {0%, 50%, 100%} of leveraged exposure, with hysteresis +and minimum holding to keep turnover low. Specifically + + pair: SPY ↔ UPRO, QQQ ↔ TQQQ + tier 0 (core_only) : 100% core (1× equity) + tier 1 (half) : 50% core + 50% leveraged (≈ 2× equity) + tier 2 (full) : 100% leveraged (3× equity) + +Conviction is built from directional/regime-quality signals (trend strength, +drawdown depth, peak distance, downside-vol percentile). It is NOT a function +of two-sided realized vol — that throttled V5 in good periods. Tier +transitions require: + + promote (k → k+1) : conviction ≥ promote_threshold[k+1] for confirm_days + demote (k → k-1) : conviction ≤ demote_threshold[k] for demote_confirm + +with `tier_min_hold` bars between any tier change. + +Risk-off behavior is unchanged from V3 (single-pick momentum leader of the +risk_off basket), preserving V3's defensive characteristics. +""" +from __future__ import annotations + +import numpy as np +import pandas as pd + +from strategies.permanent import TrendRiderV3 + + +class TrendRiderV5(TrendRiderV3): + """V3 + leverage-tier modulator on the equity sleeve. + + Default tier thresholds aim for: full 3× only when (a) below-MA200 risk + is small, (b) we're near the 20-day high, and (c) drawdowns from the + recent peak are inside ~1 vol-unit. Otherwise step down to 1× or 1.5×. + """ + + DEFAULT_LEVERAGED_PAIR = {"SPY": "UPRO", "QQQ": "TQQQ"} + DEFAULT_CORE_PAIR = {"UPRO": "SPY", "TQQQ": "QQQ"} + + def __init__( + self, + *args, + # Conviction inputs + peak_window: int = 20, + dd_window: int = 40, + trend_lookback: int = 63, + downvol_window: int = 60, + downvol_lookback: int = 252, + # Tier thresholds [tier 1, tier 2] for promote / demote (hysteresis) + promote_thresholds: tuple[float, float] = (0.40, 0.65), + demote_thresholds: tuple[float, float] = (0.30, 0.50), + promote_confirm: int = 5, + demote_confirm: int = 3, + tier_min_hold: int = 10, + starting_tier: int = 2, # if regime is risk_on at first placement, start at 2 (full lev) + # Panic demote — bypasses min-hold when fast vol regime detected. + # Defaults below were chosen by walk-forward Calmar maximization on + # IS (2015-2020, which does NOT contain the 2024-08 crash) — not + # curve-fit to that specific event. + panic_vol_short: int = 7, + panic_vol_long: int = 60, + panic_vol_ratio: float = 1.6, + panic_peak_drop_pct: float = 0.06, + panic_peak_window: int = 5, + # Conviction component weights + w_trend: float = 0.30, + w_dd: float = 0.30, + w_peak: float = 0.25, + w_downvol: float = 0.15, + # Pair mapping + leveraged_pair: dict[str, str] | None = None, + core_pair: dict[str, str] | None = None, + **kwargs, + ) -> None: + super().__init__(*args, **kwargs) + self.peak_window = peak_window + self.dd_window = dd_window + self.trend_lookback = trend_lookback + self.downvol_window = downvol_window + self.downvol_lookback = downvol_lookback + self.promote_thresholds = promote_thresholds + self.demote_thresholds = demote_thresholds + self.promote_confirm = promote_confirm + self.demote_confirm = demote_confirm + self.tier_min_hold = tier_min_hold + self.starting_tier = starting_tier + self.panic_vol_short = panic_vol_short + self.panic_vol_long = panic_vol_long + self.panic_vol_ratio = panic_vol_ratio + self.panic_peak_drop_pct = panic_peak_drop_pct + self.panic_peak_window = panic_peak_window + self.w_trend = w_trend + self.w_dd = w_dd + self.w_peak = w_peak + self.w_downvol = w_downvol + self.leveraged_pair = leveraged_pair or dict(self.DEFAULT_LEVERAGED_PAIR) + self.core_pair = core_pair or dict(self.DEFAULT_CORE_PAIR) + + # ---- Conviction features ---- + @staticmethod + def _clip01(x: float) -> float: + if not np.isfinite(x): + return 0.0 + return float(min(1.0, max(0.0, x))) + + def _panic_demote(self, sig_closes: np.ndarray) -> bool: + """Detect fast vol regime / sharp peak velocity → panic demote tier 2→0.""" + if sig_closes.size < self.panic_vol_long + 1: + return False + # Short vs long realized vol + rets = np.diff(sig_closes[-(self.panic_vol_long + 1):]) / np.maximum( + sig_closes[-(self.panic_vol_long + 1):-1], 1e-12 + ) + if rets.size < self.panic_vol_long: + return False + long_vol = float(rets.std(ddof=1)) + short_rets = rets[-self.panic_vol_short:] + short_vol = float(short_rets.std(ddof=1)) if short_rets.size > 1 else 0.0 + if long_vol > 0 and short_vol / long_vol >= self.panic_vol_ratio: + return True + # Peak-velocity: drop > X% in last N days from rolling peak + window = sig_closes[-self.panic_peak_window:] + if window.size >= 2: + peak = float(window.max()) + drop = (peak - float(sig_closes[-1])) / max(peak, 1e-12) + if drop >= self.panic_peak_drop_pct: + return True + return False + + def _conviction(self, sig_closes: np.ndarray) -> float: + """Directional conviction in [0, 1] — higher means cleaner trend.""" + n = sig_closes.size + if n < max(self.ma_long, self.trend_lookback, + self.downvol_lookback + self.downvol_window) + 1: + return 0.0 + + last = float(sig_closes[-1]) + + # 1) Trend score: distance above MA200 in vol-units + ma_long = float(sig_closes[-self.ma_long:].mean()) + rets = np.diff(sig_closes[-self.downvol_window - 1:]) / np.maximum( + sig_closes[-self.downvol_window - 1:-1], 1e-12 + ) + ann_vol = float(rets.std(ddof=1) * np.sqrt(252)) if rets.size > 1 else 0.20 + ann_vol = max(ann_vol, 1e-3) + trend_units = (last / ma_long - 1.0) / ann_vol # vol-units (annualized) + trend_score = self._clip01(trend_units / 0.50) # ~0.50 vol-unit = strong + + # 2) Drawdown score: shallower = better + dd_window_arr = sig_closes[-self.dd_window:] + dd = float(last / dd_window_arr.max() - 1.0) # ≤ 0 + period_vol = ann_vol / np.sqrt(252) * np.sqrt(self.dd_window) + dd_units = -dd / max(period_vol, 1e-4) + dd_score = self._clip01(1.0 - dd_units / 2.5) # 2.5 vol-units → 0 + + # 3) Peak-distance score + peak_arr = sig_closes[-self.peak_window:] + peak_ratio = float(last / peak_arr.max()) + peak_period_vol = ann_vol / np.sqrt(252) * np.sqrt(self.peak_window) + peak_drop_units = (1.0 - peak_ratio) / max(peak_period_vol, 1e-4) + peak_score = self._clip01(1.0 - peak_drop_units / 2.0) + + # 4) Downside-vol percentile (lower = better) + full_rets = np.diff(sig_closes[-(self.downvol_lookback + self.downvol_window):]) / np.maximum( + sig_closes[-(self.downvol_lookback + self.downvol_window):-1], 1e-12 + ) + # Rolling downside semideviation + s = pd.Series(full_rets) + downside = s.where(s < 0, 0.0) + dv_series = downside.rolling(self.downvol_window).std(ddof=1) * np.sqrt(252) + dv_now = float(dv_series.iloc[-1]) if not dv_series.empty else np.nan + dv_history = dv_series.dropna().to_numpy() + if dv_history.size == 0 or not np.isfinite(dv_now): + downvol_score = 0.5 + else: + pct = float((dv_history < dv_now).mean()) + downvol_score = 1.0 - pct # low downvol → high score + + score = ( + self.w_trend * trend_score + + self.w_dd * dd_score + + self.w_peak * peak_score + + self.w_downvol * downvol_score + ) + total_w = self.w_trend + self.w_dd + self.w_peak + self.w_downvol + return float(score / max(total_w, 1e-9)) + + # ---- Tier state ---- + def _tier_for(self, conviction: float, current: int, + pending_promote: int, pending_demote: int) -> tuple[int, int, int]: + """Update tier given conviction. Returns (new_tier, new_pp, new_pd).""" + new_tier = current + # Demote first (safety > greed) + if current >= 1 and conviction <= self.demote_thresholds[current - 1]: + pending_demote += 1 + pending_promote = 0 + if pending_demote >= self.demote_confirm: + new_tier = max(0, current - 1) + pending_demote = 0 + return new_tier, pending_promote, pending_demote + else: + pending_demote = 0 + + # Promote + target = current + if current < 2 and conviction >= self.promote_thresholds[current]: + pending_promote += 1 + if pending_promote >= self.promote_confirm: + target = min(2, current + 1) + pending_promote = 0 + else: + pending_promote = 0 + + return target, pending_promote, pending_demote + + def _equity_blend(self, sym: str, tier: int, cols: list[str]) -> dict[str, float]: + """Blend a chosen symbol with its leveraged/core counterpart by tier.""" + # If V3 picked a leveraged sym (TQQQ/UPRO), map to core counterpart + if sym in self.core_pair: + lev_sym = sym + core_sym = self.core_pair[sym] + elif sym in self.leveraged_pair: + core_sym = sym + lev_sym = self.leveraged_pair[sym] + else: + # No leveraged variant available → 100% as-is + return {sym: 1.0} + + if core_sym not in cols and lev_sym not in cols: + return {sym: 1.0} + if core_sym not in cols: + return {lev_sym: 1.0} + if lev_sym not in cols: + return {core_sym: 1.0} + + if tier == 0: + return {core_sym: 1.0} + if tier == 1: + return {core_sym: 0.5, lev_sym: 0.5} + return {lev_sym: 1.0} + + # ---- Override: post-process V3 weights ---- + def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame: + # 1) Get V3's PIT-safe weights (already shifted) + v3_w = super().generate_signals(data) + + # We need to "un-shift" V3 weights to align with the day they were decided, + # apply tier blending in that frame, then re-shift. Easier: work directly + # in the signal frame (which is v3_w's index, with row t = position for t). + # Since super() already shifted by 1, v3_w.iloc[t] is the *position* held + # on day t (decided on close of t-1). We modulate row-by-row. + + sig = data[self.signal] if self.signal in data.columns else None + if sig is None: + return v3_w + + sig_arr = sig.to_numpy() + cols = list(v3_w.columns) + # Make sure leveraged/core counterparts exist as columns; expand if not + extra_cols = [] + for sym in (*self.core_pair.keys(), *self.leveraged_pair.keys()): + if sym in data.columns and sym not in cols: + extra_cols.append(sym) + if extra_cols: + for c in extra_cols: + v3_w[c] = 0.0 + cols = list(v3_w.columns) + + out = pd.DataFrame(0.0, index=v3_w.index, columns=cols) + + # Tier state + tier = 0 # start at 0 — promotions happen via confirm + pending_promote = 0 + pending_demote = 0 + tier_age = 0 + prev_active_sym: str | None = None + first_risk_on_seen = False + + for t in range(len(v3_w)): + row = v3_w.iloc[t] + active = row[row > 0] + if active.empty: + # No position → no modulation + tier = 0 + pending_promote = pending_demote = 0 + tier_age = 0 + prev_active_sym = None + continue + + sym = active.idxmax() # V3 outputs 100% to one symbol + # Compute conviction from signal closes through t-1 (already PIT) + # v3_w.iloc[t] reflects position decided on close(t-1), so we can + # use sig_arr[:t] as available info. + sig_closes = sig_arr[: t] + if sig_closes.size == 0: + continue + conviction = self._conviction(sig_closes) + + # Detect new active position + is_equity = sym in self.core_pair or sym in self.leveraged_pair + if not is_equity: + # Risk-off: pass through, reset tier state + tier = 0 + pending_promote = pending_demote = 0 + tier_age = 0 + prev_active_sym = sym + out.iloc[t] = row + continue + + if prev_active_sym != sym: + # Fresh entry into equity sleeve + if not first_risk_on_seen: + tier = self.starting_tier + first_risk_on_seen = True + else: + # Initialize tier from current conviction + if conviction >= self.promote_thresholds[1]: + tier = 2 + elif conviction >= self.promote_thresholds[0]: + tier = 1 + else: + tier = 0 + pending_promote = pending_demote = 0 + tier_age = 0 + + # Panic demote — bypasses min-hold and conviction logic + panic = self._panic_demote(sig_closes) + if panic and tier > 0: + tier = 0 + tier_age = 0 + pending_promote = pending_demote = 0 + else: + # Tier transition logic with min-hold + new_tier = tier + if tier_age >= self.tier_min_hold: + new_tier, pending_promote, pending_demote = self._tier_for( + conviction, tier, pending_promote, pending_demote + ) + if new_tier != tier: + tier_age = 0 + tier = new_tier + else: + tier_age += 1 + else: + tier_age += 1 + # Even within min-hold, allow emergency demote if conviction crashes + if tier > 0 and conviction <= self.demote_thresholds[tier - 1] * 0.6: + tier = max(0, tier - 1) + tier_age = 0 + pending_promote = pending_demote = 0 + + # Blend + blend = self._equity_blend(sym, tier, cols) + for s, ww in blend.items(): + out.at[v3_w.index[t], s] = ww + prev_active_sym = sym + + return out + + +__all__ = ["TrendRiderV5"] diff --git a/strategies/trend_rider_v6.py b/strategies/trend_rider_v6.py new file mode 100644 index 0000000..cc779ee --- /dev/null +++ b/strategies/trend_rider_v6.py @@ -0,0 +1,304 @@ +"""TrendRiderV6 — V5 regime engine on top of a stock-picking sleeve. + +Goal +---- +Lift portfolio Sharpe from V5's ~1.10 to ≥ 1.50 by replacing the +single-instrument leveraged ETF (TQQQ/UPRO) with a diversified +top-N stock momentum portfolio (≈ 10–20 names, inverse-volatility +weighted, monthly rebalanced) — wrapped in V5's regime / panic / +tier state machine. + +Why diversified stocks instead of TQQQ? +-------------------------------------- +TQQQ is a single instrument with ~70% annualized vol and idiosyncratic +NDX path dependence. Even with perfect timing, its Sharpe is bounded +by the underlying. A 10–20 stock momentum portfolio has comparable or +higher mean return (factor literature: cross-sectional momentum + +recovery have meaningful IC) but substantially lower vol due to +diversification, lifting Sharpe. + +Architecture +------------ +Three sleeves, gated by V5's tier state: + + tier 2 (high conviction) : 100% stock momentum portfolio + (top_n stocks, inv-vol weighted) + tier 1 (moderate) : 50% stock portfolio + 50% SPY + tier 0 (defensive) : inv-vol risk_off basket (SHY+GLD+DBC) + +Tier transitions, panic demote, conviction signals, and regime FSM +are all inherited from V5's machinery, applied to the SPY signal. + +The strategy expects a price panel containing both stocks AND the +required ETFs: at minimum {SPY, SHY, GLD, DBC} for non-stock sleeves, +plus enough stocks for a meaningful top_n selection. +""" +from __future__ import annotations + +import numpy as np +import pandas as pd + +from strategies.permanent import TrendRiderV3 +from strategies.trend_rider_v5 import TrendRiderV5 +from strategies.factor_combo import SIGNAL_REGISTRY + + +class TrendRiderV6(TrendRiderV5): + """Stock-sleeve TrendRider with V5 regime engine.""" + + def __init__( + self, + *args, + # Stock selection + signal_name: str = "rec_mfilt+deep_upvol", + top_n: int = 15, + rebal_freq: int = 21, + stock_universe: list[str] | None = None, + risk_off_basket: tuple[str, ...] = ("GLD", "DBC"), # V3-style single-pick + moderate_anchor: str = "SPY", + # Tier-2 leverage overlay (0.0 = pure stocks; 0.3 = 70% stocks + 30% TQQQ) + tier2_leverage_overlay: float = 0.0, + leverage_overlay_symbol: str = "TQQQ", + # Mode: "blend" (default) → tier1=mixed; "regime" → tier1=stocks, tier2=TQQQ + tier_mode: str = "blend", + # Inv-vol weighting parameters + invvol_window: int = 60, + invvol_floor: float = 0.10, + invvol_cap: float = 0.20, + **kwargs, + ) -> None: + super().__init__(*args, **kwargs) + if signal_name not in SIGNAL_REGISTRY: + raise ValueError(f"Unknown signal: {signal_name}. " + f"Available: {list(SIGNAL_REGISTRY.keys())}") + self.signal_name = signal_name + self.signal_func = SIGNAL_REGISTRY[signal_name] + self.top_n = top_n + self.rebal_freq = rebal_freq + self.stock_universe = stock_universe + self.risk_off_basket = risk_off_basket + self.moderate_anchor = moderate_anchor + self.tier2_leverage_overlay = tier2_leverage_overlay + self.leverage_overlay_symbol = leverage_overlay_symbol + self.tier_mode = tier_mode + self.invvol_window = invvol_window + self.invvol_floor = invvol_floor + self.invvol_cap = invvol_cap + + # ---- Helpers ---- + def _resolve_universe(self, prices: pd.DataFrame) -> list[str]: + if self.stock_universe is not None: + return [s for s in self.stock_universe if s in prices.columns] + # Heuristic: stocks are columns NOT in our known ETF/leveraged set + non_stock = (set(self.core_equity) + | set(self.leveraged_equity) + | set(self.risk_off) + | {self.signal, *self.risk_off_basket, self.moderate_anchor}) + return [c for c in prices.columns if c not in non_stock] + + def _stock_top_n_weights(self, prices: pd.DataFrame, universe: list[str]) -> pd.DataFrame: + """Top-N selection by signal, inv-vol weighted within selection.""" + stock_panel = prices[universe] + sig = self.signal_func(stock_panel) + # Top-N by signal rank (highest score = top) + rank = sig.rank(axis=1, ascending=False, na_option="bottom") + n_valid = sig.notna().sum(axis=1) + enough = n_valid >= self.top_n + top_mask = (rank <= self.top_n) & enough.values.reshape(-1, 1) + + # Inv-vol within the selection + rets = stock_panel.pct_change(fill_method=None) + vol = rets.rolling(self.invvol_window, min_periods=self.invvol_window // 2).std() * np.sqrt(252) + vol_clipped = vol.clip(lower=self.invvol_floor, upper=self.invvol_cap) + invvol = (1.0 / vol_clipped).where(top_mask, 0.0) + + row_sums = invvol.sum(axis=1).replace(0, np.nan) + w = invvol.div(row_sums, axis=0).fillna(0.0) + + # Monthly rebalance + warmup = 252 + rebal_mask = pd.Series(False, index=prices.index) + rebal_indices = list(range(warmup, len(prices), self.rebal_freq)) + rebal_mask.iloc[rebal_indices] = True + w[~rebal_mask] = np.nan + w = w.ffill().fillna(0.0) + w.iloc[:warmup] = 0.0 + return w # Note: NOT shifted yet — caller shifts at the end + + def _risk_off_pick(self, prices: pd.DataFrame, t: int) -> dict[str, float]: + """V3-style single-pick: highest 63d momentum within risk_off basket. + + Single-pick captures the leader (e.g. DBC in 2022 +21%, GLD in 2020), + whereas inv-vol weighting drags the upside down with low-vol SHY. + """ + cols = [c for c in self.risk_off_basket if c in prices.columns] + if not cols: + return {} + best, best_r = None, -np.inf + lookback = self.mom_lookback + for c in cols: + arr = prices[c].to_numpy() + if t < lookback + 1 or t >= arr.size or arr[t - lookback] <= 0 or np.isnan(arr[t]): + continue + r = float(arr[t] / arr[t - lookback] - 1.0) + if np.isfinite(r) and r > best_r: + best_r, best = r, c + if best is None: + # fallback to first available + for c in cols: + if c in prices.columns: + return {c: 1.0} + return {} + return {best: 1.0} + + # ---- Override generate_signals ---- + def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame: + if self.signal not in data.columns: + raise ValueError(f"Required regime signal {self.signal!r} not in data.") + universe = self._resolve_universe(data) + if len(universe) < self.top_n: + raise ValueError(f"Stock universe ({len(universe)}) smaller than top_n ({self.top_n}).") + + # 1) Build sleeve weights — stock sleeve, anchor sleeve + # (defensive sleeve is single-pick, computed per-bar inside the loop) + stock_w = self._stock_top_n_weights(data, universe) + anchor_w = pd.DataFrame(0.0, index=data.index, columns=[self.moderate_anchor]) + if self.moderate_anchor in data.columns: + anchor_w[self.moderate_anchor] = 1.0 + + # 2) Run V3-style regime FSM + V5 panic + tier state machine on signal + sig_arr = data[self.signal].to_numpy() + out = pd.DataFrame(0.0, index=data.index, columns=data.columns) + + current_regime: str | None = None + bars_in_regime = 0 + pending_regime: str | None = None + pending_count = 0 + cooloff_remaining = 0 + tier = self.starting_tier + tier_age = 0 + pending_promote = 0 + pending_demote = 0 + + need = max(self.ma_long, self.dd_window, self.peak_window, + self.downvol_lookback + self.downvol_window, + self.trend_lookback, 252) + 1 + + for t in range(len(data)): + if t < need: + continue + sig_closes = sig_arr[: t] + if np.isnan(sig_closes[-1]): + continue + + # Use V3's regime decision (uses self.dd_stop, vol_enter/exit, peak_enter/exit) + desired = self._desired_regime(sig_closes, current_regime) + + if cooloff_remaining > 0: + cooloff_remaining -= 1 + + if current_regime is None: + current_regime = desired + bars_in_regime = 0 + + bars_in_regime += 1 + + if desired != current_regime: + if current_regime == "risk_off" and cooloff_remaining > 0: + pending_regime, pending_count = None, 0 + elif bars_in_regime < self.regime_min_hold: + pending_regime, pending_count = None, 0 + else: + if desired != pending_regime: + pending_regime, pending_count = desired, 1 + else: + pending_count += 1 + if pending_count >= self.confirm_days: + current_regime = desired + bars_in_regime = 0 + pending_regime, pending_count = None, 0 + if current_regime == "risk_off": + cooloff_remaining = self.cooloff_days + else: + pending_regime, pending_count = None, 0 + + # --- Conviction + tier --- + conviction = self._conviction(sig_closes) + panic = self._panic_demote(sig_closes) + + if current_regime == "risk_off": + tier = 0 + tier_age = 0 + pending_promote = pending_demote = 0 + else: + if panic and tier > 0: + tier = 0 + tier_age = 0 + pending_promote = pending_demote = 0 + elif tier_age >= self.tier_min_hold: + new_tier, pending_promote, pending_demote = self._tier_for( + conviction, tier, pending_promote, pending_demote + ) + if new_tier != tier: + tier = new_tier + tier_age = 0 + else: + tier_age += 1 + else: + tier_age += 1 + if tier > 0 and conviction <= self.demote_thresholds[tier - 1] * 0.6: + tier = max(0, tier - 1) + tier_age = 0 + pending_promote = pending_demote = 0 + + # --- Apply tier to sleeve weights (in the position frame) --- + row = pd.Series(0.0, index=data.columns) + if tier == 0: + pick = self._risk_off_pick(data, t) + for c, v in pick.items(): + row[c] = v + elif self.tier_mode == "regime": + # Regime mode: tier 1 = pure stocks (medium conviction); + # tier 2 = pure TQQQ leverage (high conviction, clean trend) + if tier == 1: + for c, v in stock_w.iloc[t].items(): + if v > 0: + row[c] = row.get(c, 0.0) + v + else: # tier 2 + if self.leverage_overlay_symbol in data.columns: + row[self.leverage_overlay_symbol] = 1.0 + else: + for c, v in stock_w.iloc[t].items(): + if v > 0: + row[c] = row.get(c, 0.0) + v + else: + # Blend mode (original V6) + if tier == 1: + stock_row = stock_w.iloc[t] * 0.5 + anchor_row = anchor_w.iloc[t] * 0.5 + for c, v in stock_row.items(): + if v > 0: + row[c] = row.get(c, 0.0) + v + for c, v in anchor_row.items(): + if v > 0: + row[c] = row.get(c, 0.0) + v + else: # tier 2 + ov = float(self.tier2_leverage_overlay) + if ov > 0 and self.leverage_overlay_symbol in data.columns: + stock_row = stock_w.iloc[t] * (1.0 - ov) + for c, v in stock_row.items(): + if v > 0: + row[c] = row.get(c, 0.0) + v + row[self.leverage_overlay_symbol] = ( + row.get(self.leverage_overlay_symbol, 0.0) + ov + ) + else: + for c, v in stock_w.iloc[t].items(): + if v > 0: + row[c] = row.get(c, 0.0) + v + out.iloc[t] = row.values + + return out.shift(1).fillna(0.0) + + +__all__ = ["TrendRiderV6"]