hmo/MoFin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 港股策略独立模块——hk_strategies(三温区策略定义)+hk_backtest(独立回测资金模拟,复用portfolio_sim纯函数,零干扰A股)

Browse Source
This commit is contained in:
hmo
2026-08-15 03:03:42 +08:00
parent 216795708c
commit 57bee18ac4
2 changed files with 231 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/profile-scripts/hk_backtest.py
+157
View File
@@ -0,0 +1,157 @@
#!/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
from datetime import datetime, timedelta
import pandas as pd
sys.path.insert(0, "/home/hmo/MoFin/deploy/profile-scripts")
from hk_strategies import HK_STRATEGIES, get_hk_strategy
PANEL = "/tmp/panel_12d_hk.pkl"
COST = 0.0015 # 港股往返费率近似（佣金+印花税）
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 "ret60_max" in e:
cond &= panel["ret60"] < e["ret60_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 做简单等权组合）
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
tot = sum(t["profit_pct"] for t in wt) / slots
return tot
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 = []
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
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()
deploy/profile-scripts/hk_strategies.py
+74
View File
@@ -0,0 +1,74 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""hk_strategies.py — 港股策略定义（独立于A股strategy_lab，零干扰）
港股策略库：各温区互补策略定义（单一来源）。
面板数据：/tmp/panel_12d_hk.pkl（build_panel_hk_v2 构建，含价量/大盘/行业动量/估值分位/资金流）
策略结构：
version: 策略名
regime: 适用温区（trend_up/trend_down/choppy/all）
entry: 入场条件（面板因子，数据定阈值——step29 由果及因扫描）
exit: 出场（止盈/止损/最大持有，港股波动放宽）
"""
# ── 港股策略库（按温区互补）──────────────────────────────
HK_STRATEGIES = {
# trend_up：低PE + 行业动量高（由果及因：大涨率10.28%，2.35x基线）
"hk_pe_mom": {
"version": "hk_pe_mom",
"name": "港股低PE+行业动量（trend_up主战场）",
"regime": "trend_up",
"summary": "低PE(<0.2分位)+行业20日动量>10%。由果及因：大涨率10.28%(基线4.37%)。低PE+1.24pp(step27港股扫描)、行业动量高+1.47pp",
"entry": {
"pe_q_max": 0.2, # PE分位<0.2（低估值）
"sec_ret20_min": 10, # 行业20日动量>10%
},
"exit": {"tp_pct": 0.30, "sl_pct": 0.12, "max_hold_days": 40},
},
# trend_down：深度超卖反弹（由果及因：RSI<25/bias60<-15，trend_down温区55%胜率）
"hk_mr1": {
"version": "hk_mr1",
"name": "港股深度超卖反弹（trend_down主战场）",
"regime": "trend_down",
"summary": "RSI<25+bias60<-15+ret60<-25+止跌回升+放量。港股归因：深度超卖RSI<25胜率55.6%，trend_down温区55%胜率+3.59%",
"entry": {
"rsi_max": 25,
"bias60_max": -15,
"ret60_max": -25,
"rsi_delta_min": 2,
"vol_ratio_min": 1.5,
},
"exit": {"tp_pct": 0.25, "sl_pct": 0.15, "max_hold_days": 30},
},
# choppy：低PE+超跌反弹（震荡市价值修复）
"hk_pe_oversold": {
"version": "hk_pe_oversold",
"name": "港股低PE+超卖反弹（choppy主战场）",
"regime": "choppy",
"summary": "低PE(<0.2分位)+RSI超卖(<45)。由果及因：低PE+超卖组合大涨率6.93%(基线4.37%)",
"entry": {
"pe_q_max": 0.2,
"rsi_max": 45,
"bias60_max": 0,
},
"exit": {"tp_pct": 0.25, "sl_pct": 0.12, "max_hold_days": 40},
},
}
def get_hk_strategy(version):
"""按版本取港股策略定义"""
return HK_STRATEGIES.get(version)
def list_hk_strategies():
"""港股策略列表（按温区）"""
return [{"version": v, "regime": s["regime"], "name": s["name"]}
for v, s in HK_STRATEGIES.items()]
def strategies_for_regime(regime):
"""返回适用某温区的策略版本列表（regime 参数用于组合调度）"""
return [v for v, s in HK_STRATEGIES.items()
if s["regime"] == regime or s["regime"] == "all"]