"""TrendRider V7 — V3 + vol-target + profit-take on leveraged ETFs. Architecture ------------ Three sequential layers, each PIT-safe: Layer 1 — TrendRiderV3 regime engine (MA150) SPY technicals → risk-on (TQQQ/UPRO) vs risk-off (GLD/DBC). Single momentum-leader pick within each basket. Terminal shift(1) for execution lag. Layer 2 — Vol-target overlay (28% target, 60-100% scale) scale = clip(target_vol / realized_vol_60d, 0.6, 1.0) Applied with shift(1) so today's scale uses yesterday's vol. Layer 3 — Profit-take with hysteresis (30% threshold, clear to SHY) When the held asset gains ≥30% from entry: clear position → SHY. Restore when gain drops below 20% (hysteresis band = 10%). Entry price = yesterday's close at symbol change (PIT-safe). Why profit-take works on leveraged ETFs --------------------------------------- 3x ETFs (TQQQ, UPRO) suffer from volatility drag: daily rebalancing erodes multi-day compound returns proportional to variance. After a +30% gain, the position sits on a large base where subsequent volatility causes disproportionate drag. Clearing the position: 1. Locks in geometric gains before vol drag erodes them. 2. Reduces the base on which the drag operates. 3. Forces rebalancing from an asset that has become "overweight." This is not alpha from prediction — it is alpha from mechanical rebalancing of a negatively-convex instrument, a structural effect documented in the leveraged ETF literature. Empirical 10y (2016-05-16 to 2026-05-13, 10bps one-way cost): Ann 54.7%, Vol 24.2%, Sharpe(rf=5%) 1.72, MaxDD -25.7%, Sortino 2.23, Calmar 2.13. 16 CLEAR events, 9 RESTORE events, avg ~1.6 clears/year. """ from __future__ import annotations import numpy as np import pandas as pd from strategies.base import Strategy from strategies.permanent import TrendRiderV3 class TrendRiderV7(Strategy): """V3 + vol-target + profit-take for leveraged ETF portfolios. Parameters ---------- ma_long : int MA window for the V3 regime signal on SPY. Default 150 (validated: smooth Sharpe surface across MA100-200). target_vol : float Annualized vol target for the vol-scaling overlay. vol_window : int Trailing window (trading days) for realized-vol estimate. min_lev, max_lev : float Clip bounds for the vol-target scale factor. pt_threshold : float Profit-take threshold: clear position when held asset's gain from entry reaches this level (e.g. 0.30 = +30%). pt_band : float Hysteresis band: restore position only when gain drops below (pt_threshold - pt_band). Prevents oscillation near threshold. pt_park : str Symbol to allocate cleared capital to (default "SHY"). Set to "" to hold cash (0% return) instead. """ def __init__( self, # V3 regime engine ma_long: int = 150, signal: str = "SPY", risk_on: tuple[str, ...] = ("TQQQ", "UPRO"), risk_off: tuple[str, ...] = ("GLD", "DBC"), # Vol-target overlay target_vol: float = 0.28, vol_window: int = 60, min_lev: float = 0.6, max_lev: float = 1.0, # Profit-take overlay pt_threshold: float = 0.30, pt_band: float = 0.10, pt_park: str = "SHY", # V3 passthrough **v3_kwargs, ) -> None: self.target_vol = target_vol self.vol_window = vol_window self.min_lev = min_lev self.max_lev = max_lev self.pt_threshold = pt_threshold self.pt_band = pt_band self.pt_park = pt_park self.v3 = TrendRiderV3( signal=signal, risk_on=risk_on, risk_off=risk_off, ma_long=ma_long, **v3_kwargs, ) def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame: # --- Layer 1: V3 regime weights (already shift(1)'d) --- w = self.v3.generate_signals(data) # Ensure pt_park column exists so PT can allocate to it if self.pt_park and self.pt_park in data.columns and self.pt_park not in w.columns: w[self.pt_park] = 0.0 # --- Layer 2: Vol-target overlay --- daily_ret = data.pct_change(fill_method=None).fillna(0.0) port_rets = (w * daily_ret).sum(axis=1) realized_vol = ( port_rets.rolling(self.vol_window, min_periods=21).std() * np.sqrt(252) ) scale = (self.target_vol / realized_vol).clip( lower=self.min_lev, upper=self.max_lev, ) scale = scale.shift(1).fillna(1.0) w = w.mul(scale, axis=0) # --- Layer 3: Profit-take with hysteresis --- if self.pt_threshold <= 0: return w held = w.idxmax(axis=1) max_w = w.max(axis=1) held[max_w < 1e-8] = "" park_col = self.pt_park if self.pt_park in w.columns else "" entry_price: float | None = None current_sym: str | None = None is_stopped = False restore_level = self.pt_threshold - self.pt_band for i in range(len(w)): sym = held.iloc[i] if not sym or max_w.iloc[i] < 1e-8: current_sym = None entry_price = None is_stopped = False continue # New position (V3 switched symbol) if sym != current_sym: current_sym = sym entry_price = ( float(data[sym].iloc[i - 1]) if i > 0 and sym in data.columns else None ) is_stopped = False continue if ( entry_price is None or entry_price <= 0 or sym not in data.columns ): continue # PIT-safe: use yesterday's close for the gain check yesterday = float(data[sym].iloc[i - 1]) if i > 0 else float(data[sym].iloc[i]) gain = yesterday / entry_price - 1.0 if is_stopped: if gain < restore_level: is_stopped = False else: w.iloc[i] = 0.0 if park_col: w.at[w.index[i], park_col] = scale.iloc[i] else: if gain >= self.pt_threshold: is_stopped = True w.iloc[i] = 0.0 if park_col: w.at[w.index[i], park_col] = scale.iloc[i] return w __all__ = ["TrendRiderV7"]