Gahow Wang
149a00c458
- main.py: add IBKR-style tiered fee schedule (fee_base + fee_per_share), PIT universe support, and open-to-close execution improvements - metrics.py: add raw_summary helper for JSON-safe metric export - Misc strategy fixes: deprecation warnings, NaN handling Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
307 lines
13 KiB
Python
307 lines
13 KiB
Python
"""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"]
|