feat: 三振出局防守——个股连续3笔亏损暂停15天,自动避开持续下跌被埋

deploy/profile-scripts/hk_backtest.py

@@ -47,8 +47,8 @@ def load_panel():
     return p.sort_values(["code", "date"]).reset_index(drop=True)
 
 
-def gen_trades(panel, strat):
-    """按策略入场条件生成信号 → trades"""
+def gen_trades_defensive(panel, strat, strike=3, cooldown_days=15):
+    """带三振出局防守的信号→trades：连续 strike 笔亏损 → 暂停 cooldown_days（自动避持续下跌被埋）"""
     e = strat["entry"]
     cond = pd.Series(True, index=panel.index)
     if "pe_q_max" in e:
@@ -62,20 +62,26 @@ def gen_trades(panel, strat):
     if "bias60_max" in e:
         cond &= panel["bias60"] < e["bias60_max"]
     if "rsi_delta_min" in e:
-        # 面板无rsi_delta，用rsi与前5日差值近似（面板已有rsi）
         cond &= panel["rsi"] - panel.groupby("code")["rsi"].shift(5) >= e["rsi_delta_min"]
     if "vol_ratio_min" in e:
         cond &= panel["vol_ratio"] > e["vol_ratio_min"]
     sig = panel[cond].copy()
     sig = sig.dropna(subset=["close"])
-    print(f"  {strat['version']} 信号: {len(sig)}", flush=True)
 
     ex = strat["exit"]
     tp, sl, maxh = ex["tp_pct"], ex["sl_pct"], ex["max_hold_days"]
     bycode = {c: df for c, df in panel.groupby("code")}
     trades = []
+    last_loss_dates = {}   # code -> last loss date（个股级三振）
+    consecutive = {}       # code -> 连续亏损数
     for _, s in sig.iterrows():
-        df = bycode.get(s["code"])
+        code = s["code"]
+        # 三振出局：个股连续亏损 strike 次 → 暂停 cooldown_days
+        if consecutive.get(code, 0) >= strike:
+            if (pd.Timestamp(s["date"]) - pd.Timestamp(last_loss_dates.get(code, "1900-01-01"))).days < cooldown_days:
+                continue
+            consecutive[code] = 0  # 冷却期后恢复
+        df = bycode.get(code)
         if df is None:
             continue
         idx = df.index[df["date"] == s["date"]]
@@ -98,14 +104,18 @@ def gen_trades(panel, strat):
                 break
         if exit_p is None:
             exit_p, reason, hold = fut.iloc[-1]["close"], "time", maxh
-        # 退出日期（用该股票日历往后推 hold 个交易日）
+        pnl = (exit_p - ep) / ep * 100
         exit_date = fut.iloc[min(hold - 1, len(fut) - 1)]["date"] if hold > 0 else s["date"]
+        # 更新三振状态
+        if pnl < 0:
+            consecutive[code] = consecutive.get(code, 0) + 1
+            last_loss_dates[code] = exit_date
+        else:
+            consecutive[code] = 0
         trades.append({
-            "code": s["code"], "name": s["code"], "entry_date": s["date"],
-            "exit_date": exit_date,
+            "code": code, "name": code, "entry_date": s["date"], "exit_date": exit_date,
             "entry_price": round(ep, 2), "exit_price": round(exit_p, 2),
-            "profit_pct": round((exit_p - ep) / ep * 100, 2),
-            "exit_reason": reason, "hold_days": hold,
+            "profit_pct": round(pnl, 2), "exit_reason": reason, "hold_days": hold,
             "score": 0, "score_comp": {}, "kelly": 0,
             "stop_loss": round(ep * (1 - sl), 2), "target": round(ep * (1 + tp), 2),
             "dna": False, "factors": {},
@@ -113,6 +123,11 @@ def gen_trades(panel, strat):
     return trades
 
 
+def gen_trades(panel, strat):
+    """兼容入口：无防守的信号生成"""
+    return gen_trades_defensive(panel, strat, strike=999, cooldown_days=0)
+
+
 def portfolio_nav(trades, capital=1000000, slots=8):
     """8槽资金管理净值曲线：每日结算到期→入场(仓位满跳过)→持仓按成本估值。
     返回 (nav_series: dict date->nav, stats)"""
@@ -223,7 +238,7 @@ def main():
             print(f"未知策略: {v}")
             continue
         print(f"=== {v} ({strat['name']}) ===", flush=True)
-        trades = gen_trades(panel, strat)
+        trades = gen_trades_defensive(panel, strat, strike=3, cooldown_days=15)
         if not trades:
             print("  无交易\n", flush=True)
             continue