"""TrendRider V8 — V7 + TMF risk-off upgrade. Extends V7 (V3 regime + vol-target + profit-take) with a smarter risk-off basket: when TLT is in a bull regime (above its MA200), TMF (3x long-duration Treasuries) joins the risk-off momentum selection alongside GLD and DBC. Why this works -------------- V7 spends ~30-40% of time in risk-off, holding GLD or DBC which average low single-digit returns. During equity bear markets the Fed typically cuts rates, driving long bonds higher — TMF (3x TLT) captures this convexity. 2020 crash: TMF ~+60% while equities fell 34%. The TLT MA200 gate prevents TMF allocation during bond bear markets (e.g. 2022 rate-hiking cycle where TLT fell 31%). PIT safety ---------- V3's generate_signals uses prices through day t-1 internally, then applies a terminal shift(1). So V3's output weight at row i uses data through day i-2. The TMF swap and TLT gate must match this information set: all lookups use data through day i-2 (shift(2) for vectorized signals, iloc[i-2] for point lookups). Profit-take is applied ONLY to risk-on assets (TQQQ/UPRO). Risk-off assets (GLD, DBC, TMF) are exempt because: 1. TMF can gain 30%+ during rate-cut cycles — PT would sell at the worst possible time. 2. Risk-off is already regime-gated; PT on defensive assets is redundant. """ from __future__ import annotations import numpy as np import pandas as pd from strategies.base import Strategy from strategies.permanent import TrendRiderV3 class TrendRiderV8(Strategy): """V7 architecture + TMF risk-off with bond-regime gate. Pipeline: Layer 1 — V3 regime engine → risk-on / risk-off weights Layer 1b — TMF swap (PIT-aligned to V3's info set: data through i-2) Layer 2 — Vol-target overlay Layer 3 — Profit-take (risk-on assets only; risk-off exempt) """ 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"), # TMF risk-off tmf_symbol: str = "TMF", tlt_symbol: str = "TLT", tlt_ma_window: int = 200, tmf_mom_lookback: int = 63, # Vol-target overlay target_vol: float = 0.36, vol_window: int = 60, min_lev: float = 0.75, 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.risk_on = risk_on self.risk_off = risk_off self.tmf_symbol = tmf_symbol self.tlt_symbol = tlt_symbol self.tlt_ma_window = tlt_ma_window self.tmf_mom_lookback = tmf_mom_lookback 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[i] uses data through day i-2. w = self.v3.generate_signals(data) # --- Layer 1b: TMF risk-off swap --- # PIT: must use data through day i-2 to match V3's info set. # shift(2) on vectorized signals; iloc[i-2] for point lookups. tmf = self.tmf_symbol tlt = self.tlt_symbol if tmf in data.columns and tlt in data.columns: tlt_ma = data[tlt].rolling(self.tlt_ma_window).mean() tlt_bull = (data[tlt] > tlt_ma).shift(2).fillna(False) roff_cols = [c for c in self.risk_off if c in w.columns] if tmf not in w.columns: w[tmf] = 0.0 lb = self.tmf_mom_lookback for i in range(lb + 3, len(w)): roff_weight = sum(float(w.iat[i, w.columns.get_loc(c)]) for c in roff_cols) if roff_weight < 1e-8: continue if not tlt_bull.iloc[i]: continue best_sym, best_r = None, -np.inf for sym in roff_cols + [tmf]: if sym not in data.columns: continue p_now = data[sym].iloc[i - 2] p_past = data[sym].iloc[i - 2 - lb] if pd.notna(p_now) and pd.notna(p_past) and p_past > 0: r = float(p_now / p_past - 1.0) if r > best_r: best_r, best_sym = r, sym if best_sym is not None: for c in roff_cols: w.iat[i, w.columns.get_loc(c)] = 0.0 w.iat[i, w.columns.get_loc(tmf)] = 0.0 w.iat[i, w.columns.get_loc(best_sym)] = roff_weight # --- Layer 2: Vol-target overlay --- daily_ret = data.pct_change(fill_method=None).fillna(0.0) common = w.columns.intersection(daily_ret.columns) port_rets = (w[common] * daily_ret[common]).sum(axis=1) 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 (risk-on only) --- # Risk-off assets (GLD, DBC, TMF) are exempt from PT. if self.pt_threshold <= 0: return w risk_off_set = set(self.risk_off) | {self.tmf_symbol} 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 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 # Skip PT for risk-off assets if sym in risk_off_set: continue if ( entry_price is None or entry_price <= 0 or sym not in data.columns ): continue 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__ = ["TrendRiderV8"]