104 lines
3.6 KiB
Python
104 lines
3.6 KiB
Python
#!/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()
|