feat: 交易日9:31清空全部缓存,保证开盘后数据全新

- services/cache.py 新增 clear_all() 清空整个 cache 表
- daily_collector.py 新增 cache_cleanup_loop() 后台循环:精确 sleep 到最近一个
  工作日 9:31 清空缓存,跳过周末,出错1分钟重试
- main.py lifespan 启动清理任务,与采集任务一同优雅退出

与既有"非交易时段 TTL 截止到下次开盘前"的惰性过期形成双保险

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Sakurasan
2026-08-11 16:56:56 +08:00
co-authored by Claude
parent 7fe8074e22
commit eb66ab02d0
3 changed files with 65 additions and 6 deletions
+9 -6
View File
@@ -8,7 +8,7 @@ from dotenv import load_dotenv
from database import init_db
from routes import stock, collections, shares, sectors, themes, core_stocks
from services.daily_collector import collector_loop
from services.daily_collector import collector_loop, cache_cleanup_loop
load_dotenv()
@@ -17,14 +17,17 @@ load_dotenv()
async def lifespan(app: FastAPI):
init_db()
collector_task = asyncio.create_task(collector_loop())
cache_cleanup_task = asyncio.create_task(cache_cleanup_loop())
try:
yield
finally:
collector_task.cancel()
try:
await collector_task
except asyncio.CancelledError:
pass
for t in (collector_task, cache_cleanup_task):
t.cancel()
for t in (collector_task, cache_cleanup_task):
try:
await t
except asyncio.CancelledError:
pass
app = FastAPI(title="AUV API", version="1.0.0", lifespan=lifespan)
+10
View File
@@ -56,3 +56,13 @@ def clean_expired():
conn.commit()
finally:
conn.close()
def clear_all():
"""清空全部缓存(每日开盘后 9:31 调用,保证当日数据全新)"""
conn = get_connection()
try:
conn.execute("DELETE FROM cache")
conn.commit()
finally:
conn.close()
+46
View File
@@ -9,6 +9,7 @@ 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))
@@ -22,6 +23,9 @@ HOTMAP_TOP_N = 50 # 热点穿透采样题材数:涨幅榜+热度榜
HOTMAP_CORE_LIMIT = 100 # 热点穿透核心股前100(按覆盖题材数降序)
CORE_COVER_THRESHOLD = 2 # 热点穿透核心股门槛:覆盖题材数 ≥2
# 每日缓存清理:交易日 9:31 清空全部缓存,保证开盘后数据全新
CLEANUP_TIME = dtime(9, 31)
def _is_trading_day(d: datetime) -> bool:
"""仅按工作日判断:周一至周五视为交易日,不处理法定节假日"""
@@ -185,3 +189,45 @@ async def collector_loop(stop: Optional[asyncio.Event] = None) -> None:
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 分钟后再试