""" Hybrid Alpha Strategy - Round 2 iteration. Takes the best elements from the top performers: 1. FactorCombo's rec_mfilt + deep_upvol signal (strongest alpha) 2. Inverse-vol weighting (better risk-adjusted from ImprovedMomQuality) 3. Light regime awareness (partial scale-down, not binary) 4. Monthly rebalancing Also tests: - Recovery + quality blend without MA200 filter - Wider top_n for diversification """ import numpy as np import pandas as pd from strategies.base import Strategy def _rank(df): return df.rank(axis=1, pct=True, na_option="keep") class HybridAlphaStrategy(Strategy): """ Combines FactorCombo's best signal with risk-parity weighting. """ def __init__( self, rebal_freq: int = 21, top_n: int = 20, vol_window: int = 60, use_invvol: bool = True, regime_dampen: float = 0.5, # scale factor in bear regime (1.0 = no regime) ): self.rebal_freq = rebal_freq self.top_n = top_n self.vol_window = vol_window self.use_invvol = use_invvol self.regime_dampen = regime_dampen def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame: p = data # --- Signal: rec_mfilt + deep x upvol (from FactorCombo) --- # Recovery momentum-filtered rec = p / p.rolling(126, min_periods=126).min() - 1 mom = p.shift(21).pct_change(105) rec_mfilt = rec.where(mom > 0, np.nan) rec_mfilt_r = _rank(rec_mfilt) # Deep recovery x up-volume rec_126 = p / p.rolling(126, min_periods=126).min() - 1 ret = p.pct_change() up_vol = ret.where(ret > 0, 0).rolling(20, min_periods=15).sum() deep_upvol = _rank(rec_126) * _rank(up_vol) deep_upvol_r = _rank(deep_upvol) # Combined signal signal = 0.5 * rec_mfilt_r + 0.5 * deep_upvol_r # --- Select top_n --- rank = signal.rank(axis=1, ascending=False, na_option="bottom") n_valid = signal.notna().sum(axis=1) enough = n_valid >= self.top_n top_mask = (rank <= self.top_n) & enough.values.reshape(-1, 1) if self.use_invvol: # Inverse-vol weighting vol = ret.rolling(self.vol_window, min_periods=30).std().replace(0, np.nan) inv_vol = (1.0 / vol).where(top_mask, 0.0) row_sums = inv_vol.sum(axis=1).replace(0, np.nan) signals = inv_vol.div(row_sums, axis=0).fillna(0.0) else: # Equal weight raw = top_mask.astype(float) row_sums = raw.sum(axis=1).replace(0, np.nan) signals = raw.div(row_sums, axis=0).fillna(0.0) # --- Light regime dampening --- if self.regime_dampen < 1.0: # Use market-wide vol regime instead of MA200 (more responsive) market_vol = ret.mean(axis=1).rolling(20).std() * np.sqrt(252) vol_90th = market_vol.rolling(252, min_periods=126).quantile(0.90) high_vol = market_vol > vol_90th regime_scale = pd.Series(1.0, index=data.index) regime_scale[high_vol] = self.regime_dampen signals = signals.mul(regime_scale, axis=0) # --- Monthly rebalance --- warmup = 252 rebal_mask = pd.Series(False, index=data.index) rebal_indices = list(range(warmup, len(data), self.rebal_freq)) rebal_mask.iloc[rebal_indices] = True signals[~rebal_mask] = np.nan signals = signals.ffill().fillna(0.0) signals.iloc[:warmup] = 0.0 return signals.shift(1).fillna(0.0) class RecoveryQualityBlendStrategy(Strategy): """ Blends recovery, momentum, and quality without strict MA200 filter. Uses intermediate momentum as a soft signal (not hard gate). """ def __init__( self, recovery_window: int = 63, mom_lookback: int = 252, mom_skip: int = 21, quality_window: int = 252, vol_window: int = 60, rebal_freq: int = 21, top_n: int = 20, ): self.recovery_window = recovery_window self.mom_lookback = mom_lookback self.mom_skip = mom_skip self.quality_window = quality_window self.vol_window = vol_window self.rebal_freq = rebal_freq self.top_n = top_n def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame: p = data # Recovery recovery = p / p.rolling(self.recovery_window, min_periods=self.recovery_window).min() - 1 # 12-1 month momentum momentum = p.shift(self.mom_skip).pct_change(self.mom_lookback - self.mom_skip) # Intermediate momentum (7m) int_mom = p.shift(self.mom_skip).pct_change(147 - self.mom_skip) # Quality: consistency monthly_ret = p.pct_change(21) consistency = (monthly_ret > 0).astype(float).rolling( self.quality_window, min_periods=self.quality_window // 2 ).mean() # Up-volume proxy ret = p.pct_change() up_vol = ret.where(ret > 0, 0).rolling(20, min_periods=15).sum() # Cross-sectional ranks rec_rank = _rank(recovery) mom_rank = _rank(momentum) int_mom_rank = _rank(int_mom) con_rank = _rank(consistency) upv_rank = _rank(up_vol) # Composite: weighted blend of all factors # Recovery 25%, momentum 25%, intermediate momentum 20%, quality 15%, up_vol 15% composite = (0.25 * rec_rank + 0.25 * mom_rank + 0.20 * int_mom_rank + 0.15 * con_rank + 0.15 * upv_rank) # Select top_n rank = composite.rank(axis=1, ascending=False, na_option="bottom") n_valid = composite.notna().sum(axis=1) enough = n_valid >= self.top_n top_mask = (rank <= self.top_n) & enough.values.reshape(-1, 1) # Inverse-vol weighting vol = ret.rolling(self.vol_window, min_periods=30).std().replace(0, np.nan) inv_vol = (1.0 / vol).where(top_mask, 0.0) row_sums = inv_vol.sum(axis=1).replace(0, np.nan) signals = inv_vol.div(row_sums, axis=0).fillna(0.0) # Monthly rebalance warmup = max(self.mom_lookback, self.quality_window, self.recovery_window) + self.mom_skip rebal_mask = pd.Series(False, index=data.index) rebal_indices = list(range(warmup, len(data), self.rebal_freq)) rebal_mask.iloc[rebal_indices] = True signals[~rebal_mask] = np.nan signals = signals.ffill().fillna(0.0) signals.iloc[:warmup] = 0.0 return signals.shift(1).fillna(0.0)