Professionalize factor model: trend/MR as filters, fix vol targeting

Based on quantitative review of the system:

Factor model changes (quant/signals/factors.py):
- Trend: converted from scored factor to post-composite filter
  (price < 200d SMA → score *= 0.5) to avoid momentum correlation
- Blowoff filter: new post-composite filter penalizes stocks with
  Bollinger zscore > 3 (score *= 0.5) to avoid chasing parabolic moves
- Mean reversion: disabled as scored factor (weight=0), replaced by
  blowoff filter for extreme protection

Vol targeting fix (quant/strategy.py):
- Removed renormalization after vol scaling — this was completely
  negating the vol target by scaling weights back to sum=1.0
- Now preserves cash buffer when portfolio vol exceeds target

Config tuning (config.yaml):
- momentum_windows: [63, 126, 252] — dropped noisy 1-month window
- factor_weights: momentum 0.45, quality 0.25, vol 0.10, value 0.05
- max_positions: 18, stop_loss: 12%, sector cap: 40%
- backtest start: 2016-01-01 (was 2022, too biased toward AI bull)
- lookback_years: 5 (was 3)

Tests: 44 pass (4 new for trend_filter + blowoff_filter)

https://claude.ai/code/session_01LbgD64GPmP5tS7j4uavkCa

Author: claude

config.yaml

@@ -532,34 +532,32 @@ universe:
   benchmark: SPY
 
 data:
-  lookback_years: 3
+  lookback_years: 5
   frequency: 1d  # daily bars
 
 signals:
-  # Momentum
-  momentum_windows: [21, 63, 126, 252]  # 1m, 3m, 6m, 12m in trading days
-  # Mean reversion
+  # Momentum — 3m/6m/12m; dropped 1m (too noisy)
+  momentum_windows: [63, 126, 252]  # 3m, 6m, 12m in trading days
+  # Mean reversion — used as blowoff filter (zscore > 3 → penalize)
   mean_reversion_window: 20
-  mean_reversion_zscore_threshold: 2.0
-  # Value
-  # (fundamentals fetched from yfinance)
-  # Volatility
-  volatility_window: 63
-  # Trend
+  mean_reversion_zscore_threshold: 3.0
+  # Trend — used as filter: price < 200d SMA → score *= 0.5
   sma_short: 50
   sma_long: 200
-  # Factor weights (must sum to 1.0) — adjust to change portfolio style
-  # Higher momentum/trend = more growth/tech; higher value/volatility = more stable/defensive
+  # Volatility
+  volatility_window: 63
+  # Factor weights (must sum to 1.0) — trend/mean_reversion now applied as filters, not scored
+  # 因子权重（必须加总为 1.0）— trend 和 mean_reversion 改为过滤器，不参与评分
   factor_weights:
-    momentum: 0.30
-    mean_reversion: 0.10
-    trend: 0.25
-    volatility: 0.05
-    value: 0.15
-    quality: 0.15
+    momentum: 0.45
+    quality: 0.25
+    volatility: 0.10
+    value: 0.05
+    mean_reversion: 0.00   # disabled as scored factor; used as blowoff filter instead
+    trend: 0.00            # disabled as scored factor; used as 200d SMA filter instead
 
 portfolio:
-  max_positions: 20
+  max_positions: 18
   max_position_weight: 0.10  # 10% max single position
   min_position_weight: 0.02  # 2% min
   target_volatility: 0.15    # 15% annualized
@@ -568,11 +566,11 @@ portfolio:
 
 risk:
   max_drawdown_limit: 0.20   # 20% max drawdown halt
-  max_sector_weight: 0.30    # 30% max in one sector
-  stop_loss_pct: 0.08        # 8% individual stop loss
+  max_sector_weight: 0.40    # 40% max in one sector (raised for tech-heavy pool)
+  stop_loss_pct: 0.12        # 12% individual stop loss (wider for momentum strategy)
 
 backtest:
-  start_date: "2022-01-01"
+  start_date: "2016-01-01"
   end_date: null  # null = today
   initial_capital: 1000000
   slippage_bps: 5

quant/signals/factors.py

@@ -71,6 +71,32 @@ def trend_factor(prices: pd.DataFrame, short_window: int = 50,
     return zs
 
 
+def trend_filter(prices: pd.DataFrame, long_window: int = 200,
+                 penalty: float = 0.5) -> pd.DataFrame:
+    """Trend filter: penalize stocks trading below their 200-day SMA.
+
+    Returns a DataFrame of multipliers (1.0 for uptrend, `penalty` for downtrend).
+    Used as a post-composite filter rather than a scored factor.
+    """
+    sma_long = prices.rolling(long_window).mean()
+    above = prices >= sma_long
+    return above.astype(float).replace(0.0, penalty)
+
+
+def blowoff_filter(prices: pd.DataFrame, window: int = 20,
+                    zscore_limit: float = 3.0, penalty: float = 0.5) -> pd.DataFrame:
+    """Penalize stocks with extreme short-term gains (blowoff top protection).
+
+    If a stock's Bollinger z-score exceeds `zscore_limit`, its composite score
+    is multiplied by `penalty`.  Prevents chasing parabolic moves.
+    """
+    rolling_mean = prices.rolling(window).mean()
+    rolling_std = prices.rolling(window).std()
+    zscore = (prices - rolling_mean) / rolling_std
+    overextended = zscore > zscore_limit
+    return (~overextended).astype(float).replace(0.0, penalty)
+
+
 def volatility_factor(returns: pd.DataFrame, window: int = 63) -> pd.DataFrame:
     """Realized volatility factor (low-vol anomaly: prefer lower vol).
 
@@ -152,6 +178,13 @@ def generate(self, prices: pd.DataFrame, returns: pd.DataFrame,
                  fundamentals: pd.DataFrame = None) -> pd.DataFrame:
         """Produce a composite alpha score for each symbol on each date.
 
+        Pipeline:
+          1. Compute scored factors (momentum, volatility, value, quality, etc.)
+          2. Build weighted composite from scored factors
+          3. Apply post-composite filters:
+             - Trend filter: penalize stocks below 200d SMA
+             - Blowoff filter: penalize stocks with extreme short-term gains
+
         Returns DataFrame (dates x symbols) of composite z-scores.
         """
         symbols = [c for c in prices.columns if c != self.benchmark]
@@ -182,17 +215,26 @@ def generate(self, prices: pd.DataFrame, returns: pd.DataFrame,
                 df = df.reindex(columns=px.columns)
                 factors[name] = df
 
-        # Weighted composite
+        # Weighted composite (only factors with weight > 0 contribute)
         composite = pd.DataFrame(0.0, index=px.index, columns=px.columns)
         total_weight = 0.0
         for name, weight in self.weights.items():
-            if name in factors:
+            if weight > 0 and name in factors:
                 f = factors[name].reindex(index=px.index, columns=px.columns)
                 composite += weight * f.fillna(0)
                 total_weight += weight
 
         if total_weight > 0:
             composite /= total_weight
 
+        # --- Post-composite filters ---
+        # Trend filter: penalize stocks below 200d SMA (score *= 0.5)
+        tf = trend_filter(px, long_window=self.sma_long)
+        composite = composite * tf.reindex(index=composite.index, columns=composite.columns).fillna(1.0)
+
+        # Blowoff filter: penalize stocks with extreme overbought z-score > threshold
+        bf = blowoff_filter(px, window=self.mr_window, zscore_limit=self.mr_threshold)
+        composite = composite * bf.reindex(index=composite.index, columns=composite.columns).fillna(1.0)
+
         logger.info("Generated composite signal: shape=%s", composite.shape)
         return composite

quant/strategy.py

@@ -86,10 +86,8 @@ def run_backtest(self, start: str = None, end: str = None) -> BacktestResult:
                 selected.tolist(), day_scores, cov
             )
             weights = self.optimizer.apply_vol_scaling(weights, cov)
-
-            # Normalize back to sum=1 after vol scaling
-            if weights.sum() > 0:
-                weights /= weights.sum()
+            # Do NOT renormalize after vol scaling — the remainder is held as cash.
+            # This preserves the vol-targeting effect.
 
             target_weights[str(date.date())] = weights
 
@@ -151,8 +149,7 @@ def get_current_portfolio(self, capital: float = None) -> pd.DataFrame:
         # Optimize weights
         weights = self.optimizer.optimize_weights(symbols, day_scores, cov)
         weights = self.optimizer.apply_vol_scaling(weights, cov)
-        if weights.sum() > 0:
-            weights /= weights.sum()
+        # Do NOT renormalize — remainder is cash buffer for vol targeting.
 
         # Build output table
         latest_prices = prices[symbols].iloc[-1]

tests/conftest.py

@@ -16,9 +16,9 @@ def config():
         },
         "data": {"lookback_years": 2, "frequency": "1d"},
         "signals": {
-            "momentum_windows": [21, 63, 126],
+            "momentum_windows": [63, 126, 252],
             "mean_reversion_window": 20,
-            "mean_reversion_zscore_threshold": 2.0,
+            "mean_reversion_zscore_threshold": 3.0,
             "volatility_window": 63,
             "sma_short": 50,
             "sma_long": 200,
@@ -34,7 +34,7 @@ def config():
         "risk": {
             "max_drawdown_limit": 0.20,
             "max_sector_weight": 0.30,
-            "stop_loss_pct": 0.08,
+            "stop_loss_pct": 0.12,
         },
         "backtest": {
             "start_date": "2020-01-01",

tests/test_factors.py

@@ -5,8 +5,9 @@
 import pytest
 
 from quant.signals.factors import (
-    momentum_factor, mean_reversion_factor, trend_factor,
-    volatility_factor, value_factor, quality_factor, SignalGenerator,
+    momentum_factor, mean_reversion_factor, trend_factor, trend_filter,
+    blowoff_filter, volatility_factor, value_factor, quality_factor,
+    SignalGenerator,
 )
 
 
@@ -59,6 +60,40 @@ def test_uptrend_positive(self):
         assert last["UP"] > last["DOWN"]
 
 
+class TestTrendFilter:
+    def test_uptrend_no_penalty(self):
+        """Stocks above 200d SMA should get multiplier 1.0."""
+        dates = pd.bdate_range("2020-01-01", periods=300)
+        px = pd.DataFrame({"UP": np.linspace(50, 150, 300)}, index=dates)
+        result = trend_filter(px, long_window=200)
+        assert result["UP"].iloc[-1] == 1.0
+
+    def test_downtrend_penalized(self):
+        """Stocks below 200d SMA should get multiplier 0.5."""
+        dates = pd.bdate_range("2020-01-01", periods=300)
+        px = pd.DataFrame({"DOWN": np.linspace(150, 50, 300)}, index=dates)
+        result = trend_filter(px, long_window=200)
+        assert result["DOWN"].iloc[-1] == 0.5
+
+
+class TestBlowoffFilter:
+    def test_normal_stock_no_penalty(self):
+        """Stock with moderate z-score should get multiplier 1.0."""
+        dates = pd.bdate_range("2020-01-01", periods=50)
+        px = pd.DataFrame({"A": np.linspace(100, 110, 50)}, index=dates)
+        result = blowoff_filter(px, window=20, zscore_limit=3.0)
+        assert result["A"].iloc[-1] == 1.0
+
+    def test_parabolic_stock_penalized(self):
+        """Stock with extreme spike should get multiplier 0.5."""
+        dates = pd.bdate_range("2020-01-01", periods=50)
+        # Flat then massive vertical spike — guarantees zscore >> 3
+        prices = np.concatenate([np.full(45, 100), np.array([100, 100, 100, 100, 500])])
+        px = pd.DataFrame({"A": prices}, index=dates)
+        result = blowoff_filter(px, window=20, zscore_limit=3.0)
+        assert result["A"].iloc[-1] == 0.5
+
+
 class TestVolatilityFactor:
     def test_low_vol_preferred(self, synthetic_returns):
         ret = synthetic_returns.drop(columns=["BENCH"])

tests/test_portfolio.py

@@ -56,7 +56,7 @@ def test_stop_loss(self, config):
         opt = PortfolioOptimizer(config)
         weights = pd.Series({"A": 0.5, "B": 0.5})
         entry = pd.Series({"A": 100.0, "B": 100.0})
-        current = pd.Series({"A": 88.0, "B": 100.0})  # A dropped 12% > 8% stop
+        current = pd.Series({"A": 85.0, "B": 100.0})  # A dropped 15% > 12% stop
         result = opt.check_stop_losses(weights, entry, current)
         assert result["A"] == 0.0
         assert result["B"] > 0.0