""" Moving-average slope momentum engine. Hypothesis: steeper positive (negative) MA slope → stronger trend → better long (short) follow-through. Slope is measured over a lookback window on the MA line itself (not raw price), which smooths noise vs price ROC. Signal at bar *t* → position from *t+1* (no lookahead). Stage 4 extensions: dual-timeframe confirmation, regime gates (SMA200 / VIX), vol-scaled sizing, ATR chandelier exits. """ from __future__ import annotations from dataclasses import dataclass from typing import Literal import numpy as np import pandas as pd MaType = Literal["sma", "ema"] SlopeMethod = Literal["pct", "regression", "annualized_pct"] Direction = Literal["long_only", "long_short"] ExitMode = Literal["slope_flip", "slope_half", "price_cross", "slope_or_price", "atr_chandelier"] SignalMode = Literal["single", "dual_timeframe"] RegimeGate = Literal["none", "sma200", "vix_lt"] SizingMode = Literal["binary", "vol_scale"] RankMetric = Literal["dual_product", "dual_blend", "fast", "slow", "dual_min"] def compute_ma(close: pd.Series, *, ma_type: MaType, period: int) -> pd.Series: c = close.astype(np.float64) if period <= 0: raise ValueError("period must be > 0") if ma_type == "sma": return c.rolling(window=period, min_periods=period).mean() if ma_type == "ema": return c.ewm(span=period, adjust=False, min_periods=period).mean() raise ValueError(f"unknown ma_type: {ma_type!r}") def compute_ma_slope( ma: pd.Series, *, lookback: int, method: SlopeMethod = "pct", ) -> pd.Series: """ Slope of the MA line over ``lookback`` bars. Methods ------- pct (MA_t - MA_{t-lookback}) / MA_{t-lookback} annualized_pct pct * (252 / lookback) — comparable across lookbacks regression OLS slope of MA vs bar index, divided by MA level (dimensionless) """ if lookback <= 0: raise ValueError("lookback must be > 0") m = ma.astype(np.float64) if method == "pct": past = m.shift(lookback) return (m - past) / past.replace(0.0, np.nan) if method == "annualized_pct": past = m.shift(lookback) raw = (m - past) / past.replace(0.0, np.nan) return raw * (252.0 / float(lookback)) if method == "regression": arr = m.to_numpy(dtype=np.float64) n = len(arr) out = np.full(n, np.nan, dtype=np.float64) x = np.arange(lookback, dtype=np.float64) x_mean = x.mean() x_var = ((x - x_mean) ** 2).sum() if x_var <= 0: return pd.Series(out, index=ma.index) for i in range(lookback - 1, n): window = arr[i - lookback + 1 : i + 1] if np.any(~np.isfinite(window)): continue y_mean = window.mean() if abs(y_mean) < 1e-12: continue cov = ((x - x_mean) * (window - y_mean)).sum() slope = cov / x_var out[i] = slope / y_mean return pd.Series(out, index=ma.index) raise ValueError(f"unknown slope method: {method!r}") def compute_vol_scale( ret: pd.Series, *, target_vol: float, window: int, floor: float, cap: float, ) -> pd.Series: realized = ret.rolling(window, min_periods=window).std(ddof=1) * np.sqrt(252.0) scale = target_vol / realized.replace(0.0, np.nan) return scale.clip(floor, cap).fillna(1.0) def compute_slope_rank_score( close: pd.Series, *, ma_type: MaType = "ema", fast_period: int = 10, slow_period: int = 50, fast_lookback: int = 10, slow_lookback: int = 5, slope_method: SlopeMethod = "pct", entry_slope_min: float = 0.0, price_above_ma: bool = True, rank_metric: RankMetric = "dual_product", ) -> pd.DataFrame: """ Cross-sectional ranking score from fast/slow MA slopes (stage-4 SPY logic). Returns ``fast_slope``, ``slow_slope``, ``rank_score``. Ineligible names → NaN score. """ c = close.astype(np.float64) fast_ma = compute_ma(c, ma_type=ma_type, period=fast_period) slow_ma = compute_ma(c, ma_type=ma_type, period=slow_period) fast_slope = compute_ma_slope(fast_ma, lookback=fast_lookback, method=slope_method) slow_slope = compute_ma_slope(slow_ma, lookback=slow_lookback, method=slope_method) eligible = (fast_slope > entry_slope_min) & (slow_slope > entry_slope_min) if price_above_ma: eligible = eligible & (c > fast_ma) if rank_metric == "dual_product": score = fast_slope * slow_slope elif rank_metric == "dual_blend": score = (fast_slope + slow_slope) * 0.5 elif rank_metric == "fast": score = fast_slope elif rank_metric == "slow": score = slow_slope elif rank_metric == "dual_min": score = np.minimum(fast_slope, slow_slope) else: raise ValueError(f"unknown rank_metric: {rank_metric!r}") score = score.where(eligible) return pd.DataFrame( { "fast_ma": fast_ma, "slow_ma": slow_ma, "fast_slope": fast_slope, "slow_slope": slow_slope, "rank_score": score, }, index=c.index, ) @dataclass(frozen=True) class MaSlopeConfig: ma_type: MaType = "ema" ma_period: int = 50 slope_lookback: int = 10 slope_method: SlopeMethod = "pct" entry_slope_min: float = 0.01 exit_slope_max: float | None = None price_above_ma: bool = True direction: Direction = "long_only" exit_mode: ExitMode = "slope_flip" min_hold_days: int = 0 # Stage 4 signal_mode: SignalMode = "single" slow_ma_period: int = 50 slow_ma_type: MaType | None = None slow_slope_lookback: int | None = None regime_gate: RegimeGate = "none" regime_sma_window: int = 200 regime_exit_on_break: bool = True vix_cap: float = 25.0 sizing_mode: SizingMode = "binary" target_vol: float = 0.15 vol_scale_floor: float = 0.5 vol_scale_cap: float = 1.5 vol_window: int = 20 atr_period: int = 14 atr_high_lookback: int = 22 atr_multiplier: float = 2.5 def slug(self) -> str: exit_tag = self.exit_slope_max if self.exit_slope_max is not None else "auto" pf = "pma" if self.price_above_ma else "nopf" base = ( f"{self.ma_type}{self.ma_period}_lb{self.slope_lookback}_{self.slope_method}" f"_en{self.entry_slope_min:g}_ex{exit_tag}_{self.exit_mode}_{pf}_{self.direction}" ) if self.signal_mode == "single" and self.regime_gate == "none" and self.sizing_mode == "binary": return base slow_type = self.slow_ma_type or self.ma_type slow_lb = self.slow_slope_lookback if self.slow_slope_lookback is not None else self.slope_lookback parts = [base] if self.signal_mode == "dual_timeframe": parts.append(f"dual_{slow_type}{self.slow_ma_period}_lb{slow_lb}") if self.regime_gate != "none": rg = self.regime_gate if rg == "vix_lt": rg = f"vix{self.vix_cap:g}" elif rg == "sma200": rg = f"sma{self.regime_sma_window}" parts.append(rg) if self.exit_mode == "atr_chandelier": parts.append(f"atr{self.atr_multiplier:g}") if self.sizing_mode == "vol_scale": parts.append(f"vol{self.target_vol:g}") return "_".join(parts) @dataclass class MaSlopeEngine: """Long-only / long-short MA-slope trend follower with optional dual-TF and sizing.""" config: MaSlopeConfig = MaSlopeConfig() def transform(self, daily: pd.DataFrame) -> pd.DataFrame: if daily.empty: raise ValueError("daily is empty") if "close" not in daily.columns: raise KeyError("daily must contain 'close'") cfg = self.config df = daily.sort_index().copy() close = df["close"].astype(np.float64) ret = ( df["ret"].astype(np.float64).fillna(0.0) if "ret" in df.columns else close.pct_change().fillna(0.0) ) ma = compute_ma(close, ma_type=cfg.ma_type, period=cfg.ma_period) slope = compute_ma_slope(ma, lookback=cfg.slope_lookback, method=cfg.slope_method) slow_ma = pd.Series(np.nan, index=df.index) slow_slope = pd.Series(np.nan, index=df.index) if cfg.signal_mode == "dual_timeframe": slow_type = cfg.slow_ma_type or cfg.ma_type slow_lb = cfg.slow_slope_lookback if cfg.slow_slope_lookback is not None else cfg.slope_lookback slow_ma = compute_ma(close, ma_type=slow_type, period=cfg.slow_ma_period) slow_slope = compute_ma_slope(slow_ma, lookback=slow_lb, method=cfg.slope_method) exit_thresh = cfg.exit_slope_max if exit_thresh is None: exit_thresh = 0.0 if cfg.exit_mode in ("slope_flip", "slope_or_price", "atr_chandelier") else cfg.entry_slope_min * 0.5 long_entry = slope > cfg.entry_slope_min short_entry = slope < -cfg.entry_slope_min if cfg.price_above_ma: long_entry = long_entry & (close > ma) short_entry = short_entry & (close < ma) if cfg.signal_mode == "dual_timeframe": long_entry = long_entry & (slow_slope > cfg.entry_slope_min) short_entry = short_entry & (slow_slope < -cfg.entry_slope_min) regime_ok = pd.Series(True, index=df.index) regime_exit = pd.Series(False, index=df.index) if cfg.regime_gate == "sma200": sma_reg = close.rolling(cfg.regime_sma_window, min_periods=cfg.regime_sma_window).mean() regime_ok = close > sma_reg if cfg.regime_exit_on_break: regime_exit = close < sma_reg df["regime_sma"] = sma_reg elif cfg.regime_gate == "vix_lt": if "vix" not in df.columns: raise KeyError("regime_gate=vix_lt requires a 'vix' column on daily frame") vix = df["vix"].astype(np.float64) regime_ok = vix < cfg.vix_cap if cfg.regime_exit_on_break: regime_exit = vix >= cfg.vix_cap long_entry = long_entry & regime_ok short_entry = short_entry & regime_ok long_exit_slope = slope < exit_thresh short_exit_slope = slope > -exit_thresh long_exit_price = close < ma short_exit_price = close > ma if cfg.exit_mode == "slope_flip": long_exit = long_exit_slope short_exit = short_exit_slope elif cfg.exit_mode == "slope_half": long_exit = slope < cfg.entry_slope_min * 0.5 short_exit = slope > -cfg.entry_slope_min * 0.5 elif cfg.exit_mode == "price_cross": long_exit = long_exit_price short_exit = short_exit_price elif cfg.exit_mode == "slope_or_price": long_exit = long_exit_slope | long_exit_price short_exit = short_exit_slope | short_exit_price elif cfg.exit_mode == "atr_chandelier": long_exit = long_exit_slope short_exit = short_exit_slope else: raise ValueError(f"unknown exit_mode: {cfg.exit_mode!r}") if cfg.regime_exit_on_break and cfg.regime_gate != "none": long_exit = long_exit | regime_exit short_exit = short_exit | regime_exit if cfg.exit_mode == "atr_chandelier": low_col = df["low"].astype(np.float64) if "low" in df.columns else close target = self._atr_chandelier_state_machine( long_entry=long_entry, long_exit=long_exit, short_entry=short_entry, short_exit=short_exit, close=close, high=df["high"].astype(np.float64) if "high" in df.columns else close, low=low_col, direction=cfg.direction, min_hold_days=cfg.min_hold_days, atr_period=cfg.atr_period, atr_high_lookback=cfg.atr_high_lookback, atr_multiplier=cfg.atr_multiplier, ) else: target = self._position_state_machine( long_entry=long_entry, long_exit=long_exit, short_entry=short_entry, short_exit=short_exit, direction=cfg.direction, min_hold_days=cfg.min_hold_days, ) target_f = target.astype(np.float64) if cfg.sizing_mode == "vol_scale": scale = compute_vol_scale( ret, target_vol=cfg.target_vol, window=cfg.vol_window, floor=cfg.vol_scale_floor, cap=cfg.vol_scale_cap, ) weight = target_f * scale else: weight = target_f df["ma"] = ma df["ma_slope"] = slope if cfg.signal_mode == "dual_timeframe": df["ma_slow"] = slow_ma df["ma_slow_slope"] = slow_slope df["long_entry"] = long_entry df["long_exit"] = long_exit df["short_entry"] = short_entry df["short_exit"] = short_exit df["target_position"] = target_f df["position_weight"] = weight df["position"] = weight.shift(1).fillna(0.0) return df @staticmethod def _position_state_machine( *, long_entry: pd.Series, long_exit: pd.Series, short_entry: pd.Series, short_exit: pd.Series, direction: Direction, min_hold_days: int, ) -> pd.Series: idx = long_entry.index pos = np.zeros(len(idx), dtype=np.int8) state = 0 hold = 0 for i in range(len(idx)): if state == 0: if direction == "long_only" and bool(long_entry.iloc[i]): state = 1 hold = 0 elif direction == "long_short": if bool(long_entry.iloc[i]): state = 1 hold = 0 elif bool(short_entry.iloc[i]): state = -1 hold = 0 elif state == 1: hold += 1 if hold >= min_hold_days and bool(long_exit.iloc[i]): state = 0 hold = 0 elif state == -1: hold += 1 if hold >= min_hold_days and bool(short_exit.iloc[i]): state = 0 hold = 0 pos[i] = state return pd.Series(pos, index=idx, dtype=np.int8) @staticmethod def _atr_chandelier_state_machine( *, long_entry: pd.Series, long_exit: pd.Series, short_entry: pd.Series, short_exit: pd.Series, close: pd.Series, high: pd.Series, low: pd.Series, direction: Direction, min_hold_days: int, atr_period: int, atr_high_lookback: int, atr_multiplier: float, ) -> pd.Series: prev_close = close.shift(1) tr = pd.concat( [ high - low, (high - prev_close).abs(), (low - prev_close).abs(), ], axis=1, ).max(axis=1) atr = tr.rolling(atr_period, min_periods=atr_period).mean() hh = high.rolling(atr_high_lookback, min_periods=1).max() idx = close.index pos = np.zeros(len(idx), dtype=np.int8) state = 0 hold = 0 stop = 0.0 for i in range(len(idx)): if state == 0: if direction == "long_only" and bool(long_entry.iloc[i]): state = 1 hold = 0 stop = float(hh.iloc[i] - atr_multiplier * atr.iloc[i]) elif direction == "long_short": if bool(long_entry.iloc[i]): state = 1 hold = 0 stop = float(hh.iloc[i] - atr_multiplier * atr.iloc[i]) elif bool(short_entry.iloc[i]): state = -1 hold = 0 elif state == 1: hold += 1 if np.isfinite(atr.iloc[i]): stop = max(stop, float(hh.iloc[i] - atr_multiplier * atr.iloc[i])) chandelier_hit = float(close.iloc[i]) < stop if hold >= min_hold_days and (bool(long_exit.iloc[i]) or chandelier_hit): state = 0 hold = 0 elif state == -1: hold += 1 if hold >= min_hold_days and bool(short_exit.iloc[i]): state = 0 hold = 0 pos[i] = state return pd.Series(pos, index=idx, dtype=np.int8) def backtest_returns(self, daily: pd.DataFrame, *, ret_col: str = "ret") -> pd.Series: """Daily strategy returns from ``ret_col`` (close-to-close) and lagged position.""" if ret_col not in daily.columns: close = daily["close"].astype(np.float64) ret = close.pct_change().fillna(0.0) else: ret = daily[ret_col].astype(np.float64).fillna(0.0) frame = self.transform(daily) pos = frame["position"].astype(np.float64) return (pos * ret).rename("strategy_ret") def metrics_from_returns( r: pd.Series, *, capital: float = 100_000.0, position: pd.Series | None = None, ) -> dict[str, float]: r = r.astype(np.float64).dropna() if len(r) < 2: return {} eq = capital * (1.0 + r).cumprod() years = len(r) / 252.0 end = float(eq.iloc[-1]) tot = end / capital - 1.0 cagr = (end / capital) ** (1.0 / years) - 1.0 if years > 0 else float("nan") dd = float((eq / eq.cummax() - 1.0).min()) sd = float(r.std(ddof=1)) sharpe = float(r.mean() / sd * np.sqrt(252.0)) if sd > 1e-12 else float("nan") invested = float(position.reindex(r.index).abs().mean()) if position is not None else float("nan") return { "n_days": float(len(r)), "total_return_pct": tot * 100.0, "cagr_pct": cagr * 100.0, "max_dd_pct": dd * 100.0, "sharpe": sharpe, "vol_ann_pct": sd * np.sqrt(252.0) * 100.0, "end_equity": end, "invested_frac": invested, }