108 lines
4.6 KiB
Python
108 lines
4.6 KiB
Python
#!/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()
|