diff --git a/deploy/profile-scripts/hk_flow_collector.py b/deploy/profile-scripts/hk_flow_collector.py new file mode 100644 index 00000000..5c6c8faf --- /dev/null +++ b/deploy/profile-scripts/hk_flow_collector.py @@ -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()) diff --git a/deploy/profile-scripts/hk_fundamentals_history_collector.py b/deploy/profile-scripts/hk_fundamentals_history_collector.py new file mode 100644 index 00000000..ddfeed30 --- /dev/null +++ b/deploy/profile-scripts/hk_fundamentals_history_collector.py @@ -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())