chore: 补交港股资金流/历史估值采集器(已在cron运行,补入库)

2026-08-15 11:09:58 +08:00
parent 7db2585c15
commit dc8aa3fd05
2 changed files with 182 additions and 0 deletions
@@ -0,0 +1,89 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""hk_flow_collector.py — 港股个股资金流采集（补12维面板 flow5/flow1）
+
+数据源：东财 push2delay 港股资金流接口（secid=116.{code}，已验证可用）
+  https://push2delay.eastmoney.com/api/qt/stock/fflow/kline/get?secid=116.00700&klt=101&lmt=60
+  klines 格式: ["2026-08-14,主力流入,主力流出"]
+写入：hk_flow_daily 表（code/date/flow_in/flow_out，主力资金流）
+调度：每日盘后（资金流刷新）
+限速：0.3s/只，620只约5分钟。
+"""
+import json
+import sqlite3
+import sys
+import time
+import urllib.request
+
+DB_PATH = "/home/hmo/MoFin/data/mofin.db"
+UA = "Mozilla/5.0"
+SLEEP = 0.3
+DAYS = 60  # 拉60日资金流（够 flow5/flow1）
+
+
+def fetch_hk_flow(code, days=DAYS):
+    """东财港股个股资金流历史（每日主力流入/流出，万元）"""
+    url = (f"https://push2delay.eastmoney.com/api/qt/stock/fflow/kline/get?"
+           f"lmt={days}&klt=101&secid=116.{code}&fields1=f1,f2,f3&fields2=f51,f52,f53")
+    req = urllib.request.Request(url, headers={"User-Agent": UA})
+    opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
+    with opener.open(req, timeout=15) as r:
+        text = r.read().decode("utf-8", errors="replace")
+    data = json.loads(text)
+    klines = (data.get("data") or {}).get("klines") or []
+    out = []
+    for k in klines:
+        parts = k.split(",")
+        if len(parts) < 3:
+            continue
+        try:
+            out.append((parts[0], float(parts[1]), float(parts[2])))
+        except (ValueError, TypeError):
+            pass
+    return out
+
+
+def main():
+    conn = sqlite3.connect(DB_PATH, timeout=30)
+    conn.execute("PRAGMA busy_timeout=30000")
+    conn.execute("""
+        CREATE TABLE IF NOT EXISTS hk_flow_daily (
+            code TEXT NOT NULL, date TEXT NOT NULL,
+            flow_in REAL, flow_out REAL,
+            PRIMARY KEY (code, date)
+        )""")
+    codes = [r[0] for r in conn.execute(
+        "SELECT code FROM hk_connect_stocks WHERE is_active=1 ORDER BY code").fetchall()]
+    print(f"港股资金流采集：{len(codes)} 只 × {DAYS} 日", flush=True)
+
+    ok = fail = 0
+    t0 = time.time()
+    for idx, code in enumerate(codes):
+        try:
+            rows = fetch_hk_flow(code)
+            if not rows:
+                fail += 1
+                continue
+            for d, fin, fout in rows:
+                conn.execute(
+                    "INSERT OR REPLACE INTO hk_flow_daily (code, date, flow_in, flow_out) "
+                    "VALUES (?,?,?,?)", (code, d, fin, fout))
+            ok += 1
+            if (idx + 1) % 100 == 0:
+                conn.commit()
+                print(f"  [{idx+1}/{len(codes)}] ok={ok} fail={fail} | {time.time()-t0:.0f}s", flush=True)
+        except Exception as e:
+            fail += 1
+            if fail <= 5:
+                print(f"  FAIL {code}: {str(e)[:60]}", flush=True)
+        time.sleep(SLEEP)
+    conn.commit()
+    n = conn.execute("SELECT COUNT(*), COUNT(DISTINCT code) FROM hk_flow_daily").fetchone()
+    conn.close()
+    print(f"\n完成：ok={ok} fail={fail} 耗时{time.time()-t0:.0f}s", flush=True)
+    print(f"hk_flow_daily: {n[0]} 条, {n[1]} 只", flush=True)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -0,0 +1,93 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""hk_fundamentals_history_collector.py — 港股历史估值采集（补12维面板 pe_q/pb_q）
+
+数据源：akshare stock_hk_valuation_baidu（百度股市通，PE(TTM)/市净率 近五年每日历史）
+写入：stock_fundamentals_history（与A股同表，code=5位港股，date+code主键）
+调度：一次性补采 + 每日增量（当日估值）
+限速：0.4s/只，620只×2指标约10分钟。
+"""
+import sqlite3
+import sys
+import time
+import warnings
+warnings.filterwarnings("ignore")
+
+DB_PATH = "/home/hmo/MoFin/data/mofin.db"
+SLEEP = 0.4
+
+
+def fetch_hk_valuation(code, indicator, period="近五年"):
+    import akshare as ak
+    try:
+        df = ak.stock_hk_valuation_baidu(symbol=code, indicator=indicator, period=period)
+        if df is None or df.empty:
+            return []
+        out = []
+        for _, row in df.iterrows():
+            d = str(row.get("date"))[:10]
+            v = row.get("value")
+            if d and v is not None and str(v) != "nan":
+                try:
+                    out.append((d, float(v)))
+                except (ValueError, TypeError):
+                    pass
+        return out
+    except Exception:
+        return []
+
+
+def main():
+    conn = sqlite3.connect(DB_PATH, timeout=30)
+    conn.execute("PRAGMA busy_timeout=30000")
+    conn.execute("""
+        CREATE TABLE IF NOT EXISTS stock_fundamentals_history (
+            code TEXT NOT NULL, date TEXT NOT NULL,
+            pe_ttm REAL, pe_static REAL, pb REAL, mcap_total REAL, mcap_flow REAL,
+            PRIMARY KEY (code, date)
+        )""")
+    codes = [r[0] for r in conn.execute(
+        "SELECT code FROM hk_connect_stocks WHERE is_active=1 ORDER BY code").fetchall()]
+    print(f"港股历史估值采集：{len(codes)} 只（百度接口 PE/PB 近五年）", flush=True)
+
+    ok = fail = 0
+    t0 = time.time()
+    for idx, code in enumerate(codes):
+        try:
+            pe_list = fetch_hk_valuation(code, "市盈率(TTM)")
+            pb_list = fetch_hk_valuation(code, "市净率")
+            pe_map = dict(pe_list)
+            pb_map = dict(pb_list)
+            dates = sorted(set(pe_map.keys()) | set(pb_map.keys()))
+            if not dates:
+                fail += 1
+                continue
+            n = 0
+            for d in dates:
+                pe = pe_map.get(d)
+                pb = pb_map.get(d)
+                if pe is None and pb is None:
+                    continue
+                conn.execute(
+                    "INSERT OR REPLACE INTO stock_fundamentals_history (code, date, pe_ttm, pb) "
+                    "VALUES (?,?,?,?)", (code, d, pe, pb))
+                n += 1
+            ok += 1
+            if (idx + 1) % 50 == 0:
+                conn.commit()
+                print(f"  [{idx+1}/{len(codes)}] ok={ok} fail={fail} | {time.time()-t0:.0f}s", flush=True)
+        except Exception as e:
+            fail += 1
+            if fail <= 5:
+                print(f"  FAIL {code}: {str(e)[:60]}", flush=True)
+        time.sleep(SLEEP)
+    conn.commit()
+    n = conn.execute("SELECT COUNT(*), COUNT(DISTINCT code) FROM stock_fundamentals_history WHERE length(code)=5").fetchone()
+    conn.close()
+    print(f"\n完成：ok={ok} fail={fail} 耗时{time.time()-t0:.0f}s", flush=True)
+    print(f"stock_fundamentals_history 港股: {n[0]} 条, {n[1]} 只", flush=True)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())