feat: 缺口4修复——fundamentals_full_refresh全市场基本面刷新(4013只腾讯批量API每批100只~2分钟,cron 16:35)+news 4个包装脚本

2026-08-12 09:20:19 +08:00
parent e876e85a85
commit 958a0d51a4
5 changed files with 132 additions and 0 deletions
@@ -0,0 +1,112 @@
+#!/usr/bin/env python3
+"""fundamentals_full_refresh.py — 全市场基本面刷新（stock_fundamentals 日常刷新）
+
+背景（2026-08-12 架构补缺）：
+  stock_fundamentals 全市场靠回填（3979 只旧），fundamentals_refresh 只刷持仓/自选。
+  p_oversold 的 mcap_q/pe_q 分位需要全市场 PE/PB/市值新鲜数据。
+
+数据源：腾讯批量行情 API（qt.gtimg.cn/q=code，每批 100 只）
+  parts[39]=PE, parts[44]=总市值, parts[45]=流通市值, parts[46]=PB
+  老莫改直连后实测可用。全市场 4013 只 / 100 = 41 批，~2 分钟，单 cron 护栏内。
+
+调度：盘后 35 16 * * 1-5（stock_daily 采集完成后）
+规范：单例守卫(5.3) + INSERT OR REPLACE 幂等 + 批量限速防封 + eps=price/pe
+"""
+import sys, os, re, time, sqlite3, fcntl, urllib.request
+from pathlib import Path
+from datetime import datetime
+
+DB_PATH = Path("/home/hmo/MoFin/data/mofin.db")
+UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
+BATCH = 100         # 腾讯推荐上限 100/批
+SLEEP = 0.2         # 批间隔（防封）
+
+
+def _singleton_guard(tag="fundamentals_full_refresh.py"):
+    lock_dir = Path("/tmp/mofin_locks")
+    lock_dir.mkdir(exist_ok=True)
+    try:
+        fd = os.open(str(lock_dir / f"{tag}.lock"), os.O_CREAT | os.O_RDWR)
+        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        return fd
+    except OSError:
+        print(f"[{tag}] 已有实例在运行，退出", flush=True)
+        sys.exit(0)
+
+
+def prefix_of(code):
+    return "sh" if code.startswith(("5", "6", "9")) else "sz"
+
+
+def fetch_qq_batch(symbols):
+    """腾讯批量行情（每批100只），返回 {code: {pe,pb,price,mcap_total,mcap_flow}}"""
+    if not symbols:
+        return {}
+    results = {}
+    url = f"http://qt.gtimg.cn/q={','.join(symbols)}"
+    try:
+        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("gbk", errors="ignore")
+        for line in text.strip().split("\n"):
+            if "~" not in line:
+                continue
+            parts = line.split("~")
+            if len(parts) < 47:
+                continue
+            code = parts[2]
+            price = float(parts[3]) if parts[3] else 0
+            pe = float(parts[39]) if parts[39] else 0
+            mcap_t = float(parts[44]) if parts[44] else 0   # 总市值（亿）
+            mcap_f = float(parts[45]) if parts[45] else 0   # 流通市值（亿）
+            pb = float(parts[46]) if parts[46] else 0
+            if price > 0:
+                results[code] = {"price": price, "pe": pe, "pb": pb,
+                                 "mcap_total": mcap_t, "mcap_flow": mcap_f}
+    except Exception as e:
+        print(f"  batch fetch error: {str(e)[:60]}", flush=True)
+    return results
+
+
+def main():
+    _fd = _singleton_guard()
+    t0 = time.time()
+    print(f"[fundamentals_full_refresh] {datetime.now().strftime('%H:%M:%S')} 全市场基本面刷新开始", flush=True)
+
+    conn = sqlite3.connect(str(DB_PATH), timeout=30)
+    codes = [str(r[0]) for r in conn.execute(
+        "SELECT DISTINCT code FROM stock_daily WHERE length(code)=6 ORDER BY code").fetchall()]
+    print(f"  股票池: {len(codes)} 只", flush=True)
+
+    cur = conn.cursor()
+    ok = fail = written = 0
+    symbols = [f"{prefix_of(c)}{c}" for c in codes]
+    for i in range(0, len(symbols), BATCH):
+        batch = symbols[i:i + BATCH]
+        data = fetch_qq_batch(batch)
+        for code, d in data.items():
+            eps = round(d["price"] / d["pe"], 2) if d["pe"] > 0 else 0
+            try:
+                cur.execute(
+                    "INSERT OR REPLACE INTO stock_fundamentals (code, pe, pb, eps, mcap_total, mcap_flow, updated_at) "
+                    "VALUES (?,?,?,?,?,?,datetime('now','localtime'))",
+                    (code, d["pe"], d["pb"], eps, d["mcap_total"], d["mcap_flow"]))
+                written += 1
+                ok += 1
+            except Exception:
+                fail += 1
+        conn.commit()
+        done = min(i + BATCH, len(symbols))
+        print(f"  [{done}/{len(symbols)}] ok={ok} fail={fail} written={written} | {time.time()-t0:.0f}s", flush=True)
+        time.sleep(SLEEP)
+
+    conn.commit()
+    total = conn.execute("SELECT COUNT(*) FROM stock_fundamentals").fetchone()[0]
+    conn.close()
+    dt = time.time() - t0
+    print(f"[fundamentals_full_refresh] 完成: {written} 只写入, 表总数 {total}, 耗时 {dt:.0f}s", flush=True)
+
+
+if __name__ == "__main__":
+    main()