diff --git a/deploy/profile-scripts/hk_backtest.py b/deploy/profile-scripts/hk_backtest.py index 556aebdb..40a28f53 100644 --- a/deploy/profile-scripts/hk_backtest.py +++ b/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