feat: 核心股追踪选股并入热点穿透核心股(覆盖题材数降序前100,去重)

选股口径由仅'题材领涨股涨幅前100'改为:
- 热点穿透核心股(覆盖题材数≥2,按覆盖数降序前100)
- 合并题材领涨股涨幅前100,按股票代码去重

复用 _build_theme_graph 计算每股覆盖题材数;热点穿透构建失败
时降级为仅领涨股前100,不影响当日采集。

Co-Authored-By: Claude <noreply@anthropic.com>

backend/services/daily_collector.py

@@ -9,15 +9,18 @@ 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
+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
+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:
@@ -41,6 +44,9 @@ def _has_collected(trade_date: str) -> bool:
 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 只打印不写库（用于验证）
@@ -58,9 +64,32 @@ async def collect_daily(trade_date: str, dry_run: bool = False) -> dict:
         print(f"[collector] {trade_date} 题材列表为空（东财失败），跳过")
         return {"core_count": 0, "theme_count": 0, "skipped": True}
 
-    # 2. 核心股前100：按领涨股 f3 降序，去重（同一股票可能是多个题材领涨股）
+    # 领涨题材映射：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 -> 覆盖题材数
+    theme_count: dict[str, int] = {}  # securityCode -> 领涨题材数
     for t in themes:
         code = t.get("securityCode")
         if not code:
@@ -72,34 +101,62 @@ async def collect_daily(trade_date: str, dry_run: bool = False) -> dict:
                 "stock_name": t.get("securityName", ""),
                 "f3": t.get("f3"),
             }
-    core_stocks = sorted(
-        stock_map.values(), key=lambda x: -(x["f3"] or 0)
-    )[:CORE_STOCK_LIMIT]
+    f3_core = sorted(stock_map.values(), key=lambda x: -(x["f3"] or 0))[:CORE_STOCK_LIMIT]
 
-    # 3. 题材涨幅前20：bf3 降序
+    # 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)} 只，题材前20 {len(top_themes)} 只")
+        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}
 
-    # 4. 入库（事务，UNIQUE 幂等）
+    # 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"], theme_count.get(s["stock_code"], 0), i),
+                (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),
             )
-        # 核心股所属题材：从 themes 列表（含 securityCode/themeCode/themeName）中
-        # 为每个核心股收集其全部所属题材，写入 daily_core_stock_themes
-        core_codes = {s["stock_code"] for s in core_stocks}
-        for t in themes:
-            if t.get("securityCode") in core_codes:
-                conn.execute(
-                    "INSERT OR IGNORE INTO daily_core_stock_themes (trade_date, stock_code, theme_code, theme_name) VALUES (?,?,?,?)",
-                    (trade_date, t["securityCode"], t["themeCode"], t["themeName"]),
-                )
         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 (?,?,?,?,?,?)",