"""TrendRider V9 research lab. Scratch harness for the goal "beat every registered US strategy on the past 10y CAGR *and* Sharpe under strict PIT". Runs candidates through the same official path as precompute_backtest.py (run_market -> cash ledger, $10k, 0.1% cost + $2/order, next-session close fills via session_targets). Only ETF strategies are used here (fast, no stock panel needed). """ from __future__ import annotations import numpy as np import pandas as pd import trader from strategies.base import Strategy from trader import ETF_STRATEGY_UNIVERSES, STRATEGY_REGISTRY V9_UNIVERSE = [ "SPY", "QQQ", "TQQQ", "UPRO", "SOXL", "TECL", "TNA", "FAS", "SPXL", "SSO", "QLD", "GLD", "DBC", "TMF", "TLT", "SHY", "IEF", "SQQQ", "LABU", ] def _above_ma(closes: np.ndarray, w: int) -> bool: return closes.size >= w and float(closes[-1]) > float(closes[-w:].mean()) 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)) 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) class TrendRiderV9(Strategy): """V7 architecture (V3 regime + vol target + profit take) with: - extended risk-on basket (TECL/SOXL/... momentum leader pick) - gated risk-off additions: TMF (TLT>MA200 gate), SQQQ (QQQ 20d drop gate) - optional asset-level vol targeting or EWMA vol - optional momentum/vol pick ranking - laddered profit-taking; park bucket configurable (SHY/QQQ mix) PIT convention identical to TrendRiderV3/V7: decisions at raw row t use bars through t-1; the V3-style loop then shifts the output by one extra day before the V7 layers, and the engine applies the final -1 shift. """ def __init__( self, *, # V3 regime core signal: str = "SPY", risk_on: tuple[str, ...] = ("TQQQ", "UPRO"), risk_off: tuple[str, ...] = ("GLD", "DBC"), ma_long: int = 150, 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, # extensions tmf: str | None = None, tlt_ma_window: int = 200, sqqq: str | None = None, sqqq_trigger: float = 0.08, ro_stop_loss: float | None = None, mom_pick_divisor: bool = False, trail_pct: float | None = None, trail_window: int = 30, trail_reentry_days: int = 0, instma_exit: int | None = None, fast_reentry_days: int | None = None, breadth_ma: int | None = None, # vol target target_vol: float = 0.28, vt_window: int = 60, min_lev: float = 0.6, max_lev: float = 1.0, vt_mode: str = "strategy", vt_ewma: float = 0.0, # profit take pt_mode: str = "binary", pt_threshold: float = 0.30, pt_band: float = 0.10, pt_park: str = "SHY", pt_park_mix: dict | None = None, pt_ladder: tuple = ((0.30, 0.5), (0.60, 1.0)), pt_min_hold: int = 0, pt_only: tuple[str, ...] | None = None, # macro throttle (free-data overlays: VIX term structure, credit) throttle_factor: float = 0.6, throttle_vix_inv: bool = False, vix_inv_thresh: float = 0.0, throttle_vix_spike: bool = False, vix_spike_5d: float = 0.5, throttle_credit: bool = False, credit_ma: int = 50, throttle_recover_days: int = 5, ) -> None: self.signal = signal self.risk_on = tuple(risk_on) self.risk_off = tuple(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 self.tmf = tmf self.tlt_ma_window = tlt_ma_window self.sqqq = sqqq self.sqqq_trigger = sqqq_trigger self.ro_stop_loss = ro_stop_loss self.mom_pick_divisor = mom_pick_divisor self.trail_pct = trail_pct self.trail_window = trail_window self.trail_reentry_days = trail_reentry_days self.instma_exit = instma_exit self.fast_reentry_days = fast_reentry_days self.breadth_ma = breadth_ma self.target_vol = target_vol self.vt_window = vt_window self.min_lev = min_lev self.max_lev = max_lev self.vt_mode = vt_mode self.vt_ewma = vt_ewma self.pt_mode = pt_mode self.pt_threshold = pt_threshold self.pt_band = pt_band self.pt_park = pt_park self.pt_park_mix = pt_park_mix self.pt_ladder = tuple(pt_ladder) self.pt_min_hold = pt_min_hold self.pt_only = set(pt_only) if pt_only else None self.throttle_factor = throttle_factor self.throttle_vix_inv = throttle_vix_inv self.vix_inv_thresh = vix_inv_thresh self.throttle_vix_spike = throttle_vix_spike self.vix_spike_5d = vix_spike_5d self.throttle_credit = throttle_credit self.credit_ma = credit_ma self.throttle_recover_days = throttle_recover_days # ---- regime helpers (same rules as TrendRiderV3) ---- 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 _above_ma(closes, self.ma_long): return "risk_off" v = _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 (_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 (_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 _macro_scale(self, data: pd.DataFrame) -> pd.Series | None: """PIT macro throttle: VIX term inversion / VIX spike / credit stress each multiply exposure by `throttle_factor` (enter fast, exit after `throttle_recover_days` consecutive clean sessions).""" conds: list[tuple[pd.Series, float]] = [] if self.throttle_vix_inv and "^VIX9D" in data.columns and "^VIX3M" in data.columns: ratio = data["^VIX9D"] / data["^VIX3M"].replace(0.0, np.nan) conds.append(((ratio - 1.0) >= self.vix_inv_thresh, self.throttle_factor)) if self.throttle_vix_spike and "^VIX" in data.columns: vix = data["^VIX"] conds.append((vix / vix.shift(5).replace(0.0, np.nan) - 1.0 >= self.vix_spike_5d, self.throttle_factor)) if self.throttle_credit and "HYG" in data.columns: hyg = data["HYG"] conds.append((hyg < hyg.rolling(self.credit_ma).mean(), self.throttle_factor)) if not conds: return None out = pd.Series(1.0, index=data.index) for cond, factor in conds: cond = cond.fillna(False).to_numpy() state, clean, vals = False, 0, np.ones(len(cond)) for i, c in enumerate(cond): if c: state, clean = True, 0 elif state: clean += 1 if clean >= self.throttle_recover_days: state = False if state: vals[i] = factor out = out * pd.Series(vals, index=data.index) return out def _pick_top(self, basket: list[int], cps: dict[int, np.ndarray]) -> int | None: best_i, best_r = None, -np.inf for ix in basket: closes = cps[ix] r = _total_return(closes, self.mom_lookback) if r != r: continue if self.mom_pick_divisor: sigma = _vol(closes, self.mom_lookback) r = r / sigma if sigma == sigma and sigma > 1e-9 else -np.inf if r > best_r: best_r, best_i = r, ix return best_i def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame: gate_refs = [] if self.tmf: gate_refs.append("TLT") if self.sqqq: gate_refs.append("QQQ") if self.breadth_ma: gate_refs.append("RSP") cols = [c for c in dict.fromkeys( [self.signal, *self.risk_on, *self.risk_off, *([self.tmf] if self.tmf else []), *([self.sqqq] if self.sqqq else []), *([self.pt_park] if self.pt_park else []), *gate_refs]) if c in data.columns] sym_to_ix = {s: ix for ix, s in enumerate(cols)} w = pd.DataFrame(0.0, index=data.index, columns=cols) if self.signal not in sym_to_ix: return w.shift(1).fillna(0.0) sig_arr = data[self.signal].to_numpy() 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] tmf_ix = sym_to_ix.get(self.tmf) if self.tmf else None tlt_ix = sym_to_ix.get("TLT") sqqq_ix = sym_to_ix.get(self.sqqq) if self.sqqq else None qqq_ix = sym_to_ix.get("QQQ") if tmf_ix is not None and tmf_ix not in roff_idx: roff_idx.append(tmf_ix) if sqqq_ix is not None and sqqq_ix not in roff_idx: roff_idx.append(sqqq_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 fast_until = -1 rsp_ix = sym_to_ix.get("RSP") if self.breadth_ma else None def gate_on(ix: int, cps: dict[int, np.ndarray]) -> bool: if ix == tmf_ix: closes = cps.get(tlt_ix) return closes is not None and _above_ma(closes, self.tlt_ma_window) if ix == sqqq_ix: closes = cps.get(qqq_ix) if closes is None or closes.size < 22: return False if closes[-1] <= 0 or closes[-22] <= 0: return False return closes[-1] / closes[-22] - 1.0 <= -self.sqqq_trigger return True for t in range(len(data)): if t < need: continue 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 if rsp_ix is not None and t > self.breadth_ma + 1: ratio = sym_arrays[rsp_ix][:t] / sig_arr[:t] if ratio[-1] < float(ratio[-self.breadth_ma:].mean()): desired = "risk_off" cps = {ix: arr[:t] for ix, arr in sym_arrays.items()} ron_today = [ix for ix in ron_idx if gate_on(ix, cps)] roff_today = [ix for ix in roff_idx if gate_on(ix, cps)] def assign_one(sym_ix: int) -> None: nonlocal current_sym, bars_in_sym, sym_entry_close current_sym = sym_ix bars_in_sym = 0 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]) if current_regime is None: basket = ron_today if desired == "risk_on" else roff_today pick = self._pick_top(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) 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(roff_today, 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 if (not in_on and self.ro_stop_loss and sym_yclose is not None and sym_entry_close and current_sym in (tmf_ix, sqqq_ix) and sym_yclose / sym_entry_close - 1.0 <= -self.ro_stop_loss): # bail to cash-park until the regime/pick changes again park_ix = sym_to_ix.get(self.pt_park) if park_ix is not None: current_sym = None sym_entry_close = None w.iat[t, park_ix] = 1.0 continue if (in_on and self.instma_exit and sym_yclose is not None and current_sym is not None): closes = cps[current_sym] if closes.size >= self.instma_exit and sym_yclose < float( closes[-self.instma_exit:].mean()): pick = self._pick_top(roff_today, cps) if pick is not None: current_regime = "risk_off" bars_in_regime = 0 assign_one(pick) pending_regime = None pending_count = 0 if self.fast_reentry_days: fast_until = t + self.fast_reentry_days w.iat[t, pick] = 1.0 continue if (in_on and self.trail_pct and sym_yclose is not None and current_sym is not None): w0 = max(0, t - self.trail_window) hc = float(np.nanmax(sym_arrays[current_sym][w0:t])) if hc > 0 and sym_yclose / hc - 1.0 <= -self.trail_pct: pick = self._pick_top(roff_today, cps) if pick is not None: current_regime = "risk_off" bars_in_regime = 0 assign_one(pick) pending_regime = None pending_count = 0 if self.trail_reentry_days: cooloff_remaining = self.trail_reentry_days if self.fast_reentry_days: fast_until = t + self.fast_reentry_days w.iat[t, pick] = 1.0 continue if emergency and current_regime != "risk_off": pick = self._pick_top(roff_today, 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 if desired != current_regime: fast = bool(self.fast_reentry_days and desired == "risk_on" and t < fast_until) min_hold_eff = 5 if fast else self.regime_min_hold confirm_eff = 1 if fast else self.confirm_days if current_regime == "risk_off" and cooloff_remaining > 0: pending_regime = None pending_count = 0 elif bars_in_regime < min_hold_eff: pending_regime = None pending_count = 0 else: if desired != pending_regime: pending_regime = desired pending_count = 1 else: pending_count += 1 if pending_count >= confirm_eff: basket = ron_today if desired == "risk_on" else roff_today pick = self._pick_top(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 if pick is not None: w.iat[t, pick] = 1.0 continue if current_sym is not None: w.iat[t, current_sym] = 1.0 continue pending_regime = None pending_count = 0 basket = ron_today if in_on else roff_today top = self._pick_top(basket, cps) carry_ix = current_sym if carry_ix is None: park_ix = sym_to_ix.get(self.pt_park) if park_ix is not None: w.iat[t, park_ix] = 1.0 continue if top is None or top == carry_ix: w.iat[t, carry_ix] = 1.0 continue if bars_in_sym < self.instrument_min_hold: w.iat[t, carry_ix] = 1.0 continue assign_one(top) w.iat[t, top] = 1.0 w = w.shift(1).fillna(0.0) # ---- Layer 2: vol target ---- if self.vt_mode == "asset": assv = {} for ix, arr in sym_arrays.items(): s = pd.Series(arr, index=data.index) assv[ix] = s.pct_change(fill_method=None).rolling( self.vt_window, min_periods=21).std() * np.sqrt(252) held_ix = w.to_numpy().argmax(axis=1) held_vol = np.full(len(w), np.nan) for i in range(len(w)): if w.iloc[i].max() > 1e-8: held_vol[i] = assv[held_ix[i]].iloc[i] realized = pd.Series(held_vol, index=data.index) else: daily_ret = data.pct_change(fill_method=None).fillna(0.0) port_rets = (w * daily_ret).sum(axis=1) if self.vt_mode == "downside": downside = port_rets.where(port_rets < 0.0, 0.0) realized = downside.rolling(self.vt_window, min_periods=21).std() * np.sqrt(252) realized = realized * np.sqrt(2.0) # semi-deviation -> full-vol scale elif self.vt_ewma and self.vt_ewma > 0: realized = port_rets.ewm(halflife=self.vt_ewma, min_periods=21).std() * np.sqrt(252) else: realized = port_rets.rolling(self.vt_window, min_periods=21).std() * np.sqrt(252) scale = (self.target_vol / realized).clip(lower=self.min_lev, upper=self.max_lev) macro = self._macro_scale(data) if macro is not None: scale = scale * macro.reindex(scale.index).shift(1).fillna(1.0) scale = scale.shift(1).fillna(1.0) w = w.mul(scale, axis=0) # ---- Layer 3: profit take ---- if self.pt_mode == "none": return w park_cols = [self.pt_park] if self.pt_park else [] for c in [*park_cols, *(self.pt_park_mix or {})]: if c and c in data.columns and c not in w.columns: w[c] = 0.0 if self.pt_park_mix: mix = {c: f for c, f in self.pt_park_mix.items() if c in w.columns} if mix: total = sum(mix.values()) mix = {c: f / total for c, f in mix.items()} park_cols = list(mix) else: mix, park_cols = None, [] else: mix = None levels = sorted(self.pt_ladder, key=lambda x: x[0]) thresholds = [g for g, _ in levels] cumfracs = [min(1.0, max(0.0, f)) for _, f in levels] held = w.idxmax(axis=1) held[w.max(axis=1) < 1e-8] = "" entry_price: float | None = None current_sym: str | None = None is_cleared = False cleared_idx = -1 last_restore_i = -10 ** 9 for i in range(len(w)): sym = held.iloc[i] if not sym: current_sym = None entry_price = None is_cleared = False cleared_idx = -1 continue if self.pt_only is not None and sym not in self.pt_only: current_sym = sym entry_price = None is_cleared = False cleared_idx = -1 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_cleared = False cleared_idx = -1 last_restore_i = -10 ** 9 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 self.pt_mode == "binary": if is_cleared: if gain < self.pt_threshold - self.pt_band: is_cleared = False last_restore_i = i elif gain >= self.pt_threshold and i - last_restore_i >= self.pt_min_hold: is_cleared = True frac = 1.0 if is_cleared else 0.0 else: # ladder if cleared_idx >= 0 and gain < thresholds[cleared_idx] - self.pt_band: cleared_idx = -1 for k, g in enumerate(thresholds): if gain >= g and k > cleared_idx: cleared_idx = k frac = cumfracs[cleared_idx] if cleared_idx >= 0 else 0.0 if frac > 0: cur_scale = float(w.iloc[i].sum()) w.iloc[i] = 0.0 for c in park_cols: w.at[w.index[i], c] = cur_scale * frac * (mix[c] if mix else 1.0) if frac < 1.0 and sym in w.columns: w.at[w.index[i], sym] = cur_scale * (1.0 - frac) return w # --------------------------------------------------------------------------- # harness # --------------------------------------------------------------------------- _COUNT = {"n": 0} def register(name: str, factory, universe: list[str] | None = None) -> str: STRATEGY_REGISTRY[name] = factory ETF_STRATEGY_UNIVERSES[name] = sorted(set(universe or V9_UNIVERSE)) return name def metrics(equity: pd.Series) -> dict: vals = equity.to_numpy(dtype=float) rets = vals[1:] / vals[:-1] - 1.0 n_years = len(rets) / 252.0 cagr = (vals[-1] / vals[0]) ** (1 / n_years) - 1 vol = rets.std() * np.sqrt(252) sharpe = rets.mean() / rets.std() * np.sqrt(252) if rets.std() > 0 else 0.0 mdd = float((1 - vals / np.maximum.accumulate(vals)).max()) return {"cagr": cagr, "vol": vol, "sharpe": sharpe, "mdd": mdd, "total": vals[-1] / vals[0] - 1.0, "years": n_years, "end": vals[-1]} def run(names: list[str], *, start: str = "2017-12-26", end: str = "2026-09-30", capital: float = 10000.0, tx_cost: float = 0.001, fixed_fee: float = 2.0, resize_band: float | None = None) -> dict[str, pd.Series]: from backtest_runner import run_market out = run_market("us", names, None, capital, tx_cost, fixed_fee, refresh=False, start=start, end=end) res = {} for run_ in out["strategy_runs"]: res[run_["name"]] = run_["equity"] for e in out["errors"]: print("ERROR:", e) return res def report(curves: dict[str, pd.Series], title: str = "") -> pd.DataFrame: rows = [] for name, eq in curves.items(): m = metrics(eq) rows.append({"name": name, **m}) df = pd.DataFrame(rows).sort_values("cagr", ascending=False) pd.set_option("display.width", 200) print(f"== {title} == ({len(df)} strategies)") for _, r in df.iterrows(): print(f"{r['name']:44s} CAGR {r['cagr']*100:6.1f}% vol {r['vol']*100:5.1f}% " f"Sharpe {r['sharpe']:5.2f} MDD {r['mdd']*100:6.1f}% end {r['end']:,.0f}") return df def yearly_table(curves: dict[str, pd.Series], names: list[str]) -> pd.DataFrame: cols = {} for n in names: if n not in curves: continue eq = curves[n] last = eq.groupby(eq.index.year).last() prev = last.shift(1) first_year_ret = last.iloc[0] / eq.iloc[0] - 1 yr = (last / prev - 1).fillna(0.0) yr.iloc[0] = first_year_ret cols[n] = yr.drop(index=eq.index.year[0] - 1, errors="ignore") df = pd.DataFrame(cols) print("== yearly returns ==") print((df * 100).round(1)) return df