feat: 港股温区历史回填脚本——对hkHSI全历史逐日算温区入market_regime(market=hk),复用compute_regime算法

2026-08-14 17:36:03 +08:00
parent a7449332d9
commit 0e06a7e0ea
1 changed files with 124 additions and 0 deletions
@@ -0,0 +1,124 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""backfill_hk_regime.py — 港股温区历史回填（一次性脚本）
+
+背景（阶段3 港股接入）：market_regime 表刚加 market 维度，港股温区只有今天
+1 条，regime_tracker 的 K=5 滞回平滑需要足够历史。本脚本对 stock_daily 的
+hkHSI 全历史（2014 起，约 3000 条）逐日计算温区，写入 market_regime(market='hk')。
+
+算法：完全复用 market_regime.compute_regime 的判定逻辑（ma20/adx/roc → 三态），
+只是按日期逐日循环（compute_regime 只算最新一天）。
+
+用法：
+  python3 backfill_hk_regime.py            # 回填港股温区全历史
+  python3 backfill_hk_regime.py --days 500 # 只回填最近 500 个交易日
+
+幂等：INSERT OR REPLACE（同日同市场覆盖），可重复执行。
+"""
+import sqlite3
+import sys
+from pathlib import Path
+
+_SCRIPT_DIR = Path(__file__).resolve().parent
+import sys as _sys
+_sys.path.insert(0, str(_SCRIPT_DIR))
+from market_regime import calc_ma, calc_trend_strength, calc_roc, ADX_TREND_MIN
+
+DB_PATH = Path("/home/hmo/MoFin/data/mofin.db")
+INDEX_HK = "hkHSI"
+LOOKBACK = 120  # 与 compute_regime 的 lookback_days 一致
+
+
+def classify(above_ma20, adx):
+    """与 market_regime.compute_regime 的 regime 分类完全一致"""
+    if above_ma20 is True and adx is not None and adx >= ADX_TREND_MIN:
+        return "trend_up"
+    elif adx is not None and adx < ADX_TREND_MIN:
+        return "choppy"
+    else:
+        return "trend_down"
+
+
+def main():
+    days = None
+    for a in sys.argv[1:]:
+        if a.startswith("--days"):
+            days = int(a.split("=")[-1] if "=" in a else sys.argv[sys.argv.index(a) + 1])
+
+    conn = sqlite3.connect(str(DB_PATH), timeout=30)
+    rows = conn.execute(
+        "SELECT date, close, high, low FROM stock_daily WHERE code=? ORDER BY date ASC",
+        (INDEX_HK,)
+    ).fetchall()
+    if not rows or len(rows) < 30:
+        print(f"{INDEX_HK} 数据不足（{len(rows)} 条），无法回填")
+        return 1
+
+    dates = [r[0] for r in rows]
+    closes = [r[1] for r in rows]
+    highs = [r[2] for r in rows]
+    lows = [r[3] for r in rows]
+    n = len(rows)
+    if days:
+        n = min(n, days)
+    print(f"{INDEX_HK} 共 {len(rows)} 条（{dates[0]} ~ {dates[-1]}），回填最近 {n} 个交易日温区")
+
+    # 逐日计算（i 从第 30 天起，窗口用前 LOOKBACK 天）
+    written = 0
+    start_i = max(30, n - days) if days else 30
+    for i in range(start_i, n):
+        # 取截至 i 的最近 LOOKBACK 天
+        j0 = max(0, i - LOOKBACK + 1)
+        c = closes[j0:i + 1]
+        h = highs[j0:i + 1]
+        l = lows[j0:i + 1]
+        if len(c) < 30:
+            continue
+        ma20 = calc_ma(c, 20)
+        trend = calc_trend_strength(h, l, c)
+        roc = calc_roc(c)
+        k = len(c) - 1
+        close = c[k]
+        m20 = ma20[k]
+        above = (close > m20) if m20 else None
+        slope = None
+        if k >= 5 and ma20[k - 5] and ma20[k - 5] > 0 and m20:
+            slope = round((m20 - ma20[k - 5]) / ma20[k - 5] * 100, 3)
+        adx = trend[k] if k < len(trend) else None
+        roc_v = roc[k] if k < len(roc) else None
+        regime = classify(above, adx)
+        conn.execute("""
+            INSERT OR REPLACE INTO market_regime
+                (date, market, above_ma20, ma20_slope, roc, adx, regime, close, created_at)
+            VALUES (?, 'hk', ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
+        """, (
+            dates[i],
+            1 if above else 0,
+            slope,
+            round(roc_v, 3) if roc_v is not None else None,
+            round(adx, 2) if adx is not None else None,
+            regime,
+            close,
+        ))
+        written += 1
+        if written % 500 == 0:
+            conn.commit()
+            print(f"  已回填 {written} 条（至 {dates[i]}）...")
+    conn.commit()
+
+    # 验证
+    total = conn.execute(
+        "SELECT COUNT(*), MIN(date), MAX(date) FROM market_regime WHERE market='hk'"
+    ).fetchone()
+    dist = conn.execute(
+        "SELECT regime, COUNT(*) FROM market_regime WHERE market='hk' GROUP BY regime"
+    ).fetchall()
+    conn.close()
+    print(f"\n回填完成：写入 {written} 条")
+    print(f"market_regime 港股温区：共 {total[0]} 条，{total[1]} ~ {total[2]}")
+    print(f"温区分布: {dict(dist)}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())