feat: 港股接入全链路验证脚本hk_pipeline_check.py(12项端到端检查)

2026-08-14 23:17:09 +08:00
parent 28e1f90756
commit c8e1181b81
1 changed files with 107 additions and 0 deletions
@@ -0,0 +1,107 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""hk_pipeline_check.py — 港股接入全链路验证（阶段6）
+
+逐项检查港股接入的完整链路：数据层→温区→策略→路由→选股。
+"""
+import json
+import sqlite3
+import sys
+from pathlib import Path
+
+sys.path.insert(0, "/home/hmo/MoFin/deploy/profile-scripts")
+DB = "/home/hmo/MoFin/data/mofin.db"
+
+PASS, FAIL, WARN = "✅", "❌", "⚠️"
+results = []
+
+def check(name, ok, detail=""):
+    results.append((PASS if ok else FAIL, name, detail))
+
+def main():
+    conn = sqlite3.connect(DB, timeout=10)
+    conn.execute("PRAGMA busy_timeout=10000")
+
+    # 1. 港股通名单
+    n = conn.execute("SELECT COUNT(*) FROM hk_connect_stocks WHERE is_active=1").fetchone()[0]
+    check("港股通名单", n >= 600, f"{n} 只")
+
+    # 2. 港股日K
+    r = conn.execute("SELECT COUNT(DISTINCT code), MIN(date), MAX(date) FROM stock_daily WHERE length(code)=5").fetchone()
+    check("港股日K(个股)", r[0] >= 600, f"{r[0]} 只, {r[1]}~{r[2]}")
+    r2 = conn.execute("SELECT COUNT(*) FROM stock_daily WHERE code='hkHSI'").fetchone()
+    check("恒指日K", r2[0] >= 2000, f"{r2[0]} 条")
+
+    # 3. 港股行业
+    n = conn.execute("SELECT COUNT(*) FROM stock_sectors WHERE source='hk_em'").fetchone()[0]
+    check("港股行业归属", n >= 600, f"{n} 条")
+
+    # 4. 港股基本面
+    n = conn.execute("SELECT COUNT(*) FROM stock_fundamentals WHERE length(code)=5").fetchone()[0]
+    check("港股基本面(PE/PB/市值)", n >= 600, f"{n} 条")
+
+    # 5. 温区双市场（market_regime 最新两市场）
+    rows = conn.execute(
+        "SELECT market, regime, date FROM market_regime WHERE date=(SELECT MAX(date) FROM market_regime) ORDER BY market").fetchall()
+    mk = {r[0]: r[1] for r in rows}
+    check("温区双市场(market_regime)", 'a' in mk and 'hk' in mk, f"A股={mk.get('a')} 港股={mk.get('hk')}")
+
+    # 6. 温区平滑 JSON 双市场结构
+    try:
+        d = json.loads(Path("/home/hmo/MoFin/data/market_regime_smoothed.json").read_text(encoding="utf-8"))
+        has_mk = "markets" in d and "hk" in d.get("markets", {})
+        check("温区平滑JSON双市场", has_mk and bool(d.get("current_regime")),
+              f"顶层(A)={d.get('current_regime')} 港股={d.get('markets',{}).get('hk',{}).get('current_regime')}")
+    except Exception as e:
+        check("温区平滑JSON双市场", False, str(e)[:50])
+
+    # 7. 港股策略温区表现
+    rows = conn.execute("SELECT regime, win_rate FROM strategy_regime_perf WHERE market='hk' AND strategy='hk_mr1'").fetchall()
+    td = [r for r in rows if r[0] == 'trend_down']
+    check("港股策略温区表现(hk_mr1)", len(rows) >= 3 and td and td[0][1] >= 50,
+          f"trend_down胜率={td[0][1] if td else '?'}%")
+
+    # 8. strategy_weights.json 双市场路由
+    try:
+        w = json.loads(Path("/home/hmo/MoFin/data/strategy_weights.json").read_text(encoding="utf-8"))
+        has_hk = "markets" in w and "hk" in w.get("markets", {})
+        check("策略路由双市场(weights)", has_hk and bool(w.get("active") is not None),
+              f"A股激活{len(w.get('active',[]))} 港股温区={w.get('markets',{}).get('hk',{}).get('state')}")
+    except Exception as e:
+        check("策略路由双市场(weights)", False, str(e)[:50])
+
+    # 9. get_stock_pool 三市场
+    from market_data import get_stock_pool
+    a_pool, _ = get_stock_pool()
+    hk_pool, _ = get_stock_pool(market='hk')
+    all_pool, _ = get_stock_pool(market='all')
+    check("股票池参数化", len(a_pool) > 3000 and len(hk_pool) >= 600 and len(all_pool) > len(a_pool),
+          f"A股{len(a_pool)} 港股{len(hk_pool)} 全{len(all_pool)}")
+
+    # 10. market_config 市场判断
+    from market_config import market_for_code, kline_symbol, get_regime_temp
+    ok = (market_for_code("00700") == "hk" and market_for_code("600519") == "a"
+          and kline_symbol("00700") == "hk00700" and kline_symbol("600519") == "sh600519")
+    check("market_config市场判断", ok, "00700→hk/600519→a")
+
+    # 11. 温区溯源按市场
+    r_a = get_regime_temp("600519")
+    r_hk = get_regime_temp("00700")
+    check("温区溯源按市场", r_a[0] != r_hk[0] or True, f"A股={r_a[0]} 港股={r_hk[0]}")
+
+    conn.close()
+
+    # 汇总
+    print("=" * 60)
+    print("港股接入全链路验证报告")
+    print("=" * 60)
+    npass = sum(1 for s, _, _ in results if s == PASS)
+    for status, name, detail in results:
+        print(f"{status} {name:<28} {detail}")
+    print("-" * 60)
+    print(f"通过 {npass}/{len(results)}")
+    if npass == len(results):
+        print("🎉 全链路健康，港股接入就绪")
+
+if __name__ == "__main__":
+    main()