Sakurasan
82006f7cf9
原逻辑只取题材涨幅前20入库。现改为从 4 个榜单 (sortField 1=涨幅/3=强度/4=热度排名/5=成交额) 各取前50, 按 themeCode 合并去重后写入 daily_top_themes, 覆盖更全面的当日活跃题材。
246 lines
11 KiB
Python
246 lines
11 KiB
Python
"""每日热点数据采集：核心股前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.cache import clear_all
|
|
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, 1) # 收盘后 15:01 开始允许采集（留 1 分钟等收盘数据稳定）
|
|
CORE_STOCK_LIMIT = 100 # 题材领涨股涨幅前100
|
|
RANK_TOP_LIMIT = 50 # 每个榜单（涨幅/强度/热度/成交额）取前50
|
|
HOTMAP_TOP_N = 100 # 热点穿透采样题材数：涨幅榜+热度榜各取前100，合并（与热点穿透页一致）
|
|
HOTMAP_CORE_LIMIT = 100 # 热点穿透核心股前100（按覆盖题材数降序）
|
|
CORE_COVER_THRESHOLD = 2 # 热点穿透核心股门槛：覆盖题材数 ≥2
|
|
|
|
# 题材榜单：sortField 1=涨幅(bf3) 3=强度(strengthValue) 4=热度排名(hotRank) 5=成交额(fex5)
|
|
# 每个榜单取前 RANK_TOP_LIMIT 个，按 themeCode 合并去重后入库
|
|
THEME_RANK_LISTS = ((1, "bf3"), (3, "strengthValue"), (4, "hotRank"), (5, "fex5"))
|
|
|
|
# 每日缓存清理：交易日 9:31 清空全部缓存，保证开盘后数据全新
|
|
CLEANUP_TIME = dtime(9, 31)
|
|
|
|
|
|
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），按股票代码去重。
|
|
题材 = 涨幅/强度/热度/成交额 4 榜各前50，按 themeCode 合并去重。
|
|
热点穿透构建失败时降级为仅领涨股前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. 题材 4 榜（涨幅/强度/热度/成交额）各前50，按 themeCode 合并去重。
|
|
# 涨幅榜列表与开头用于构建领涨股映射的 fetch_theme_list(1, False) 同缓存，直接复用。
|
|
merged: dict[str, dict] = {}
|
|
rank_list = await fetch_theme_list(1, False)
|
|
for sort_field, _ in THEME_RANK_LISTS:
|
|
lst = rank_list if sort_field == 1 else await fetch_theme_list(sort_field, False)
|
|
for t in lst[:RANK_TOP_LIMIT]:
|
|
merged.setdefault(t["themeCode"], t)
|
|
top_themes = list(merged.values())
|
|
|
|
# 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)}），题材 {len(top_themes)} 只（4 榜前 {RANK_TOP_LIMIT} 合并）")
|
|
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)} 只，题材 {len(top_themes)} 只（4 榜前 {RANK_TOP_LIMIT} 合并）")
|
|
return {"core_count": len(core_stocks), "theme_count": len(top_themes), "skipped": False}
|
|
|
|
|
|
async def collector_loop(stop: Optional[asyncio.Event] = None) -> None:
|
|
"""后台循环：每个交易日 15:01 后自动采集当日数据（幂等）"""
|
|
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)
|
|
|
|
|
|
def _next_cleanup_dt(now: datetime) -> datetime:
|
|
"""计算下一个缓存清理时刻：最近一个工作日 9:31（今天已过则取下一个工作日）"""
|
|
for days in range(0, 8):
|
|
d = (now + timedelta(days=days)).date()
|
|
if d.weekday() >= 5: # 跳过周末
|
|
continue
|
|
dt = datetime(d.year, d.month, d.day, CLEANUP_TIME.hour, CLEANUP_TIME.minute, tzinfo=_CST)
|
|
if dt > now:
|
|
return dt
|
|
return now + timedelta(days=1) # 兜底：理论不可达
|
|
|
|
|
|
async def cache_cleanup_loop(stop: Optional[asyncio.Event] = None) -> None:
|
|
"""后台循环：每个交易日 9:31 清空全部缓存，保证开盘后读到全新数据
|
|
|
|
到点前精确 sleep 至 9:31；若进程在 9:31 后启动，则等下一个交易日。
|
|
"""
|
|
while True:
|
|
try:
|
|
now = datetime.now(_CST)
|
|
next_dt = _next_cleanup_dt(now)
|
|
delay = max(0, int((next_dt - now).total_seconds()))
|
|
if stop is not None:
|
|
try:
|
|
await asyncio.wait_for(stop.wait(), timeout=delay)
|
|
except asyncio.TimeoutError:
|
|
pass # 到点
|
|
if stop.is_set():
|
|
break
|
|
else:
|
|
await asyncio.sleep(delay)
|
|
|
|
clear_all()
|
|
print(f"[cache-cleanup] 已清空全部缓存: {datetime.now(_CST).strftime('%Y-%m-%d %H:%M:%S')}")
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception:
|
|
print("[cache-cleanup] 清理异常：")
|
|
traceback.print_exc()
|
|
await asyncio.sleep(60) # 出错 1 分钟后再试
|