feat: 新闻采集分层——盘前全量(BATCHES 4→8,每批534只×0.9s=481s安全) + 盘中池子增量(pool_news_collector,自选+持仓82只×0.9s=74s,*/30盘中高频)

2026-08-14 01:57:32 +08:00
parent 563029bab6
commit 23686e28a1
2 changed files with 104 additions and 1 deletions
@@ -22,7 +22,7 @@ from news_collector import fetch_stock_news, init_table

 DB_PATH = Path("/home/hmo/MoFin/data/mofin.db")
 SLEEP = 0.4          # 请求间隔（防东财封）
-BATCHES = 4          # 4 批
+BATCHES = 8          # 8 批（2026-08-13 由4改8：4272只/4批=1068只×0.9s=961s超600s超时，改8批每批534只×0.9s=481s安全）
 RECENT_DAYS = 3      # 只存近 3 天新闻（news3 因子口径）


@@ -0,0 +1,103 @@
+#!/usr/bin/env python3
+"""pool_news_collector.py — 池子新闻采集（自选+持仓，盘中高频）
+
+2026-08-13 新增（老莫架构）：盘前全量跑（选股），盘中只跑池子里的（自选+持仓），增加频率。
+- 盘前：news_collector_full b0-b7（全市场 4272 只，8批）
+- 盘中：pool_news_collector（池子 82 只，*/30 盘中）
+
+池子 = 持仓(holdings shares>0) + 自选(holding_strategies 自选策略)
+"""
+import sys, os, json, time, sqlite3
+from datetime import datetime, timedelta
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent))
+from news_collector import fetch_stock_news, init_table
+
+DB_PATH = Path("/home/hmo/MoFin/data/mofin.db")
+SLEEP = 0.4          # 请求间隔（防东财封）
+RECENT_DAYS = 3      # 只存近 3 天新闻（news3 因子口径）
+
+
+def _singleton_guard(tag):
+    lock_dir = Path("/tmp/mofin_locks")
+    lock_dir.mkdir(exist_ok=True)
+    try:
+        import fcntl
+        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:
+        return None
+
+
+def get_pool_codes(conn):
+    """池子：持仓 + 自选（去重）"""
+    codes = set()
+    # 持仓
+    for r in conn.execute("SELECT DISTINCT code FROM holdings WHERE is_active=1 AND shares>0"):
+        codes.add(r[0])
+    # 自选
+    for r in conn.execute("SELECT DISTINCT code FROM holding_strategies WHERE status='active' AND decision_type='自选策略'"):
+        codes.add(r[0])
+    return sorted(codes)
+
+
+def main():
+    tag = "pool_news_collector"
+    guard = _singleton_guard(tag)
+    if guard is None:
+        print(f"[{tag}] 已有实例在运行，跳过", file=sys.stderr)
+        return
+
+    conn = sqlite3.connect(DB_PATH)
+    init_table(conn)
+    cur = conn.cursor()
+
+    codes = get_pool_codes(conn)
+    print(f"[{tag}] {datetime.now().strftime('%H:%M:%S')} 池子新闻采集 开始", flush=True)
+    print(f"  池子: {len(codes)} 只（持仓+自选）", flush=True)
+
+    since = (datetime.now() - timedelta(days=RECENT_DAYS)).strftime("%Y-%m-%d")
+    t0 = time.time()
+    ok = empty = fail = new = 0
+
+    for idx, code in enumerate(codes, 1):
+        try:
+            arts = fetch_stock_news(code, max_pages=1)
+            if not arts:
+                empty += 1
+                continue
+            n = 0
+            for a in arts:
+                dstr = (a.get("date") or "")[:10]
+                if dstr and dstr >= since:
+                    try:
+                        cur.execute(
+                            "INSERT OR IGNORE INTO stock_news (code, title, content, source, date, url) "
+                            "VALUES (?,?,?,?,?,?)",
+                            (code, a.get("title", ""), a.get("content", ""),
+                             a.get("source", "eastmoney"), a.get("date", ""), a.get("url", "")))
+                        if cur.rowcount:
+                            n += 1
+                    except Exception:
+                        pass
+            new += n
+            ok += 1
+        except Exception as e:
+            fail += 1
+            if fail <= 5:
+                print(f"  FAIL {code}: {str(e)[:60]}", flush=True)
+        if idx % 20 == 0:
+            conn.commit()
+            print(f"  [{idx}/{len(codes)}] ok={ok} empty={empty} fail={fail} 新增={new} | {time.time()-t0:.0f}s", flush=True)
+        time.sleep(SLEEP)
+
+    conn.commit()
+    conn.close()
+    dt = time.time() - t0
+    print(f"[{tag}] 完成: {ok}/{len(codes)} 成功, {empty} 无新闻, {fail} 失败, 新增 {new} 条, 耗时 {dt:.0f}s", flush=True)
+
+
+if __name__ == "__main__":
+    main()