TradingAgents-CN 多数据源同步实战指南Tushare/AKShare/BaoStock 分级与 Fallback 机制详解【免费下载链接】TradingAgents-CN基于多智能体LLM的中文金融交易框架 - TradingAgents中文增强版项目地址: https://gitcode.com/GitHub_Trending/tr/TradingAgents-CNTradingAgents-CN 内置的多数据源同步功能为 A 股行情与财务数据的获取提供了「数据源分级 自动 Fallback」的高可用方案当 Tushare 不可用时自动降级到 AKShare 或 BaoStock并配套提供数据源状态检查、连通性测试、同步建议等 API 与命令行工具。读完本文你将掌握三大数据源的优先级配置、多源同步服务的完整调用链API、Python SDK 与 CLI 三种方式以及选择性同步、全历史同步、多周期数据同步三类进阶用法并理解其底层实现原理。功能定位与核心特性多数据源同步功能解决的是金融数据获取链路中的单点故障问题单一数据源可能因 Token 失效、接口限流、网络波动或服务商维护而中断导致股票基础信息与财务指标无法更新。该功能通过以下三组能力保证数据获取的高可用数据源分级为 Tushare、AKShare、BaoStock 三个数据源定义明确的优先级顺序自动 Fallback 机制主数据源失败时自动切换到备用数据源并记录每次同步实际使用的数据源灵活配置支持指定优先数据源、动态调整优先级、实时状态检查与多周期日线/周线/月线数据同步。在代码层面这一功能的核心实现位于 app/services/data_sources/适配器与管理器与 app/services/multi_source_basics_sync_service.py同步服务对外暴露的 API 路由在 app/routers/multi_source_sync.py。数据源分级与优先级机制默认优先级三个数据源按「数据全面性」划分默认优先级其定义散落在各自适配器的_get_default_priority()方法中数据源默认优先级值说明Tushare3最高专业金融数据 API提供最全面的财务指标支持日线/周线/月线AKShare2中等开源金融数据库提供基础股票信息支持日线/周线/月线BaoStock1最低免费证券数据平台作为最后备用支持日线/周线/月线注意一个容易混淆的细节这里的优先级数值越大越优先。例如 TushareAdapter 的默认优先级实现 返回3并注释 highest priorityBaoStockAdapter 返回1注释 lowest priority。而 DataSourceManager 初始化时按priority降序排序self.adapters.sort(keylambda x: x.priority, reverseTrue)因此adapters列表中的顺序恒为 Tushare → AKShare → BaoStockFallback 遍历时也按此顺序逐个尝试。数据库动态优先级除了代码中的默认优先级系统还支持从 MongoDB 的datasource_groupings集合动态加载 A 股市场market_category_id: a_shares且enabled: true的优先级配置见 manager.py 中的_load_priority_from_database。加载成功后会覆盖适配器的_priority属性未配置的数据源则回退到默认优先级。这意味着运维人员无需修改代码即可通过数据库调整数据源优先级。可用性判定is_available三个适配器的可用性判定逻辑各不相同见各适配器is_available()实现Tushare检查 provider 是否已连接、connected状态与api是否可用未连接时会尝试自动连接tushare_adapter.pyAKShare尝试import akshare抛ImportError即判定不可用akshare_adapter.pyBaoStock同样通过import baostock判定baostock_adapter.py。从源码结构看AKShare 与 BaoStock 的可用性判定是「依赖包是否安装」级别的轻量检查而 Tushare 则是「Token 与连接是否有效」级别的真实连接检查。指定优先数据源的重排序逻辑当调用方通过preferred_sources指定优先数据源时DataSourceManager 会重排适配器顺序将指定名称的适配器按用户给定的顺序排到最前其余适配器保持默认顺序跟在后面。例如传入[akshare, baostock]时即使 Tushare 可用也会优先尝试 AKShare。环境变量配置方法在项目根目录的.env文件中配置以下环境变量# Tushare 配置推荐作为主数据源 TUSHARE_ENABLEDtrue TUSHARE_TOKENyour_tushare_token_here # AKShare 配置免费数据源无需 Token AKSHARE_ENABLEDtrue # BaoStock 配置免费数据源无需 Token BAOSTOCK_ENABLEDtrue # 默认数据源 DEFAULT_CHINA_DATA_SOURCEtushare关于DEFAULT_CHINA_DATA_SOURCE的语义仓库架构文档>GET /api/sync/multi-source/sources/status响应示例实际返回由代码中的DataSourceStatus模型与描述映射生成见 multi_source_sync.py{ success: true, message: Data sources status retrieved successfully, data: [ { name: tushare, priority: 3, available: true, description: 专业金融数据API提供高质量的A股数据和财务指标 (Token来源: 数据库) }, { name: akshare, priority: 2, available: true, description: 开源金融数据库提供基础的股票信息 } ] }说明接口对 Tushare 额外返回token_source字段database或env并在描述中追加 Token 来源标注。获取当前生效数据源GET /api/sync/multi-source/sources/current返回优先级最高且可用的数据源实现上用max(available_adapters, keylambda x: x.priority)选取无可用数据源时返回success: false。获取同步状态与历史# 获取最近一次同步状态含总记录数、新增/更新/错误数、实际使用的数据源列表 GET /api/sync/multi-source/status # 分页获取同步历史支持按状态筛选 GET /api/sync/multi-source/history?page1page_size10statussuccess同步状态持久化在 MongoDB 的sync_status集合中job字段为stock_basics_multi_source历史记录按started_at倒序返回。运行多数据源同步POST /api/sync/multi-source/stock_basics/run?forcefalsepreferred_sourcestushare,akshare请求参数参数类型默认值说明forceboolfalse是否强制运行。当服务已有同步任务在运行时非 force 请求会直接返回当前状态而不会重复启动preferred_sourcesstringnull优先使用的数据源逗号分隔如tushare,akshare注意该接口为同步执行仓库注释说明前端设置了 10 分钟超时multi_source_sync.py。响应中的data.status可能为success、success_with_errors、failed或running。测试数据源连接# 测试全部数据源并发测试每个 10 秒超时 POST /api/sync/multi-source/test-sources # 仅测试指定数据源 POST /api/sync/multi-source/test-sources Content-Type: application/json {source_name: tushare}连通性测试的实现细节multi_source_sync.py每个适配器以 10 秒超时运行轻量级连通性检查对 Tushare 会先强制重置连接状态并重新连接以确保使用数据库中最新的 Token 配置。获取同步建议GET /api/sync/multi-source/recommendations返回primary_source优先级最高的可用数据源、fallback_sources其余备用源、suggestions与warnings。典型场景当没有任何数据源可用时给出配置告警只有一个数据源时建议增加冗余Tushare 不可用时建议优先配置 Tushare 以获取最全面的财务数据。清空同步缓存DELETE /api/sync/multi-source/cache删除sync_status集合中jobstock_basics_multi_source的记录并重置服务运行状态用于同步卡死后的恢复。Python 代码使用在业务代码中可以直接调用同步服务与数据源管理器入口与文档一致from app.services.multi_source_basics_sync_service import get_multi_source_sync_service from app.services.data_sources.manager import DataSourceManager # 获取数据源管理器并查看可用适配器 manager DataSourceManager() available_adapters manager.get_available_adapters() for adapter in available_adapters: print(f{adapter.name} (priority{adapter.priority})) # 运行多数据源同步 service get_multi_source_sync_service() result await service.run_full_sync( forceFalse, preferred_sources[tushare, akshare] ) print(result[status], result[inserted], result[updated], result[errors])get_multi_source_sync_service()返回进程内的单例服务实例multi_source_basics_sync_service.py内部通过asyncio.Lock保证同一时间只有一个同步任务在运行。数据源管理器的 Fallback 方法族除了股票基础信息同步DataSourceManager还提供一系列带 Fallback 的通用方法供 K 线、新闻、实时快照等场景复用manager.py方法作用get_stock_list_with_fallback(preferred_sources)获取股票列表返回(DataFrame, 数据源名)get_daily_basic_with_fallback(trade_date, preferred_sources)获取每日基础财务数据PE、PB、市值等find_latest_trade_date_with_fallback(preferred_sources)查找最新交易日期YYYYMMDD 格式get_realtime_quotes_with_fallback()获取全市场实时快照get_kline_with_fallback(code, period, limit, adj)获取 K 线get_news_with_fallback(code, days, limit, include_announcements)获取新闻与公告所有方法遵循同一模式按优先级遍历可用适配器逐个尝试返回第一个成功结果并附带数据源名称全部失败时返回(None, None)。命令行测试与调试仓库提供了现成的端到端测试脚本 scripts/test_multi_source_sync.py默认向http://localhost:8000发起请求依次测试数据源状态检查、数据源连接测试等环节python scripts/test_multi_source_sync.py对应的手工调试命令# 检查数据源状态 curl http://localhost:8000/api/sync/multi-source/sources/status # 测试数据源连接 curl -X POST http://localhost:8000/api/sync/multi-source/test-sources # 触发一次同步并指定优先数据源 curl -X POST http://localhost:8000/api/sync/multi-source/stock_basics/run?preferred_sourcestushare,akshare同步流程深度解析MultiSourceBasicsSyncService.run_full_sync的完整执行流程multi_source_basics_sync_service.py可归纳为四步1. 数据源检查获取DataSourceManager实例调用get_available_adapters()得到所有可用适配器若无任何可用数据源直接抛出RuntimeError(No available data sources found)对应故障排除章节的症状。2. 股票列表获取调用manager.get_stock_list_with_fallback(preferred_sources)优先从 Tushare 获取完整股票列表失败时自动切换到 AKShare 或 BaoStock返回(stock_df, source_used)二元组。日志形如INFO: Trying to fetch stock list from TushareAdapter INFO: Successfully fetched 5427 stocks from TushareAdapter3. 财务数据获取调用find_latest_trade_date_with_fallback查找最新交易日期再调用get_daily_basic_with_fallback获取当日的基础财务数据按ts_code建立映射。目前财务指标主要依赖 Tushare 的daily_basic接口字段包括total_mv总市值、circ_mv流通市值、pe、pb、ps、turnover_rate、volume_ratio、pe_ttm、pb_mrq、ps_ttm、total_share、float_sharetushare_adapter.py。AKShare 与 BaoStock 暂不支持这些财务指标这也是故障排除中「只有部分股票有扩展字段」问题的根源。4. 数据处理与存储统一数据格式从ts_code如000001.SZ提取 6 位股票代码依据后缀识别交易所.SH/.SZ/.BJfull_symbol 标准化_generate_full_symbol按代码前缀规则生成标准化代码——60/68/90开头补.SS00/30/20开头补.SZ8/4开头补.BJ无法识别时原样返回以保证非空multi_source_basics_sync_service.py批量写入 MongoDB每 500 条记录一批执行bulk_writeorderedFalse以(code, source)为联合查询条件做UpdateOneupsert从而区分同一股票来自不同数据源的记录写入重试_execute_bulk_write_with_retry对超时场景采用指数退避重试2 秒、4 秒、8 秒最多 3 次状态持久化将SyncStatstotal/inserted/updated/errors/status/data_sources_used 等写入sync_status集合。SyncStats的状态流转为idle → running → success / success_with_errors / failed其中success_with_errors表示主流程完成但部分记录处理出错。故障排除1. 所有数据源都不可用症状API 返回No available data sources found实际报错为RuntimeError(No available data sources found)。解决方案检查TUSHARE_ENABLED、AKSHARE_ENABLED、BAOSTOCK_ENABLED等环境变量是否配置正确确认至少安装了一个数据源的依赖包pip show akshare/pip show baostock/pip show tushare验证 Tushare Token 是否有效可通过POST /api/sync/multi-source/test-sources单独测试并注意响应中的 Token 来源标注。2. 只有部分股票有扩展字段症状PE、PB 等财务指标缺失。解决方案确保 Tushare 可用其余数据源暂不支持daily_basic财务指标检查最新交易日期是否正确获取last_trade_date字段验证daily_basic数据是否可正常返回。3. 同步速度慢解决方案优先配置 Tushare数据最全面一次可获取全市场数据检查网络连接与 MongoDB 写入性能利用批量写入与缓存机制降低请求频率见下节性能优化。性能优化与最佳实践数据源选择策略环境推荐配置生产环境优先 Tushare配置 AKShare 作为备用开发环境使用 AKShare 或 BaoStock 降低成本测试环境使用任何可用的数据源缓存与并发控制仓库为多源同步设计了明确的并发与批量策略股票列表缓存 24 小时、财务数据缓存 1 小时文档约定的缓存策略同一时间只允许一个同步任务运行asyncio.Lock_running标志forcetrue可跳过该限制每批 500 条记录批量写库配合orderedFalse与指数退避重试避免 MongoDB 写入超时所有耗时 IO 通过asyncio.to_thread放到线程池执行避免阻塞事件循环。配置与维护建议# 推荐配置生产环境 TUSHARE_ENABLEDtrue TUSHARE_TOKENyour_token AKSHARE_ENABLEDtrue BAOSTOCK_ENABLEDtrue DEFAULT_CHINA_DATA_SOURCEtushare定期通过GET /api/sync/multi-source/sources/status检查数据源状态监控同步成功率与success_with_errors状态定期更新数据源依赖包akshare、baostock、tushare测试故障切换机制可用preferred_sourcesakshare模拟主数据源降级场景。选择性数据同步选择性数据同步允许只更新特定类型的数据适用于增量更新与数据修复场景。cli/tushare_init.py通过--sync-items参数逗号分隔指定同步类型支持的可选值与 CLI 帮助文本一致tushare_init.py取值数据类型basic_info股票基础信息historical历史行情日线weekly周线数据monthly月线数据financial财务数据quotes最新行情news新闻数据# 仅更新历史数据最近30天 python cli/tushare_init.py --full --sync-items historical --historical-days 30 # 仅更新财务数据 python cli/tushare_init.py --full --sync-items financial # 同步多个数据类型 python cli/tushare_init.py --full --sync-items historical,financial,quotes相关 CLI 参数还包括--full完整初始化、--basic-only仅基础信息、--historical-days默认 365 天、--multi-period多周期、--force覆盖已有数据、--batch-size默认 100与--check-only仅检查数据库状态。更多细节可参考 Tushare 数据初始化指南。全历史数据同步阈值机制当--historical-days 365010 年时系统自动切换为全历史模式从 1990-01-01 起同步至今的全部数据historical_days同步范围说明 3650指定天数从当前日期往前推算指定天数 3650全历史从 1990-01-01 至今的所有数据使用示例# Tushare 全历史初始化 python cli/tushare_init.py --full --historical-days 10000 # AKShare 全历史初始化 python cli/akshare_init.py --full --historical-days 10000 # BaoStock 全历史初始化 python cli/baostock_init.py --full --historical-days 10000 # 全历史多周期初始化推荐生产环境 python cli/tushare_init.py --full --multi-period --historical-days 10000数据量参考与注意事项同步模式日线记录数存储空间同步耗时默认 1 年~1,250,000 条~500MB30-60 分钟全历史~8,000,000 条2-5GB2-4 小时适用场景生产环境首次部署获取完整历史数据、长期回测研究、历史数据补全。注意事项耗时较长全历史同步需 2-4 小时建议在非交易时间执行API 限流注意各数据源的调用频率限制存储空间确保有 2-5GB 可用磁盘空间推荐策略首次全历史初始化日常增量更新。更完整的参数说明见 Tushare 数据初始化指南。多周期数据支持支持的数据周期三个数据源均支持多周期历史数据日线数据daily每个交易日的 OHLCV 数据周线数据weekly每周的 OHLCV 数据月线数据monthly每月的 OHLCV 数据。数据存储模型所有周期的数据统一存储在 MongoDB 的stock_daily_quotes集合中通过period字段区分period: daily—— 日线数据period: weekly—— 周线数据period: monthly—— 月线数据同步统计接口的实现印证了这一点get_sync_statistics使用聚合管道按period与data_source分组统计记录数与最新交易日期app/worker/multi_period_sync_service.py。初始化多周期数据# Tushare 多周期初始化默认1年 python cli/tushare_init.py --full --multi-period # 指定历史数据范围6个月 python cli/tushare_init.py --full --multi-period --historical-days 180 # 全历史多周期初始化从1990年至今推荐生产环境 python cli/tushare_init.py --full --multi-period --historical-days 10000查询多周期数据from tradingagents.config.database_manager import get_mongodb_client client get_mongodb_client() db client.get_database(tradingagents) collection db.stock_daily_quotes # 查询日线数据 daily_data list(collection.find({ symbol: 000001, period: daily, data_source: tushare })) # 查询周线数据 weekly_data list(collection.find({ symbol: 000001, period: weekly, data_source: tushare })) # 查询月线数据 monthly_data list(collection.find({ symbol: 000001, period: monthly, data_source: tushare }))多周期同步 API仓库还提供了独立的多周期同步 API 路由app/routers/multi_period_sync.py前缀为/api/multi-period-sync端点功能POST /start启动自定义多周期同步可指定 symbols/periods/data_sources/日期范围POST /start-daily启动日线同步POST /start-weekly启动周线同步POST /start-monthly启动月线同步POST /start-all-history启动全历史多周期同步1990 年至今POST /start-incremental?days_back30启动最近 N 天增量同步GET /statistics获取各周期/数据源的记录统计GET /period-comparison/{symbol}对比同一股票同一交易日各周期数据GET /supported-periods查询支持的周期与数据源组合GET /health健康检查底层服务MultiPeriodSyncService.sync_multi_period_dataapp/worker/multi_period_sync_service.py按「数据源 × 周期」双重循环逐个同步累计各周期的记录数并支持all_history模式自动换算全历史日期范围。扩展新的数据源适配器如需为系统添加新的数据源如东方财富、同花顺等遵循以下步骤继承DataSourceAdapter基类必须实现name、priority、is_available、get_stock_list、get_daily_basic、find_latest_trade_date、get_realtime_quotes、get_kline、get_news等抽象接口见 app/services/data_sources/base.py实现必要的抽象方法其中_get_default_priority()决定默认优先级数值在DataSourceManager中注册在 manager.py 的__init__的self.adapters列表中追加适配器实例添加相应的测试用例可参照 scripts/test_multi_source_sync.py 的端到端测试模式更新文档补充数据源说明与优先级描述。此外DataSourceManager已预留可选的数据一致性检查器DataConsistencyChecker位于 app/services/data_sources/data_consistency_checker.py当依赖可用且可用数据源不少于两个时可对主、次数据源的daily_basic数据进行交叉比对、置信度评分与冲突消解manager.py 的get_daily_basic_with_consistency_check这是向「跨数据源数据对比与自动数据修复」演进的现成基础。未来规划据仓库文档所述多数据源同步的演进方向包括更多数据源支持东方财富 API、同花顺 API、Wind API企业版智能数据源选择基于数据质量自动选择、成本优化算法、实时性能监控数据验证和清洗跨数据源数据对比、异常数据检测、自动数据修复。注意以上为项目规划方向尚未在当前仓库代码中全部实现目前仓库已具备的是一致性检查器基础设施与多源 Fallback 能力。小结多数据源同步是 TradingAgents-CN 数据链路的可靠性基石通过「Tushare AKShare BaoStock」的默认优先级与逐级 Fallback保证股票基础信息与财务指标在任意单点故障下仍可持续供给配合状态检查、连通性测试、同步建议等 API运维可以快速定位数据源问题而选择性同步、全历史同步与多周期同步则覆盖了从首次部署、增量更新到长期回测的完整数据生命周期。文中涉及的源码入口包括 app/services/data_sources/、app/services/multi_source_basics_sync_service.py、app/routers/multi_source_sync.py、app/worker/multi_period_sync_service.py 与 cli/tushare_init.py读者可按需深入阅读。【免费下载链接】TradingAgents-CN基于多智能体LLM的中文金融交易框架 - TradingAgents中文增强版项目地址: https://gitcode.com/GitHub_Trending/tr/TradingAgents-CN创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考