"""盘中调度器 使用 APScheduler 管理盘前/盘中/盘后定时任务。 """ import logging import traceback from datetime import datetime from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.cron import CronTrigger from app.engine.recommender import refresh_recommendations from app.engine.watchlist import analyze_watchlist_for_all_users from app.api.websocket import broadcast_update logger = logging.getLogger(__name__) scheduler = AsyncIOScheduler(timezone="Asia/Shanghai") async def _run_scan(session_name: str): """执行一次扫描并推送结果""" logger.info(f"=== 定时扫描: {session_name} ({datetime.now().strftime('%H:%M:%S')}) ===") try: result = await refresh_recommendations(scan_session=session_name) rec_count = len(result.get("recommendations", [])) logger.info(f"扫描完成: {rec_count} 只推荐股票") # 通过 WebSocket 推送更新 await broadcast_update({ "type": "scan_update", "session": session_name, "count": rec_count, "timestamp": datetime.now().isoformat(), }) except Exception as e: logger.error(f"定时扫描失败 ({session_name}): {e}") from app.db.error_logger import log_error await log_error("scheduler", f"定时扫描失败 ({session_name}): {e}", detail=traceback.format_exc()) async def _run_watchlist_analysis(): """收盘后自动分析所有用户自选股。""" logger.info("=== 开始自选股定时分析 ===") try: count = await analyze_watchlist_for_all_users(mode="scheduled") logger.info(f"自选股定时分析完成: {count} 条") except Exception as e: logger.error(f"自选股定时分析失败: {e}") from app.db.error_logger import log_error await log_error("scheduler", f"自选股定时分析失败: {e}", detail=traceback.format_exc()) def setup_scheduler(): """配置所有定时任务(交易日时间)""" # 盘前准备 09:00 - 计算前一日市场温度和板块数据 scheduler.add_job( _run_scan, CronTrigger(hour=9, minute=0, day_of_week="mon-fri"), args=["pre_market"], id="pre_market", replace_existing=True ) # 盘中扫描:按交易节奏执行,避免高频重复计算 scan_schedule = [ ("morning_open_0935", 9, 35, "morning_open"), ("morning_open_0950", 9, 50, "morning_open"), ("morning_mid_1020", 10, 20, "morning_mid"), ("morning_mid_1050", 10, 50, "morning_mid"), ("morning_mid_1120", 11, 20, "morning_mid"), ("afternoon_1310", 13, 10, "afternoon"), ("afternoon_1340", 13, 40, "afternoon"), ("late_1410", 14, 10, "late_session"), ("late_1440", 14, 40, "late_session"), ("close_1500", 15, 0, "late_session"), ] for job_id, hour, minute, session_name in scan_schedule: scheduler.add_job( _run_scan, CronTrigger(hour=hour, minute=minute, day_of_week="mon-fri"), args=[session_name], id=job_id, replace_existing=True, ) # 收盘总结 16:00(Tushare 日线数据通常在 15:30 后更新完成) scheduler.add_job( _run_scan, CronTrigger(hour=16, minute=0, day_of_week="mon-fri"), args=["post_market"], id="post_market", replace_existing=True ) scheduler.add_job( _run_watchlist_analysis, CronTrigger(hour=16, minute=20, day_of_week="mon-fri"), id="watchlist_analysis", replace_existing=True ) logger.info("盘中调度器已配置完成") def start_scheduler(): setup_scheduler() scheduler.start() logger.info("调度器已启动") def stop_scheduler(): if scheduler.running: scheduler.shutdown() logger.info("调度器已停止")