"""每日热点数据采集：核心股前100 + 题材涨幅前20，收盘后自动入库 由 main.lifespan 启动后台任务；幂等（按交易日 UNIQUE 去重）。 """ import asyncio import traceback from datetime import datetime, time as dtime, timezone, timedelta from typing import Optional from database import get_connection from services.themes import fetch_theme_list, _build_theme_graph _CST = timezone(timedelta(hours=8)) # 每天采集的后台任务：每 300 秒（5 分钟）检查一次 CHECK_INTERVAL_SECONDS = 300 COLLECT_AFTER_TIME = dtime(15, 0) # 收盘后 15:00 开始允许采集 CORE_STOCK_LIMIT = 100 # 题材领涨股涨幅前100 TOP_THEME_LIMIT = 20 # 题材涨幅前20 HOTMAP_TOP_N = 50 # 热点穿透采样题材数：涨幅榜+热度榜各取前50，合并（与热点穿透页一致） HOTMAP_CORE_LIMIT = 100 # 热点穿透核心股前100（按覆盖题材数降序） CORE_COVER_THRESHOLD = 2 # 热点穿透核心股门槛：覆盖题材数 ≥2 def _is_trading_day(d: datetime) -> bool: """仅按工作日判断：周一至周五视为交易日，不处理法定节假日""" return d.weekday() < 5 def _has_collected(trade_date: str) -> bool: """当日核心股是否已采集""" conn = get_connection() try: row = conn.execute( "SELECT 1 FROM daily_core_stocks WHERE trade_date = ? LIMIT 1", (trade_date,), ).fetchone() return row is not None finally: conn.close() async def collect_daily(trade_date: str, dry_run: bool = False) -> dict: """采集指定交易日数据并入库。 核心股 = 题材领涨股涨幅前100 ∪ 热点穿透核心股（覆盖题材数≥2、按覆盖数降序前100），按股票代码去重。 热点穿透构建失败时降级为仅领涨股前100，不影响当日采集。 Args: trade_date: YYYY-MM-DD dry_run: True 只打印不写库（用于验证） Returns: {"core_count": int, "theme_count": int, "skipped": bool} """ if _has_collected(trade_date): print(f"[collector] {trade_date} 已采集，跳过") return {"core_count": 0, "theme_count": 0, "skipped": True} # 1. 拉取全部题材列表（含领涨股，作为当日全部股票的采样来源） themes = await fetch_theme_list(1, False) if not themes: print(f"[collector] {trade_date} 题材列表为空（东财失败），跳过") return {"core_count": 0, "theme_count": 0, "skipped": True} # 领涨题材映射：securityCode -> [(themeCode, themeName), ...]（用于核心股"所属题材"） lead_themes: dict[str, list[tuple[str, str]]] = {} for t in themes: code = t.get("securityCode") if not code: continue lead_themes.setdefault(code, []).append((t["themeCode"], t["themeName"])) # 2. 热点穿透核心股：覆盖题材数≥2，按覆盖题材数降序前100（去重） # _build_theme_graph 采样 涨幅榜+热度榜 各前 HOTMAP_TOP_N 个题材并拉每股覆盖题材数， # 已按 (-coverCount, -f3) 降序；失败时降级为空集，仅保留领涨股。 hotmap_core: list[dict] = [] graph_theme_name: dict[str, str] = {} try: graph = await _build_theme_graph(1, HOTMAP_TOP_N) graph_theme_name = {t["themeCode"]: t["themeName"] for t in graph.get("themes", [])} hotmap_core = [ s for s in graph.get("stocks", []) if s.get("coverCount", 0) >= CORE_COVER_THRESHOLD ][:HOTMAP_CORE_LIMIT] except Exception as e: print(f"[collector] 热点穿透核心股构建失败，降级为仅领涨股: {e}") # 3. 题材领涨股涨幅前100：按领涨股 f3 降序，去重（同一股票可能是多个题材领涨股） stock_map: dict[str, dict] = {} theme_count: dict[str, int] = {} # securityCode -> 领涨题材数 for t in themes: code = t.get("securityCode") if not code: continue theme_count[code] = theme_count.get(code, 0) + 1 if code not in stock_map or (t.get("f3") or 0) > (stock_map[code].get("f3") or 0): stock_map[code] = { "stock_code": code, "stock_name": t.get("securityName", ""), "f3": t.get("f3"), } f3_core = sorted(stock_map.values(), key=lambda x: -(x["f3"] or 0))[:CORE_STOCK_LIMIT] # 4. 题材涨幅前20：bf3 降序 top_themes = sorted(themes, key=lambda x: -(x.get("bf3") or 0))[:TOP_THEME_LIMIT] # 5. 合并去重：热点穿透核心股在前（覆盖数降序，rank 优先），随后补领涨股涨幅前100 core_stocks: list[dict] = [] seen: set[str] = set() theme_pairs: set[tuple[str, str, str]] = set() # (stock_code, theme_code, theme_name) for s in hotmap_core: code = s["securityCode"] seen.add(code) core_stocks.append({ "stock_code": code, "stock_name": s.get("securityName", ""), "f3": s.get("f3"), "cover_count": s.get("coverCount", 0), }) # 所属题材：采样榜内覆盖的题材 + 全量题材列表中的领涨题材 for tc in s.get("themeCodes", []): theme_pairs.add((code, tc, graph_theme_name.get(tc, tc))) for tc, tn in lead_themes.get(code, []): theme_pairs.add((code, tc, tn)) for s in f3_core: code = s["stock_code"] if code in seen: continue seen.add(code) core_stocks.append({ "stock_code": code, "stock_name": s["stock_name"], "f3": s["f3"], "cover_count": theme_count.get(code, 0), }) for tc, tn in lead_themes.get(code, []): theme_pairs.add((code, tc, tn)) if dry_run: print(f"[collector] {trade_date} 核心股 {len(core_stocks)} 只（热点穿透 {len(hotmap_core)} + 领涨股 {len(f3_core)}），题材前20 {len(top_themes)} 只") return {"core_count": len(core_stocks), "theme_count": len(top_themes), "skipped": False} # 6. 入库（事务，UNIQUE 幂等） conn = get_connection() try: for i, s in enumerate(core_stocks, start=1): conn.execute( "INSERT OR IGNORE INTO daily_core_stocks (trade_date, stock_code, stock_name, f3, cover_count, rank) VALUES (?,?,?,?,?,?)", (trade_date, s["stock_code"], s["stock_name"], s["f3"], s["cover_count"], i), ) for code, tc, tn in theme_pairs: conn.execute( "INSERT OR IGNORE INTO daily_core_stock_themes (trade_date, stock_code, theme_code, theme_name) VALUES (?,?,?,?)", (trade_date, code, tc, tn), ) for i, t in enumerate(top_themes, start=1): conn.execute( "INSERT OR IGNORE INTO daily_top_themes (trade_date, theme_code, theme_name, bf3, hot_rank, rank) VALUES (?,?,?,?,?,?)", (trade_date, t["themeCode"], t["themeName"], t.get("bf3"), t.get("hotRank"), i), ) conn.commit() finally: conn.close() print(f"[collector] {trade_date} 已采集：核心股 {len(core_stocks)} 只，题材前20 {len(top_themes)} 只") return {"core_count": len(core_stocks), "theme_count": len(top_themes), "skipped": False} async def collector_loop(stop: Optional[asyncio.Event] = None) -> None: """后台循环：每个交易日 15:00 后自动采集当日数据（幂等）""" while True: try: now = datetime.now(_CST) if _is_trading_day(now) and now.time() >= COLLECT_AFTER_TIME: trade_date = now.strftime("%Y-%m-%d") if not _has_collected(trade_date): await collect_daily(trade_date) except Exception: print("[collector] 采集异常：") traceback.print_exc() if stop is not None and stop.is_set(): break await asyncio.sleep(CHECK_INTERVAL_SECONDS)