129 lines
5.2 KiB
Python
129 lines
5.2 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()
|
||
dt = time.time() - t0
|
||
print(f"[{tag}] 完成: {ok}/{len(codes)} 成功, {empty} 无新闻, {fail} 失败, 新增 {new} 条, 耗时 {dt:.0f}s", flush=True)
|
||
# ── 2026-08-17 老莫：推荐必须带所依据的策略 ──
|
||
# 打印池子内每只股票的当前策略快照（SSOT：holding_strategies 最新记录）
|
||
# 知微执行本 job 时据此引用真实策略（版本/重评时间/止损/止盈/买入区/信号/操作），
|
||
# 不得凭 strategy_history 旧快照或记忆自由发挥。
|
||
print("\n===== 当前策略快照（推荐必须引用以下真实策略） =====", flush=True)
|
||
try:
|
||
_cols = [d[1] for d in conn.execute("PRAGMA table_info(holding_strategies)").fetchall()]
|
||
for _code in codes:
|
||
_r = conn.execute(
|
||
"SELECT * FROM holding_strategies WHERE code=? AND status='active' ORDER BY rowid DESC LIMIT 1",
|
||
(_code,)).fetchone()
|
||
if not _r:
|
||
continue
|
||
_d = dict(zip(_cols, _r))
|
||
_sig = _d.get("timing_signal") or "?"
|
||
_sl = _d.get("stop_loss"); _tp = _d.get("take_profit")
|
||
_el = _d.get("entry_low"); _eh = _d.get("entry_high")
|
||
_ver = _d.get("version") or _d.get("decision_type") or "?"
|
||
_ra = (_d.get("reassessed_at") or _d.get("updated_at") or "")[:19]
|
||
_act = str(_d.get("action") or "")[:120]
|
||
print(f"[策略] {_code} {_d.get('name','')} | 版本={_ver} 重评={_ra} | "
|
||
f"信号={_sig} 止损={_sl} 止盈={_tp} 买入区={_el}~{_eh} | 操作: {_act}", flush=True)
|
||
print("===== 策略快照结束 =====", flush=True)
|
||
except Exception as _e:
|
||
print(f"[策略快照输出失败] {_e}", file=sys.stderr)
|
||
conn.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|