TradingAgents-CN Tushare 统一方案 APScheduler 集成实践从 Celery 到原生调度器的一次架构收敛【免费下载链接】TradingAgents-CN基于多智能体LLM的中文金融交易框架 - TradingAgents中文增强版项目地址: https://gitcode.com/GitHub_Trending/tr/TradingAgents-CN导读本文完整记录 TradingAgents-CN 将 Tushare 统一数据同步方案从错误的 Celery 实现迁移到原生 APSchedulerAsyncIOScheduler的过程涵盖架构决策、任务函数设计、调度注册、配置管理、Pydantic 兼容性修复、测试验证与部署运维要点。阅读完本文你将掌握如何让 Tushare 的股票基础信息、实时行情、历史行情、财务数据与状态检查五类同步任务在应用主进程内统一调度并理解为什么这一方案能显著降低部署复杂度与资源消耗。一、集成背景与架构决策1.1 原始问题Celery 方案的架构冲突在本次集成之前Tushare 数据同步任务被错误地实现为 Celery 任务对应文件为app/worker/tasks/tushare_tasks.py与app/worker/tasks/__init__.py本次集成中已删除。Celery 方案带来了一系列与项目现有架构不匹配的问题调度器不兼容项目原本就基于 APScheduler 构建调度体系Celery 的引入形成了两套并行的任务调度模型产生架构冲突额外基础设施Celery 需要独立的 Worker 进程、Beat 调度进程以及 Redis 作为消息中间件任何一环缺失都可能导致任务丢失或无法触发运维复杂度陡增部署时需同时管理应用进程与 Celery 多服务日志分散、监控割裂。1.2 修正方案原生 APScheduler修正后的方案回归项目既有的 APScheduler 技术栈使用AsyncIOScheduler作为调度核心与现有调度系统完美融合所有定时任务统一在同一个调度器中管理在主应用进程中运行无需额外 Worker、Beat 与 Redis 服务简化部署和维护单进程即可承载全部同步能力。从仓库实现看这一决策贯穿了整个应用生命周期。在 app/main.py 中调度器统一以AsyncIOScheduler(timezonesettings.TIMEZONE)创建Tushare 任务与 AKShare、BaoStock 等数据源的同步任务共享同一个调度器实例从源码结构可以推断这正是统一管理、单一调度器设计意图的直接体现。二、任务函数设计与实现2.1 六个 APScheduler 兼容任务函数Tushare 统一方案在 app/worker/tushare_sync_service.py 中定义了 6 个 APScheduler 可直接调用的异步任务函数全部以run_tushare_*命名async def run_tushare_basic_info_sync(force_update: bool False) # 股票基础信息同步 async def run_tushare_quotes_sync(force: bool False) # 实时行情同步 async def run_tushare_historical_sync(incremental: bool True) # 历史数据同步 async def run_tushare_financial_sync() # 财务数据同步 async def run_tushare_status_check() # 状态检查 async def run_tushare_news_sync(hours_back: int 24, max_news_per_stock: int 20) # 新闻数据同步每个任务函数的实现模式高度一致先通过单例工厂get_tushare_sync_service()获取全局同步服务实例该实例惰性初始化首次调用时完成 provider 连接与历史/新闻服务装配再调用对应业务方法并将返回的统计字典写入应用日志async def run_tushare_basic_info_sync(force_update: bool False): APScheduler任务同步股票基础信息 try: service await get_tushare_sync_service() result await service.sync_stock_basic_info(force_update, job_idtushare_basic_info_sync) logger.info(f✅ Tushare基础信息同步完成: {result}) return result except Exception as e: logger.error(f❌ Tushare基础信息同步失败: {e}) raise注意任务函数中job_id与函数名保持一致这是后续进度跟踪与取消控制的关键——TushareSyncService内部通过_should_stop()app/worker/tushare_sync_service.py轮询 MongoDBscheduler_executions集合中的cancel_requested标记从而支持用户在前端手动取消长任务。2.2 底层同步服务的核心逻辑任务函数只是薄壳真正的数据同步逻辑由TushareSyncService承担app/worker/tushare_sync_service.py值得展开的工程细节包括1速率限制Rate Limiter服务在初始化时根据积分等级动态构建速率限制器tushare_tier getattr(settings, TUSHARE_TIER, standard) # free/basic/standard/premium/vip safety_margin float(getattr(settings, TUSHARE_RATE_LIMIT_SAFETY_MARGIN, 0.8)) self.rate_limiter get_tushare_rate_limiter(tiertushare_tier, safety_marginsafety_margin)历史数据与财务数据同步在每只股票处理前都会调用await self.rate_limiter.acquire()排队避免触发 Tushare 的每分钟最多访问类限流。限流错误通过_is_rate_limit_error()识别匹配每分钟最多访问rate limit请求过于频繁等关键词并会在实时行情同步的统计结果中以stopped_by_rate_limit标记。2实时行情的小单切换策略sync_realtime_quotes内置了一个智能策略app/worker/tushare_sync_service.py当指定股票数量 ≤10 只时自动切换到 AKShare 接口同步避免浪费 Tusharert_k接口宝贵的调用配额免费用户每小时仅 2 次大量股票或全市场同步时才使用 Tushare 批量接口一次获取。同步前还会通过_is_trading_time()检查 A 股交易时段周一至周五 9:30-11:30、13:00-15:00非交易时段直接跳过可用forceTrue强制跳过检查。3增量历史同步sync_historical_data支持incrementalTrue增量模式通过_get_last_sync_date()查询该股票在数据库中的最后一条记录日期并加一天作为起始日若无历史数据则回退到上市日期stock_basic_info.list_date甚至1990-01-01全量拉取保证数据无空洞。三、调度注册从主应用到调度器3.1 注册逻辑Tushare 任务的注册集中在 app/main.py 的启动流程中。以基础信息同步为例# 基础信息同步任务 scheduler.add_job( run_tushare_basic_info_sync, CronTrigger.from_crontab(settings.TUSHARE_BASIC_INFO_SYNC_CRON, timezonesettings.TIMEZONE), idtushare_basic_info_sync, name股票基础信息同步Tushare, kwargs{force_update: False} ) if not (settings.TUSHARE_UNIFIED_ENABLED and settings.TUSHARE_BASIC_INFO_SYNC_ENABLED): scheduler.pause_job(tushare_basic_info_sync) logger.info(f⏸️ Tushare基础信息同步已添加但暂停: {settings.TUSHARE_BASIC_INFO_SYNC_CRON}) else: logger.info(f Tushare基础信息同步已配置: {settings.TUSHARE_BASIC_INFO_SYNC_CRON})这段代码体现了两层设计CRON 驱动统一使用CronTrigger.from_crontab()时区取自settings.TIMEZONE默认Asia/Shanghai见 app/core/config.py保证所有定时任务遵循同一时区语义总开关 任务开关的双重控制任务始终被注册但当TUSHARE_UNIFIED_ENABLED与对应任务的*_SYNC_ENABLED未同时为真时通过scheduler.pause_job()将任务置于暂停状态。这种注册但不激活的模式让运维无需改动代码即可在运行时通过配置开关切换任务状态。3.2 五类任务的默认调度配置任务 ID任务函数CRON 表达式默认执行时间说明tushare_basic_info_syncrun_tushare_basic_info_sync0 2 * * *每日凌晨 2 点全量同步股票基础信息force_updateFalsetushare_quotes_syncrun_tushare_quotes_sync*/5 9-15 * * 1-5交易时段每 5 分钟工作日 9:00-15:00 时段内行情更新tushare_historical_syncrun_tushare_historical_sync0 16 * * 1-5工作日 16 点收盘后增量同步历史数据incrementalTruetushare_financial_syncrun_tushare_financial_sync0 3 * * 0周日凌晨 3 点每周财务数据更新最近 20 期约 5 年tushare_status_checkrun_tushare_status_check0 * * * *每小时整点数据源与集合状态监控这些默认值均可在 app/core/config.py 中查到其中财务同步固定通过kwargs语义在任务函数内以limit20传入即每次同步拉取最近 20 个报告期数据。四、配置管理增强4.1 配置项清单统一方案在 app/core/config.py 中通过 Pydantic Settings 定义了完整的配置族# Tushare基础配置 TUSHARE_TOKEN: str Field(default, descriptionTushare API Token) TUSHARE_ENABLED: bool Field(defaultTrue, description启用Tushare数据源) TUSHARE_TIER: str Field(defaultstandard, descriptionTushare积分等级 (free/basic/standard/premium/vip)) TUSHARE_RATE_LIMIT_SAFETY_MARGIN: float Field(default0.8, ge0.1, le1.0, description速率限制安全边际) # Tushare统一数据同步配置 TUSHARE_UNIFIED_ENABLED: bool Field(defaultTrue) TUSHARE_BASIC_INFO_SYNC_ENABLED: bool Field(defaultTrue) TUSHARE_BASIC_INFO_SYNC_CRON: str Field(default0 2 * * *) # 每日凌晨2点 TUSHARE_QUOTES_SYNC_ENABLED: bool Field(defaultTrue) TUSHARE_QUOTES_SYNC_CRON: str Field(default*/5 9-15 * * 1-5) # 交易时间每5分钟 TUSHARE_HISTORICAL_SYNC_ENABLED: bool Field(defaultTrue) TUSHARE_HISTORICAL_SYNC_CRON: str Field(default0 16 * * 1-5) # 工作日16点 TUSHARE_FINANCIAL_SYNC_ENABLED: bool Field(defaultTrue) TUSHARE_FINANCIAL_SYNC_CRON: str Field(default0 3 * * 0) # 周日凌晨3点 TUSHARE_STATUS_CHECK_ENABLED: bool Field(defaultTrue) TUSHARE_STATUS_CHECK_CRON: str Field(default0 * * * *) # 每小时各配置项的作用边界TUSHARE_UNIFIED_ENABLED统一同步总开关关闭后所有 Tushare 任务被暂停*_SYNC_ENABLED5 个任务的独立开关支持总开关开启但单独停用某个任务的精细控制*_SYNC_CRON5 个任务的独立 CRON 表达式默认值按基础信息每日、行情交易时段高频、历史收盘后、财务每周、状态每小时的节奏设计TUSHARE_TIER/TUSHARE_RATE_LIMIT_SAFETY_MARGIN驱动速率限制器影响历史/财务同步的请求排队节奏safety_margin取值范围 0.1-1.0默认 0.8 意为按等级上限的 80% 节流QUOTES_TUSHARE_HOURLY_LIMIT/QUOTES_AUTO_DETECT_TUSHARE_PERMISSIONapp/core/config.py与实时行情配额相关免费用户默认每小时 2 次调用上限付费用户可自动切换到高频模式。4.2 与其他数据源的配置隔离从 app/core/config.py 可以看到AKShareAKSHARE_*_SYNC_CRON如行情每 30 分钟与 BaoStockBAOSTOCK_*_SYNC_CRON如日 K 线收盘后 16 点都遵循完全相同的总开关 任务开关 CRON三段式配置结构。这说明 Tushare 的 APScheduler 集成并非孤立设计而是作为项目多数据源统一同步体系的模板存在——这也呼应了报告长期规划中提到的为 AKShare、BaoStock 应用相同模式而从仓库现状看该模式已经落地。五、关键技术修复Pydantic 模型兼容性5.1 问题现象集成过程中遇到的典型报错为StockBasicInfoExtended object has no attribute get根因是 Tushare Provider 返回的数据对象在不同代码路径下可能是 Pydantic 模型如StockBasicInfoExtended而非字典。数据库服务期望接收字典格式直接调用.get()或按字典下标取值便会抛出AttributeError。这一问题在 Pydantic v1 与 v2 并存的环境中尤为突出v1 用.dict()v2 用.model_dump()。5.2 解决方案统一的三段式转换同步服务在 app/worker/tushare_sync_service.py 中实现了兼容 v1/v2 的安全转换模式# 先转换为字典格式如果是Pydantic模型 if hasattr(stock_info, model_dump): stock_data stock_info.model_dump() # Pydantic v2 elif hasattr(stock_info, dict): stock_data stock_info.dict() # Pydantic v1 else: stock_data stock_info # 本来就是字典同一兼容策略贯穿整个服务批量处理入口_process_basic_info_batch对每只股票先做上述转换再取code字段存量数据读取get_stock_basic_info返回的existing也可能是 Pydantic 模型读取updated_at前同样先转换app/worker/tushare_sync_service.py异常分支兜底异常处理中对code的提取依次尝试stock_info.code、model_dump().get(code)、dict().get(code)、stock_info.get(code)四级降级确保错误统计不因取码失败而二次崩溃app/worker/tushare_sync_service.py行情保存_get_and_save_quotes对单只股票行情对象做同样的 model → dict 转换后再写入update_market_quotes。这一修复保证了所有数据传递给数据库服务时均为字典格式同时向后兼容 Pydantic v1 与 v2 的不同方法签名。六、测试验证体系6.1 测试覆盖范围本次集成配套的测试位于 tests/test_tushare_unified/test_tushare_sync_service.py覆盖四个层面配置读取测试所有TUSHARE_*配置项正确读取CRON 表达式格式合法任务函数测试状态检查任务正常执行、数据库连接与查询正常、Provider 连接正常。测试通过patch(app.worker.tushare_sync_service.get_mongo_db)与patch(app.worker.tushare_sync_service.get_stock_data_service)注入 Mock验证初始化成功/失败、基础信息同步成功/无数据等场景如test_initialize_failure断言RuntimeError消息包含 Tushare连接失败APScheduler 兼容性测试任务可成功添加、CRON 表达式被正确解析、调度器启动与关闭行为正常应用启动集成测试5 个任务全部正确注册、调度器状态正常、时区配置正确。6.2 验证结果集成验证时调度器状态输出如下 调度器状态: 总任务数: 5 任务: tushare_basic_info_sync (函数: run_tushare_basic_info_sync) 任务: tushare_quotes_sync (函数: run_tushare_quotes_sync) 任务: tushare_historical_sync (函数: run_tushare_historical_sync) 任务: tushare_financial_sync (函数: run_tushare_financial_sync) 任务: tushare_status_check (函数: run_tushare_status_check)需要说明的是该结果记录于集成完成时的验证现场当前仓库源码app/main.py在此基础上还包含run_tushare_news_sync新闻同步任务与配套的新闻同步配置任务总数会随配置开关不同而变化读者以实际运行环境输出为准。七、部署指南7.1 环境变量配置在.env文件中按需配置以下环境变量均可选未配置时使用上文默认值# .env 文件配置 TUSHARE_TOKENyour_tushare_token_here TUSHARE_UNIFIED_ENABLEDtrue TUSHARE_BASIC_INFO_SYNC_ENABLEDtrue TUSHARE_BASIC_INFO_SYNC_CRON0 2 * * * TUSHARE_QUOTES_SYNC_ENABLEDtrue TUSHARE_QUOTES_SYNC_CRON*/5 9-15 * * 1-5 TUSHARE_HISTORICAL_SYNC_ENABLEDtrue TUSHARE_HISTORICAL_SYNC_CRON0 16 * * 1-5 TUSHARE_FINANCIAL_SYNC_ENABLEDtrue TUSHARE_FINANCIAL_SYNC_CRON0 3 * * 0 TUSHARE_STATUS_CHECK_ENABLEDtrue TUSHARE_STATUS_CHECK_CRON0 * * * * # 按需调整速率限制可选 TUSHARE_TIERstandard TUSHARE_RATE_LIMIT_SAFETY_MARGIN0.8应用前提需保证 MongoDB 可连接同步结果写入stock_basic_info、market_quotes等标准化集合且TUSHARE_TOKEN有效否则同步任务会因 Provider 连接失败而报错并写入日志。7.2 应用启动由于调度器内嵌于主应用进程无需启动任何额外服务# 正常启动应用即可无需额外服务 python -m app # 或 uvicorn app.main:app --host 0.0.0.0 --port 8000应用启动时即完成AsyncIOScheduler初始化、任务注册与启用/暂停判断随后由调度器按 CRON 触发各任务。7.3 监控与日志所有任务执行状态、进度与统计结果均记录在应用日志中使用标准的应用日志级别logger.info/logger.warning/logger.error与格式历史同步每处理 50 只股票输出一次进度与速率限制器统计当前调用次数、等待次数、总等待时间长任务支持通过 MongoDBscheduler_executions中的cancel_requested标记实现运行时取消前端调度管理页面可据此展示实时进度可参考 docs/guides/scheduler_management.md 与 docs/guides/scheduler_frontend_complete.md。八、与 Celery 方案的对比8.1 量化对比指标Celery方案APScheduler方案改进部署复杂度高需要WorkerBeatRedis低主进程内运行-70%资源消耗高多进程低单进程多任务-50%维护成本高多服务管理低统一管理-60%启动时间慢多服务启动快单服务启动80%监控复杂度高多服务监控低单服务监控-70%系统兼容性低新增依赖高原生集成100%说明上表为集成报告记录的相对改进幅度用于定性呈现方向性收益具体数值会因部署环境不同而有所差异。8.2 核心优势零额外依赖复用项目既有的 APScheduler无需 Redis 或额外 Worker/Beat 服务原生集成与 AKShare、BaoStock 等既有同步任务同处一个调度器架构一致简化部署单一应用进程一条命令即可启动全部同步能力统一管理所有定时任务在同一调度器中管理进度、取消、状态查询走同一套机制资源高效避免多进程开销异步任务在事件循环内协作执行提高资源利用率。九、集成成果与后续演进9.1 已完成事项架构统一消除了 Celery 与 APScheduler 的架构冲突数据同步完全回归原生调度体系功能完整基础信息、实时行情、历史数据、财务数据、状态检查以及当前仓库中新增的新闻同步全部正常工作配置灵活总开关 任务开关 独立 CRON 三层配置支持对每个任务独立控制启用状态与执行节奏测试验证单元测试覆盖初始化、同步成功/失败、批次处理等关键路径集成验证通过文档完善调度管理、前端展示等配套文档齐全可参见 docs/guides/scheduler_management_summary.md 与 docs/guides/scheduler_metadata_feature.md。9.2 后续建议短期优化监控增强为任务执行状态增加 Web 界面监控面板告警机制实施任务失败时的邮件或消息通知性能调优根据实际运行情况调整批处理大小batch_size默认 100与同步频率。长期规划扩展支持将相同的调度模式继续向更多数据源复制当前仓库中 AKShare、BaoStock 已采用相同模式智能调度基于市场状态动态调整同步频率分布式支持如数据量增长可考虑多实例负载均衡方案。本次 APScheduler 集成的核心价值在于一次彻底的架构收敛删除了引入额外基础设施负担的 Celery 实现让 Tushare 数据同步回归项目原生的异步调度体系实现了单进程、零额外依赖、统一管理的目标且配置、测试、日志与进度跟踪全部对齐既有架构具备直接投产的条件。【免费下载链接】TradingAgents-CN基于多智能体LLM的中文金融交易框架 - TradingAgents中文增强版项目地址: https://gitcode.com/GitHub_Trending/tr/TradingAgents-CN创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 TradingAgents-CN Tushare 统一方案 APScheduler 集成实践：从 Celery 到原生调度器的一次架构收敛
TradingAgents-CN Tushare 统一方案 APScheduler 集成实践从 Celery 到原生调度器的一次架构收敛【免费下载链接】TradingAgents-CN基于多智能体LLM的中文金融交易框架 - TradingAgents中文增强版项目地址: https://gitcode.com/GitHub_Trending/tr/TradingAgents-CN导读本文完整记录 TradingAgents-CN 将 Tushare 统一数据同步方案从错误的 Celery 实现迁移到原生 APSchedulerAsyncIOScheduler的过程涵盖架构决策、任务函数设计、调度注册、配置管理、Pydantic 兼容性修复、测试验证与部署运维要点。阅读完本文你将掌握如何让 Tushare 的股票基础信息、实时行情、历史行情、财务数据与状态检查五类同步任务在应用主进程内统一调度并理解为什么这一方案能显著降低部署复杂度与资源消耗。一、集成背景与架构决策1.1 原始问题Celery 方案的架构冲突在本次集成之前Tushare 数据同步任务被错误地实现为 Celery 任务对应文件为app/worker/tasks/tushare_tasks.py与app/worker/tasks/__init__.py本次集成中已删除。Celery 方案带来了一系列与项目现有架构不匹配的问题调度器不兼容项目原本就基于 APScheduler 构建调度体系Celery 的引入形成了两套并行的任务调度模型产生架构冲突额外基础设施Celery 需要独立的 Worker 进程、Beat 调度进程以及 Redis 作为消息中间件任何一环缺失都可能导致任务丢失或无法触发运维复杂度陡增部署时需同时管理应用进程与 Celery 多服务日志分散、监控割裂。1.2 修正方案原生 APScheduler修正后的方案回归项目既有的 APScheduler 技术栈使用AsyncIOScheduler作为调度核心与现有调度系统完美融合所有定时任务统一在同一个调度器中管理在主应用进程中运行无需额外 Worker、Beat 与 Redis 服务简化部署和维护单进程即可承载全部同步能力。从仓库实现看这一决策贯穿了整个应用生命周期。在 app/main.py 中调度器统一以AsyncIOScheduler(timezonesettings.TIMEZONE)创建Tushare 任务与 AKShare、BaoStock 等数据源的同步任务共享同一个调度器实例从源码结构可以推断这正是统一管理、单一调度器设计意图的直接体现。二、任务函数设计与实现2.1 六个 APScheduler 兼容任务函数Tushare 统一方案在 app/worker/tushare_sync_service.py 中定义了 6 个 APScheduler 可直接调用的异步任务函数全部以run_tushare_*命名async def run_tushare_basic_info_sync(force_update: bool False) # 股票基础信息同步 async def run_tushare_quotes_sync(force: bool False) # 实时行情同步 async def run_tushare_historical_sync(incremental: bool True) # 历史数据同步 async def run_tushare_financial_sync() # 财务数据同步 async def run_tushare_status_check() # 状态检查 async def run_tushare_news_sync(hours_back: int 24, max_news_per_stock: int 20) # 新闻数据同步每个任务函数的实现模式高度一致先通过单例工厂get_tushare_sync_service()获取全局同步服务实例该实例惰性初始化首次调用时完成 provider 连接与历史/新闻服务装配再调用对应业务方法并将返回的统计字典写入应用日志async def run_tushare_basic_info_sync(force_update: bool False): APScheduler任务同步股票基础信息 try: service await get_tushare_sync_service() result await service.sync_stock_basic_info(force_update, job_idtushare_basic_info_sync) logger.info(f✅ Tushare基础信息同步完成: {result}) return result except Exception as e: logger.error(f❌ Tushare基础信息同步失败: {e}) raise注意任务函数中job_id与函数名保持一致这是后续进度跟踪与取消控制的关键——TushareSyncService内部通过_should_stop()app/worker/tushare_sync_service.py轮询 MongoDBscheduler_executions集合中的cancel_requested标记从而支持用户在前端手动取消长任务。2.2 底层同步服务的核心逻辑任务函数只是薄壳真正的数据同步逻辑由TushareSyncService承担app/worker/tushare_sync_service.py值得展开的工程细节包括1速率限制Rate Limiter服务在初始化时根据积分等级动态构建速率限制器tushare_tier getattr(settings, TUSHARE_TIER, standard) # free/basic/standard/premium/vip safety_margin float(getattr(settings, TUSHARE_RATE_LIMIT_SAFETY_MARGIN, 0.8)) self.rate_limiter get_tushare_rate_limiter(tiertushare_tier, safety_marginsafety_margin)历史数据与财务数据同步在每只股票处理前都会调用await self.rate_limiter.acquire()排队避免触发 Tushare 的每分钟最多访问类限流。限流错误通过_is_rate_limit_error()识别匹配每分钟最多访问rate limit请求过于频繁等关键词并会在实时行情同步的统计结果中以stopped_by_rate_limit标记。2实时行情的小单切换策略sync_realtime_quotes内置了一个智能策略app/worker/tushare_sync_service.py当指定股票数量 ≤10 只时自动切换到 AKShare 接口同步避免浪费 Tusharert_k接口宝贵的调用配额免费用户每小时仅 2 次大量股票或全市场同步时才使用 Tushare 批量接口一次获取。同步前还会通过_is_trading_time()检查 A 股交易时段周一至周五 9:30-11:30、13:00-15:00非交易时段直接跳过可用forceTrue强制跳过检查。3增量历史同步sync_historical_data支持incrementalTrue增量模式通过_get_last_sync_date()查询该股票在数据库中的最后一条记录日期并加一天作为起始日若无历史数据则回退到上市日期stock_basic_info.list_date甚至1990-01-01全量拉取保证数据无空洞。三、调度注册从主应用到调度器3.1 注册逻辑Tushare 任务的注册集中在 app/main.py 的启动流程中。以基础信息同步为例# 基础信息同步任务 scheduler.add_job( run_tushare_basic_info_sync, CronTrigger.from_crontab(settings.TUSHARE_BASIC_INFO_SYNC_CRON, timezonesettings.TIMEZONE), idtushare_basic_info_sync, name股票基础信息同步Tushare, kwargs{force_update: False} ) if not (settings.TUSHARE_UNIFIED_ENABLED and settings.TUSHARE_BASIC_INFO_SYNC_ENABLED): scheduler.pause_job(tushare_basic_info_sync) logger.info(f⏸️ Tushare基础信息同步已添加但暂停: {settings.TUSHARE_BASIC_INFO_SYNC_CRON}) else: logger.info(f Tushare基础信息同步已配置: {settings.TUSHARE_BASIC_INFO_SYNC_CRON})这段代码体现了两层设计CRON 驱动统一使用CronTrigger.from_crontab()时区取自settings.TIMEZONE默认Asia/Shanghai见 app/core/config.py保证所有定时任务遵循同一时区语义总开关 任务开关的双重控制任务始终被注册但当TUSHARE_UNIFIED_ENABLED与对应任务的*_SYNC_ENABLED未同时为真时通过scheduler.pause_job()将任务置于暂停状态。这种注册但不激活的模式让运维无需改动代码即可在运行时通过配置开关切换任务状态。3.2 五类任务的默认调度配置任务 ID任务函数CRON 表达式默认执行时间说明tushare_basic_info_syncrun_tushare_basic_info_sync0 2 * * *每日凌晨 2 点全量同步股票基础信息force_updateFalsetushare_quotes_syncrun_tushare_quotes_sync*/5 9-15 * * 1-5交易时段每 5 分钟工作日 9:00-15:00 时段内行情更新tushare_historical_syncrun_tushare_historical_sync0 16 * * 1-5工作日 16 点收盘后增量同步历史数据incrementalTruetushare_financial_syncrun_tushare_financial_sync0 3 * * 0周日凌晨 3 点每周财务数据更新最近 20 期约 5 年tushare_status_checkrun_tushare_status_check0 * * * *每小时整点数据源与集合状态监控这些默认值均可在 app/core/config.py 中查到其中财务同步固定通过kwargs语义在任务函数内以limit20传入即每次同步拉取最近 20 个报告期数据。四、配置管理增强4.1 配置项清单统一方案在 app/core/config.py 中通过 Pydantic Settings 定义了完整的配置族# Tushare基础配置 TUSHARE_TOKEN: str Field(default, descriptionTushare API Token) TUSHARE_ENABLED: bool Field(defaultTrue, description启用Tushare数据源) TUSHARE_TIER: str Field(defaultstandard, descriptionTushare积分等级 (free/basic/standard/premium/vip)) TUSHARE_RATE_LIMIT_SAFETY_MARGIN: float Field(default0.8, ge0.1, le1.0, description速率限制安全边际) # Tushare统一数据同步配置 TUSHARE_UNIFIED_ENABLED: bool Field(defaultTrue) TUSHARE_BASIC_INFO_SYNC_ENABLED: bool Field(defaultTrue) TUSHARE_BASIC_INFO_SYNC_CRON: str Field(default0 2 * * *) # 每日凌晨2点 TUSHARE_QUOTES_SYNC_ENABLED: bool Field(defaultTrue) TUSHARE_QUOTES_SYNC_CRON: str Field(default*/5 9-15 * * 1-5) # 交易时间每5分钟 TUSHARE_HISTORICAL_SYNC_ENABLED: bool Field(defaultTrue) TUSHARE_HISTORICAL_SYNC_CRON: str Field(default0 16 * * 1-5) # 工作日16点 TUSHARE_FINANCIAL_SYNC_ENABLED: bool Field(defaultTrue) TUSHARE_FINANCIAL_SYNC_CRON: str Field(default0 3 * * 0) # 周日凌晨3点 TUSHARE_STATUS_CHECK_ENABLED: bool Field(defaultTrue) TUSHARE_STATUS_CHECK_CRON: str Field(default0 * * * *) # 每小时各配置项的作用边界TUSHARE_UNIFIED_ENABLED统一同步总开关关闭后所有 Tushare 任务被暂停*_SYNC_ENABLED5 个任务的独立开关支持总开关开启但单独停用某个任务的精细控制*_SYNC_CRON5 个任务的独立 CRON 表达式默认值按基础信息每日、行情交易时段高频、历史收盘后、财务每周、状态每小时的节奏设计TUSHARE_TIER/TUSHARE_RATE_LIMIT_SAFETY_MARGIN驱动速率限制器影响历史/财务同步的请求排队节奏safety_margin取值范围 0.1-1.0默认 0.8 意为按等级上限的 80% 节流QUOTES_TUSHARE_HOURLY_LIMIT/QUOTES_AUTO_DETECT_TUSHARE_PERMISSIONapp/core/config.py与实时行情配额相关免费用户默认每小时 2 次调用上限付费用户可自动切换到高频模式。4.2 与其他数据源的配置隔离从 app/core/config.py 可以看到AKShareAKSHARE_*_SYNC_CRON如行情每 30 分钟与 BaoStockBAOSTOCK_*_SYNC_CRON如日 K 线收盘后 16 点都遵循完全相同的总开关 任务开关 CRON三段式配置结构。这说明 Tushare 的 APScheduler 集成并非孤立设计而是作为项目多数据源统一同步体系的模板存在——这也呼应了报告长期规划中提到的为 AKShare、BaoStock 应用相同模式而从仓库现状看该模式已经落地。五、关键技术修复Pydantic 模型兼容性5.1 问题现象集成过程中遇到的典型报错为StockBasicInfoExtended object has no attribute get根因是 Tushare Provider 返回的数据对象在不同代码路径下可能是 Pydantic 模型如StockBasicInfoExtended而非字典。数据库服务期望接收字典格式直接调用.get()或按字典下标取值便会抛出AttributeError。这一问题在 Pydantic v1 与 v2 并存的环境中尤为突出v1 用.dict()v2 用.model_dump()。5.2 解决方案统一的三段式转换同步服务在 app/worker/tushare_sync_service.py 中实现了兼容 v1/v2 的安全转换模式# 先转换为字典格式如果是Pydantic模型 if hasattr(stock_info, model_dump): stock_data stock_info.model_dump() # Pydantic v2 elif hasattr(stock_info, dict): stock_data stock_info.dict() # Pydantic v1 else: stock_data stock_info # 本来就是字典同一兼容策略贯穿整个服务批量处理入口_process_basic_info_batch对每只股票先做上述转换再取code字段存量数据读取get_stock_basic_info返回的existing也可能是 Pydantic 模型读取updated_at前同样先转换app/worker/tushare_sync_service.py异常分支兜底异常处理中对code的提取依次尝试stock_info.code、model_dump().get(code)、dict().get(code)、stock_info.get(code)四级降级确保错误统计不因取码失败而二次崩溃app/worker/tushare_sync_service.py行情保存_get_and_save_quotes对单只股票行情对象做同样的 model → dict 转换后再写入update_market_quotes。这一修复保证了所有数据传递给数据库服务时均为字典格式同时向后兼容 Pydantic v1 与 v2 的不同方法签名。六、测试验证体系6.1 测试覆盖范围本次集成配套的测试位于 tests/test_tushare_unified/test_tushare_sync_service.py覆盖四个层面配置读取测试所有TUSHARE_*配置项正确读取CRON 表达式格式合法任务函数测试状态检查任务正常执行、数据库连接与查询正常、Provider 连接正常。测试通过patch(app.worker.tushare_sync_service.get_mongo_db)与patch(app.worker.tushare_sync_service.get_stock_data_service)注入 Mock验证初始化成功/失败、基础信息同步成功/无数据等场景如test_initialize_failure断言RuntimeError消息包含 Tushare连接失败APScheduler 兼容性测试任务可成功添加、CRON 表达式被正确解析、调度器启动与关闭行为正常应用启动集成测试5 个任务全部正确注册、调度器状态正常、时区配置正确。6.2 验证结果集成验证时调度器状态输出如下 调度器状态: 总任务数: 5 任务: tushare_basic_info_sync (函数: run_tushare_basic_info_sync) 任务: tushare_quotes_sync (函数: run_tushare_quotes_sync) 任务: tushare_historical_sync (函数: run_tushare_historical_sync) 任务: tushare_financial_sync (函数: run_tushare_financial_sync) 任务: tushare_status_check (函数: run_tushare_status_check)需要说明的是该结果记录于集成完成时的验证现场当前仓库源码app/main.py在此基础上还包含run_tushare_news_sync新闻同步任务与配套的新闻同步配置任务总数会随配置开关不同而变化读者以实际运行环境输出为准。七、部署指南7.1 环境变量配置在.env文件中按需配置以下环境变量均可选未配置时使用上文默认值# .env 文件配置 TUSHARE_TOKENyour_tushare_token_here TUSHARE_UNIFIED_ENABLEDtrue TUSHARE_BASIC_INFO_SYNC_ENABLEDtrue TUSHARE_BASIC_INFO_SYNC_CRON0 2 * * * TUSHARE_QUOTES_SYNC_ENABLEDtrue TUSHARE_QUOTES_SYNC_CRON*/5 9-15 * * 1-5 TUSHARE_HISTORICAL_SYNC_ENABLEDtrue TUSHARE_HISTORICAL_SYNC_CRON0 16 * * 1-5 TUSHARE_FINANCIAL_SYNC_ENABLEDtrue TUSHARE_FINANCIAL_SYNC_CRON0 3 * * 0 TUSHARE_STATUS_CHECK_ENABLEDtrue TUSHARE_STATUS_CHECK_CRON0 * * * * # 按需调整速率限制可选 TUSHARE_TIERstandard TUSHARE_RATE_LIMIT_SAFETY_MARGIN0.8应用前提需保证 MongoDB 可连接同步结果写入stock_basic_info、market_quotes等标准化集合且TUSHARE_TOKEN有效否则同步任务会因 Provider 连接失败而报错并写入日志。7.2 应用启动由于调度器内嵌于主应用进程无需启动任何额外服务# 正常启动应用即可无需额外服务 python -m app # 或 uvicorn app.main:app --host 0.0.0.0 --port 8000应用启动时即完成AsyncIOScheduler初始化、任务注册与启用/暂停判断随后由调度器按 CRON 触发各任务。7.3 监控与日志所有任务执行状态、进度与统计结果均记录在应用日志中使用标准的应用日志级别logger.info/logger.warning/logger.error与格式历史同步每处理 50 只股票输出一次进度与速率限制器统计当前调用次数、等待次数、总等待时间长任务支持通过 MongoDBscheduler_executions中的cancel_requested标记实现运行时取消前端调度管理页面可据此展示实时进度可参考 docs/guides/scheduler_management.md 与 docs/guides/scheduler_frontend_complete.md。八、与 Celery 方案的对比8.1 量化对比指标Celery方案APScheduler方案改进部署复杂度高需要WorkerBeatRedis低主进程内运行-70%资源消耗高多进程低单进程多任务-50%维护成本高多服务管理低统一管理-60%启动时间慢多服务启动快单服务启动80%监控复杂度高多服务监控低单服务监控-70%系统兼容性低新增依赖高原生集成100%说明上表为集成报告记录的相对改进幅度用于定性呈现方向性收益具体数值会因部署环境不同而有所差异。8.2 核心优势零额外依赖复用项目既有的 APScheduler无需 Redis 或额外 Worker/Beat 服务原生集成与 AKShare、BaoStock 等既有同步任务同处一个调度器架构一致简化部署单一应用进程一条命令即可启动全部同步能力统一管理所有定时任务在同一调度器中管理进度、取消、状态查询走同一套机制资源高效避免多进程开销异步任务在事件循环内协作执行提高资源利用率。九、集成成果与后续演进9.1 已完成事项架构统一消除了 Celery 与 APScheduler 的架构冲突数据同步完全回归原生调度体系功能完整基础信息、实时行情、历史数据、财务数据、状态检查以及当前仓库中新增的新闻同步全部正常工作配置灵活总开关 任务开关 独立 CRON 三层配置支持对每个任务独立控制启用状态与执行节奏测试验证单元测试覆盖初始化、同步成功/失败、批次处理等关键路径集成验证通过文档完善调度管理、前端展示等配套文档齐全可参见 docs/guides/scheduler_management_summary.md 与 docs/guides/scheduler_metadata_feature.md。9.2 后续建议短期优化监控增强为任务执行状态增加 Web 界面监控面板告警机制实施任务失败时的邮件或消息通知性能调优根据实际运行情况调整批处理大小batch_size默认 100与同步频率。长期规划扩展支持将相同的调度模式继续向更多数据源复制当前仓库中 AKShare、BaoStock 已采用相同模式智能调度基于市场状态动态调整同步频率分布式支持如数据量增长可考虑多实例负载均衡方案。本次 APScheduler 集成的核心价值在于一次彻底的架构收敛删除了引入额外基础设施负担的 Celery 实现让 Tushare 数据同步回归项目原生的异步调度体系实现了单进程、零额外依赖、统一管理的目标且配置、测试、日志与进度跟踪全部对齐既有架构具备直接投产的条件。【免费下载链接】TradingAgents-CN基于多智能体LLM的中文金融交易框架 - TradingAgents中文增强版项目地址: https://gitcode.com/GitHub_Trending/tr/TradingAgents-CN创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考