"""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. # We inherit V3's risk_on (e.g. TQQQ/UPRO) and risk_off (GLD/DBC), # plus V6's risk_off_basket + moderate_anchor + signal + overlay sym. non_stock = (set(self.risk_on) | set(self.risk_off) | {self.signal, self.moderate_anchor, self.leverage_overlay_symbol, *self.risk_off_basket}) 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"]