Add configurable risk penalty and top1 5-day candidate

index-rotation/README.md

@@ -100,6 +100,7 @@ final_score = score_mom - 0.30 * score_risk_penalty
 - `Top2`：若合格标的数 `= 1`，单标的 `100%`
 - `Top1`：只持有第一名
 - 若合格标的数 `= 0`，空仓
+- 可通过策略配置中的 `risk_penalty_multiplier` 调整风险惩罚强度（默认 `0.30`）
 
 ### 4. 调仓频率
 
@@ -159,6 +160,12 @@ Top2 每 5 个交易日：
 python3 -m src.backtest.run --config configs/strategy/top2_every_5_days.yaml
 ```
 
+Top1 每 5 个交易日（较强风险惩罚候选配置）：
+
+```bash
+python3 -m src.backtest.run --config configs/strategy/top1_every_5_days_p05.yaml
+```
+
 默认输出目录：
 
 ```text

index-rotation/configs/strategy/top1_every_5_days_p05.yaml

@@ -0,0 +1,8 @@
+name: top1_every_5_days_p05
+top_n: 1
+rebalance_frequency: every_5_days
+commission_bps: 0.0
+slippage_bps: 0.0
+cash_return: 0.0
+risk_penalty_multiplier: 0.5
+start_date: "2019-12-31"

index-rotation/src/backtest/engine.py

@@ -46,6 +46,7 @@ class BacktestConfig:
     slippage_bps: float = 0.0
     cash_return: float = 0.0
     annualization: int = 252
+    risk_penalty_multiplier: float = 0.30
 
 
 def load_feature_panel(
@@ -91,7 +92,11 @@ def run_backtest(features_panel: pd.DataFrame, config: BacktestConfig) -> dict[s
     )
     closes = features_panel.pivot(index="trade_date", columns="instrument", values="close").reindex(trade_dates)
 
-    signal_panel = build_signal_panel(features_panel, top_n=config.top_n)
+    signal_panel = build_signal_panel(
+        features_panel,
+        top_n=config.top_n,
+        risk_penalty_multiplier=config.risk_penalty_multiplier,
+    )
     allocated = allocate_weights(signal_panel, top_n=config.top_n)
     rebalance_plan = build_rebalance_plan(allocated, frequency=config.rebalance_frequency)
 

index-rotation/src/backtest/run.py

@@ -50,6 +50,7 @@ def build_backtest_config(payload: dict[str, Any]) -> tuple[str, BacktestConfig,
             commission_bps=float(payload.get("commission_bps", 0.0)),
             slippage_bps=float(payload.get("slippage_bps", 0.0)),
             cash_return=float(payload.get("cash_return", 0.0)),
+            risk_penalty_multiplier=float(payload.get("risk_penalty_multiplier", 0.30)),
         ),
         payload.get("start_date"),
         payload.get("end_date"),

index-rotation/src/signals/scorer.py

@@ -16,10 +16,14 @@ RISK_WEIGHTS = {
     "vol_20d": 0.40,
 }
 
-RISK_PENALTY_MULTIPLIER = 0.30
+DEFAULT_RISK_PENALTY_MULTIPLIER = 0.30
 
 
-def add_composite_scores(frame: pd.DataFrame) -> pd.DataFrame:
+def add_composite_scores(
+    frame: pd.DataFrame,
+    *,
+    risk_penalty_multiplier: float = DEFAULT_RISK_PENALTY_MULTIPLIER,
+) -> pd.DataFrame:
     scored = add_cross_sectional_ranks(
         frame,
         columns=[*MOMENTUM_WEIGHTS.keys(), *RISK_WEIGHTS.keys()],
@@ -31,5 +35,5 @@ def add_composite_scores(frame: pd.DataFrame) -> pd.DataFrame:
     scored["score_risk_penalty"] = 0.0
     for column, weight in RISK_WEIGHTS.items():
         scored["score_risk_penalty"] = scored["score_risk_penalty"] + scored[f"{column}_rank"] * weight
-    scored["final_score"] = scored["score_mom"] - RISK_PENALTY_MULTIPLIER * scored["score_risk_penalty"]
+    scored["final_score"] = scored["score_mom"] - risk_penalty_multiplier * scored["score_risk_penalty"]
     return scored

index-rotation/src/signals/selector.py

@@ -6,11 +6,19 @@ from src.signals.scorer import add_composite_scores
 from src.signals.trend import apply_trend_filter
 
 
-def build_signal_panel(features_frame: pd.DataFrame, *, top_n: int) -> pd.DataFrame:
+def build_signal_panel(
+    features_frame: pd.DataFrame,
+    *,
+    top_n: int,
+    risk_penalty_multiplier: float = 0.30,
+) -> pd.DataFrame:
     if top_n < 1:
         raise ValueError("top_n must be >= 1")
 
-    signals = add_composite_scores(apply_trend_filter(features_frame))
+    signals = add_composite_scores(
+        apply_trend_filter(features_frame),
+        risk_penalty_multiplier=risk_penalty_multiplier,
+    )
     signals = signals.sort_values(["trade_date", "instrument"]).reset_index(drop=True)
     signals["eligible_for_selection"] = signals["trend_pass"] & signals["final_score"].notna()
     signals["eligible_count"] = signals.groupby("trade_date")["eligible_for_selection"].transform("sum").astype(int)

index-rotation/tests/test_phase2_signals.py

@@ -88,6 +88,72 @@ class SignalLayerTests(unittest.TestCase):
         self.assertTrue(pd.isna(signals.loc["chinext50", "selection_rank"]))
         self.assertGreater(signals.loc["sse50", "final_score"], signals.loc["hs300", "final_score"])
 
+    def test_custom_risk_penalty_can_change_selection_order(self) -> None:
+        frame = pd.DataFrame(
+            [
+                {
+                    "instrument": "high_beta",
+                    "trade_date": pd.Timestamp("2020-01-10"),
+                    "close": 110,
+                    "daily_return": 0.01,
+                    "ret_5d": 0.12,
+                    "ret_10d": 0.12,
+                    "ret_20d": 0.12,
+                    "ret_60d": 0.12,
+                    "ma_20": 100,
+                    "ma_60": 95,
+                    "vol_10d": 0.30,
+                    "vol_20d": 0.30,
+                },
+                {
+                    "instrument": "defensive",
+                    "trade_date": pd.Timestamp("2020-01-10"),
+                    "close": 108,
+                    "daily_return": 0.01,
+                    "ret_5d": 0.11,
+                    "ret_10d": 0.11,
+                    "ret_20d": 0.11,
+                    "ret_60d": 0.11,
+                    "ma_20": 100,
+                    "ma_60": 95,
+                    "vol_10d": 0.05,
+                    "vol_20d": 0.05,
+                },
+                {
+                    "instrument": "filler_a",
+                    "trade_date": pd.Timestamp("2020-01-10"),
+                    "close": 90,
+                    "daily_return": -0.01,
+                    "ret_5d": 0.02,
+                    "ret_10d": 0.02,
+                    "ret_20d": 0.02,
+                    "ret_60d": 0.02,
+                    "ma_20": 100,
+                    "ma_60": 101,
+                    "vol_10d": 0.10,
+                    "vol_20d": 0.10,
+                },
+                {
+                    "instrument": "filler_b",
+                    "trade_date": pd.Timestamp("2020-01-10"),
+                    "close": 89,
+                    "daily_return": -0.01,
+                    "ret_5d": 0.01,
+                    "ret_10d": 0.01,
+                    "ret_20d": 0.01,
+                    "ret_60d": 0.01,
+                    "ma_20": 100,
+                    "ma_60": 101,
+                    "vol_10d": 0.11,
+                    "vol_20d": 0.11,
+                },
+            ]
+        )
+        low_penalty = build_signal_panel(frame, top_n=1, risk_penalty_multiplier=0.0).set_index("instrument")
+        high_penalty = build_signal_panel(frame, top_n=1, risk_penalty_multiplier=1.0).set_index("instrument")
+        self.assertEqual(low_penalty.loc["high_beta", "selection_rank"], 1)
+        self.assertEqual(high_penalty.loc["defensive", "selection_rank"], 1)
+
     def test_top1_top2_and_empty_allocation(self) -> None:
         base_signals = build_signal_panel(make_signal_input(), top_n=2)