#!/usr/bin/env python3 # -*- coding: utf-8 -*- """hk_backtest.py — 港股策略回测+资金模拟（独立于A股strategy_lab） 功能： 1. 读港股12维面板 + 港股策略库(hk_strategies.py) 2. 生成信号 → trades（止盈/止损/最大持有出场） 3. portfolio_sim 资金模拟（复用 strategy_lab.portfolio_sim 纯函数，不改A股框架） 4. 输出组合指标：平均年化/近1年/近6月/近3月（Ralph Loop验收标准） 用法： python3 hk_backtest.py # 全部策略 + 组合 python3 hk_backtest.py --version hk_pe_mom # 单策略 """ import sys import argparse import collections import sqlite3 from datetime import datetime, timedelta import pandas as pd sys.path.insert(0, "/home/hmo/MoFin/deploy/profile-scripts") sys.path.insert(0, "/home/hmo/MoFin") # strategy_lab.portfolio_sim（纯函数复用） from hk_strategies import HK_STRATEGIES, get_hk_strategy PANEL = "/tmp/panel_12d_hk.pkl" DB = "/home/hmo/MoFin/data/mofin.db" COST = 0.0015 # 港股往返费率近似（佣金+印花税） # 港股温区映射（组合按温区调度用） _REGIME_CACHE = None def load_regime_map(): global _REGIME_CACHE if _REGIME_CACHE is None: conn = sqlite3.connect(DB) _REGIME_CACHE = dict(conn.execute( "SELECT date, regime FROM market_regime WHERE market='hk'").fetchall()) conn.close() return _REGIME_CACHE def load_panel(): p = pd.read_pickle(PANEL) return p.sort_values(["code", "date"]).reset_index(drop=True) def gen_trades(panel, strat): """按策略入场条件生成信号 → trades""" e = strat["entry"] cond = pd.Series(True, index=panel.index) if "pe_q_max" in e: cond &= panel["pe_q"] < e["pe_q_max"] if "sec_ret20_min" in e: cond &= panel["sec_ret20"] > e["sec_ret20_min"] if "rsi_max" in e: cond &= panel["rsi"] < e["rsi_max"] 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 = [] for _, s in sig.iterrows(): df = bycode.get(s["code"]) if df is None: continue idx = df.index[df["date"] == s["date"]] if len(idx) == 0: continue pos = df.index.get_loc(idx[0]) fut = df.iloc[pos + 1: pos + maxh + 2] if len(fut) < 2: continue ep = s["close"] if ep <= 0: continue exit_p, reason, hold = None, None, 0 for k, fb in enumerate(fut.itertuples()): if fb.close <= ep * (1 - sl): exit_p, reason, hold = ep * (1 - sl), "stop", k + 1 break if fb.close >= ep * (1 + tp): exit_p, reason, hold = ep * (1 + tp), "target", k + 1 break if exit_p is None: exit_p, reason, hold = fut.iloc[-1]["close"], "time", maxh trades.append({ "code": s["code"], "name": s["code"], "entry_date": s["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, "score": 0, "score_comp": {}, "kelly": 0, "stop_loss": round(ep * (1 - sl), 2), "target": round(ep * (1 + tp), 2), "dna": False, "factors": {}, }) return trades def portfolio_metrics(trades, capital=1000000, slots=8): """资金模拟 + 时间窗收益（Ralph Loop验收标准）""" import strategy_lab as lab pf = lab.portfolio_sim(trades, capital, max_positions=slots) years_span = 7.5 cagr = pf.get("cagr_pct") # 时间窗收益（按 entry_date 过滤 trades，8槽等权复利净值） def window_return(months): cutoff = (datetime(2026, 7, 24) - timedelta(days=int(months * 30.4))).strftime("%Y-%m-%d") wt = [t for t in trades if t["entry_date"] >= cutoff] if not wt: return None nav = 1.0 for t in sorted(wt, key=lambda x: x["entry_date"]): nav *= (1 + t["profit_pct"] / 100 / slots) return (nav - 1) * 100 return { "cagr": cagr, "year1": window_return(12), "month6": window_return(6), "month3": window_return(3), "trades": len(trades), "win_rate": sum(1 for t in trades if t["profit_pct"] > 0) / len(trades) * 100 if trades else 0, } def main(): ap = argparse.ArgumentParser() ap.add_argument("--version", default=None) ap.add_argument("--slots", type=int, default=8) args = ap.parse_args() panel = load_panel() print(f"港股面板: {len(panel)} 行\n", flush=True) versions = [args.version] if args.version else list(HK_STRATEGIES.keys()) all_trades = [] rm = load_regime_map() # 港股温区映射（组合温区调度） for v in versions: strat = get_hk_strategy(v) if not strat: print(f"未知策略: {v}") continue print(f"=== {v} ({strat['name']}) ===", flush=True) trades = gen_trades(panel, strat) if not trades: print(" 无交易\n", flush=True) continue # 温区调度：只保留策略适用温区的交易（组合正确性关键） reg = strat.get("regime", "all") if reg != "all" and args.version is None: trades = [t for t in trades if rm.get(t["entry_date"]) == reg] print(f" 温区调度({reg}): 保留 {len(trades)} 笔", flush=True) m = portfolio_metrics(trades, slots=args.slots) print(f" 交易{m['trades']} 胜率{m['win_rate']:.0f}% 组合年化{m['cagr']}% " f"近1年{m['year1']:+.1f}% 近6月{m['month6']:+.1f}% 近3月{m['month3']:+.1f}%\n", flush=True) all_trades += trades if all_trades and not args.version: print("=== 港股策略组合（全温区）===", flush=True) m = portfolio_metrics(all_trades, slots=args.slots) print(f" 组合: 交易{m['trades']} 胜率{m['win_rate']:.0f}% 组合年化{m['cagr']}% " f"近1年{m['year1']:+.1f}% 近6月{m['month6']:+.1f}% 近3月{m['month3']:+.1f}%", flush=True) ok = (m['cagr'] or 0) > 10 and (m['year1'] or 0) > 10 and (m['month6'] or 0) > 5 and (m['month3'] or 0) > 0 print(f" 验收: {'✅ 达标' if ok else '❌ 未达标'}（年化>10/近1年>10/近6月>5/近3月>0）", flush=True) if __name__ == "__main__": main()