TradingAgents-CN 新闻数据系统实战指南多数据源聚合、智能分析与高性能存储【免费下载链接】TradingAgents-CN基于多智能体LLM的中文金融交易框架 - TradingAgents中文增强版项目地址: https://gitcode.com/GitHub_Trending/tr/TradingAgents-CN新闻数据是金融交易决策的重要输入。TradingAgents-CN 新闻数据系统为 A 股市场提供了从新闻采集、智能分析到存储查询的完整链路解决方案支撑分析师智能体News Analyst的舆情研判与投资决策。阅读本文你将掌握系统的三层架构设计、AKShare/Tushare/实时聚合三类数据源的接入方式、情绪分析与重要性评估的实现原理、MongoDB 存储与索引优化策略以及完整的 REST API 与 Python SDK 调用方法可直接在项目中落地使用。一、系统定位与整体架构新闻数据系统在 TradingAgents-CN 中承担着信息输入层的角色它将东方财富、CCTV 财经、新浪财经、Tushare 及实时聚合源等多路新闻统一采集、标准化、打标分类/情绪/重要性/关键词后写入 MongoDB再由上层分析链路按需查询。系统的核心能力体现在四个方面多数据源新闻获取AKShare、Tushare、实时新闻聚合三路并进互为补充智能数据分析新闻分类、情绪分析sentiment、重要性评估importance、关键词提取高性能存储基于 MongoDB 的标准化集合设计与多索引优化支持毫秒级查询与全文检索灵活查询提供 REST API 与 Python SDK 双入口覆盖基础查询、高级查询、全文搜索、统计聚合与数据清理。三层架构设计系统采用经典的三层架构各层职责清晰、通过依赖注入解耦app/routers/news_data.py 是 API 层app/services/news_data_service.py 与 app/worker/news_data_sync_service.py 构成业务服务层数据提供层则由 AKShare、Tushare 提供者与实时新闻聚合器组成┌─────────────────────────────────────────────────────────────┐ │ RESTful API 层 │ │ app/routers/news_data.py - 新闻数据API接口 │ └─────────────────────────────────────────────────────────────┘ │ ┌─────────────────────────────────────────────────────────────┐ │ 业务服务层 │ │ app/services/news_data_service.py - 新闻数据管理服务 │ │ app/worker/news_data_sync_service.py - 新闻数据同步服务 │ └─────────────────────────────────────────────────────────────┘ │ ┌─────────────────────────────────────────────────────────────┐ │ 数据提供层 │ │ AKShare Provider - 东方财富、CCTV财经、新浪财经新闻 │ │ Tushare Provider - Tushare新闻数据 │ │ Realtime Provider - 实时新闻聚合 │ └─────────────────────────────────────────────────────────────┘从路由注册代码可以看到news_data路由在 app/main.py 中通过app.include_router(news_data.router, tags[news-data])挂载到 FastAPI 应用所有接口统一以/api/news-data为前缀见 app/routers/news_data.py 中的router APIRouter(prefix/api/news-data, tags[新闻数据])。所有接口均通过Depends(get_current_user)接入认证需要登录态才能调用。二、多数据源新闻获取1. AKShare 新闻源AKShare 提供者封装在 tradingagents/dataflows/providers/china/akshare.py 中通过get_stock_news_sync同步版返回原始 DataFrame与get_stock_news异步版返回结构化列表两个入口对外提供服务个股新闻调用东方财富的ak.stock_news_em(symbol...)接口传入 6 位 A 股代码市场新闻调用ak.news_cctv()CCTV 财经等接口获取宏观市场资讯数据字段标题、内容、摘要、链接、来源、作者、发布时间。一个值得注意的实现细节是由于stock_news_em()接口本身未设置必要的请求头AKShare 提供者在初始化时会对requests.get进行补丁见 tradingagents/dataflows/providers/china/akshare.py 中_akshare_headers_patched相关逻辑对eastmoney.com域名的请求注入浏览器 User-Agent 并加入请求延迟以规避反爬虫限制如果环境中可用还会优先使用curl_cffi模拟真实浏览器 TLS 指纹。获取新闻遇到json.JSONDecodeError时会按max_retries次重试并间隔retry_delay秒提升采集稳定性。2. Tushare 新闻源Tushare 提供者封装在 tradingagents/dataflows/providers/china/tushare.py 中其get_stock_news方法签名如下async def get_stock_news(self, symbol: str None, limit: int 10, hours_back: int 24, src: str None) - Optional[List[Dict[str, Any]]]:个股新闻与市场新闻symbol为空时获取市场新闻否则获取对应个股新闻数据字段标题、内容、来源、发布时间、重要性权限提示Tushare 新闻接口需要单独开通权限。同步服务在处理异常时会特别识别权限/permission/unauthorized积分不足/point等关键词并输出可读性更高的警告日志见 app/worker/news_data_sync_service.py 的_sync_tushare_news便于运维人员快速定位是权限问题还是网络问题。3. 实时新闻聚合实时聚合器实现在 tradingagents/dataflows/news/realtime_news.py 的RealtimeNewsAggregator类中核心入口为def get_realtime_stock_news(self, ticker: str, hours_back: int 6, max_news: int 10) - List[NewsItem]其内部按专业 API 新闻 API 搜索引擎的优先级依次尝试多个信息源最终统一封装为NewsItem包含title、content、source等字段。聚合器具备两级去重能力标题去重对标题做小写、去空白归一化后比对标题长度 ≤10 的过短新闻会被直接过滤URL/内容维度去重结合标题与链接识别重复项去重后会按publish_time倒序排序输出并通过日志输出原始 N 条 → 去重后 M 条、移除重复 X 条、标题过短 Y 条的统计信息方便观察聚合质量。三、智能数据分析分类、情绪与重要性同步服务在将新闻入库前会对每条新闻自动执行四步智能分析实现在 app/worker/news_data_sync_service.py新闻分类_classify_news_category基于标题关键词进行规则分类优先级从高到低类别标识触发关键词说明company_announcement年报、季报、业绩、财报、公告公司公告类policy_news政策、央行、监管、法规政策监管类market_news市场、行情、指数、板块市场行情类research_report研报、分析、评级、推荐研究报告类general其余兜底分类情绪分析_analyze_sentiment采用词典计数法对标题 内容全文打分正向词表增长、上涨、利好、盈利、成功、突破、创新、优秀负向词表下跌、亏损、风险、问题、困难、下滑、减少、警告规则正向词命中数 负向词命中数 →positive反之为negative相等则为neutral。情绪分析结果会写入sentiment字段并可在查询时作为过滤条件如sentimentpositive筛选利好新闻。需要说明的是这是项目内置的轻量词典方案文档中给出的情绪词典示例positive: [利好, 上涨, 增长, 盈利, 突破]、negative: [利空, 下跌, 亏损, 风险, 暴跌]、neutral: [公告, 会议, 发布, 披露, 变更]对应同一套思路读者可在此基础上按扩展情绪分析词典的路径自定义更细的规则。重要性评估_assess_importance同样基于标题关键词高重要性high重大、紧急、突发、年报、业绩、重组、收购中重要性medium公告、通知、变更、调整、计划低重要性low未命中上述关键词的一般新闻。关键词提取_extract_keywords从预置的常见金融关键词库业绩、年报、季报、增长、利润、营收、股价、投资、市场、行业、政策、监管、风险、机会、创新、发展等中扫描文本命中项最多返回 10 个关键词写入keywords数组供后续筛选与标签云统计使用。四、高性能存储MongoDB 集合设计与索引优化stock_news 集合结构新闻数据统一写入 MongoDB 的stock_news集合见 app/services/news_data_service.py 的_get_collection每条文档结构如下// stock_news 集合结构 { _id: ObjectId, symbol: 000001, // 股票代码 symbols: [000001, 000002], // 多股票代码 full_symbol: 000001.SZ, // 完整股票代码含市场后缀 market: CN, // 市场标识 title: 新闻标题, // 新闻标题 content: 新闻内容, // 新闻内容 summary: 新闻摘要, // 新闻摘要 url: https://..., // 新闻链接 source: 东方财富, // 新闻来源 author: 记者姓名, // 作者 publish_time: ISODate, // 发布时间 category: company_announcement, // 新闻类别 sentiment: positive, // 情绪分析 sentiment_score: 0.8, // 情绪得分 importance: high, // 重要性 keywords: [关键词1, 关键词2], // 关键词 data_source: akshare, // 数据源 region: CN, // 地区 created_at: ISODate, // 创建时间 updated_at: ISODate, // 更新时间 version: 1 // 数据结构版本 }标准化逻辑_standardize_news_data在入库前完成字段规整其中几个关键点多代码关联若新闻同时涉及多只股票symbols数组会收录全部代码symbol字段保留主代码当symbol存在但不在symbols中时会自动将其置为首元素见 app/services/news_data_service.py完整代码推导_get_full_symbol根据市场与代码前缀推导带后缀代码——60/68开头 →.SH00/30开头 →.SZ时间解析容错_parse_datetime依次尝试%Y-%m-%d %H:%M:%S、%Y-%m-%dT%H:%M:%S、%Y-%m-%dT%H:%M:%SZ、%Y-%m-%d四种格式解析失败时回退为当前 UTC 时间保证数据不因脏时间格式而丢弃注意标准化结构刻意不包含language字段以避免与 MongoDB 文本索引冲突源码注释中明确说明。优化索引设计服务在首次写入数据前会通过_ensure_indexes自动检查并创建索引全部使用backgroundTrue后台创建避免阻塞线上写入源码中的索引清单如下// 唯一约束防止重复新闻URL标题发布时间 db.stock_news.createIndex({url: 1, title: 1, publish_time: 1}, {unique: true, background: true}) // 股票代码 db.stock_news.createIndex({symbol: 1}, {background: true}) // 多股票代码 db.stock_news.createIndex({symbols: 1}, {background: true}) // 发布时间倒序 db.stock_news.createIndex({publish_time: -1}, {background: true}) // 股票时间复合常用查询 db.stock_news.createIndex({symbol: 1, publish_time: -1}, {background: true}) // 数据源 db.stock_news.createIndex({data_source: 1}, {background: true}) // 新闻类别 db.stock_news.createIndex({category: 1}, {background: true}) // 情绪 db.stock_news.createIndex({sentiment: 1}, {background: true}) // 重要性 db.stock_news.createIndex({importance: 1}, {background: true}) // 更新时间数据维护 db.stock_news.createIndex({updated_at: -1}, {background: true})文档中给出的15 个优化索引方案还进一步覆盖了symbolspublish_time复合索引、symbolcategorypublish_time、sentimentimportancepublish_time复合索引以及title/content/summary的三字段全文索引text类型与created_at索引可在数据量增长后按需补齐。全文索引是search_news全文搜索能力的底层支撑。索引创建失败不会阻止服务启动仅输出警告日志兼顾了可用性与健壮性。写入去重与批量写入写入采用pymongo.ReplaceOnebulk_write的批量 Upsert 策略以url title publish_time三元组作为唯一过滤条件命中则整条替换、未命中则插入见save_news_data的实现。这意味着同一新闻即使被多个数据源重复抓到也不会产生重复记录——配合唯一索引形成双保险。BulkWriteError会被捕获并转化为部分成功统计单条失败不影响整批数据落库。五、API 接口详解所有接口定义于 app/routers/news_data.py响应统一通过ok()包装为{success, message, data}结构。1. 新闻查询接口查询股票新闻智能获取优先数据库无数据时实时抓取GET /api/news-data/query/000001?hours_back24limit20categorycompany_announcement该接口的智能逻辑值得关注见query_stock_news的实现先按symbol hours_back从数据库查询若结果为空则自动调用 AKShare 同步服务provider.get_stock_news(symbol, limit)实时抓取并保存入库再重新查询返回。响应中的data_source字段会标明本次数据来自database还是realtime日志中也会输出 数据库无新闻数据实时获取的提示。高级查询多维条件组合POST /api/news-data/query Content-Type: application/json { symbol: 000001, symbols: [000001, 000002], start_time: 2024-01-01T00:00:00Z, end_time: 2024-12-31T23:59:59Z, category: company_announcement, sentiment: positive, importance: high, data_source: akshare, keywords: [业绩], limit: 50, skip: 0 }对应服务端NewsQueryParams支持的全部维度见 app/services/news_data_service.py其中symbols使用$in匹配多代码keywords会转换为$text文本查询排序默认按publish_time倒序。获取最新新闻GET /api/news-data/latest?symbol000001limit10hours_back24symbol为空时返回全市场最新新闻实现上等价于按publish_time倒序的限时查询。全文搜索GET /api/news-data/search?query银行symbol000001limit20基于 MongoDB 文本索引执行$text搜索并按textScore相关性排序返回见search_news实现。2. 新闻统计接口获取统计信息GET /api/news-data/statistics?symbol000001days_back7服务端通过 MongoDB 聚合管道一次完成分组统计见get_news_statistics$match过滤时间与代码范围后$group聚合出总量、正/负/中性情绪计数、高/中/低重要性计数并对category、data_source字段做分布统计。响应示例{ success: true, data: { symbol: 000001, days_back: 7, statistics: { total_count: 25, sentiment_distribution: { positive: 10, negative: 5, neutral: 10 }, importance_distribution: { high: 8, medium: 12, low: 5 }, categories: { company_announcement: 15, market_news: 8, industry_news: 2 }, sources: { 东方财富: 20, 新浪财经: 3, CCTV财经: 2 } } } }3. 新闻同步接口启动同步任务后台异步执行POST /api/news-data/sync/start Content-Type: application/json { symbol: 000001, data_sources: [akshare, tushare], hours_back: 24, max_news_per_source: 50 }该接口通过 FastAPIBackgroundTasks将任务放入后台执行立即返回任务已启动响应symbol为空时自动切换为市场新闻同步。data_sources支持tushare、akshare、realtime三选或组合不传则默认按[tushare, akshare, realtime]全量同步市场新闻同步默认仅realtime见 app/worker/news_data_sync_service.py 的sync_stock_news与sync_market_news实现。同步单只股票同步等待结果POST /api/news-data/sync/single?symbol000001hours_back24max_news_per_source50同步执行的版本响应中包含详细的同步统计信息{ success: true, data: { symbol: 000001, sync_stats: { total_processed: 60, successful_saves: 55, failed_saves: 5, duplicate_skipped: 10, sources_used: [tushare, akshare], duration_seconds: 12.3, success_rate: 91.7 } } }其中success_rate为successful_saves / total_processed × 100duplicate_skipped来自同步服务内部按标题 URL的去重_deduplicate_news。4. 管理接口清理过期新闻DELETE /api/news-data/cleanup?days_to_keep90按publish_time now - days_to_keep批量删除过期新闻返回删除条数。健康检查GET /api/news-data/health同时探测数据服务与同步服务的可用性返回service_status: healthy。六、Python SDK 使用示例无需走 HTTP在项目内部可直接以异步方式调用服务层。1. 获取新闻数据服务from app.services.news_data_service import get_news_data_service, NewsQueryParams # 获取服务实例全局单例 service await get_news_data_service()2. 查询新闻数据from datetime import datetime, timedelta # 查询最新新闻 latest_news await service.get_latest_news(symbol000001, limit10) # 高级查询 params NewsQueryParams( symbol000001, start_timedatetime.utcnow() - timedelta(days7), categorycompany_announcement, sentimentpositive, limit20 ) news_list await service.query_news(params) # 全文搜索 search_results await service.search_news(银行, symbol000001, limit10)3. 新闻数据同步from app.worker.news_data_sync_service import get_news_data_sync_service # 获取同步服务 sync_service await get_news_data_sync_service() # 同步股票新闻 stats await sync_service.sync_stock_news( symbol000001, data_sources[akshare], hours_back24, max_news_per_source50 ) print(f同步完成: {stats.successful_saves} 条成功保存)4. 直接保存标准化数据对于自定义抓取的数据可调用save_news_data直接入库支持传入单条 dict 或列表data_source与market参数必填saved_count await service.save_news_data( news_data[{title: ..., url: ..., publish_time: 2024-06-01 10:00:00}], data_sourceakshare, marketCN )服务还提供了同步版本save_news_data_sync内部使用同步 PyMongo 客户端get_mongo_db_sync适用于非异步上下文如定时任务、脚本。七、配置说明环境变量配置在.env文件中配置如下变量# .env 文件 TUSHARE_TOKENyour_tushare_token_here AKSHARE_TIMEOUT60 MONGODB_URLmongodb://localhost:27017 MONGODB_DBtradingagentsTUSHARE_TOKENTushare 接口令牌新闻接口需单独开通权限AKSHARE_TIMEOUTAKShare 请求超时秒数默认 60 秒MONGODB_URL/MONGODB_DBMongoDB 连接串与数据库名新闻数据存放于该库的stock_news集合。数据源与同步配置以下为同步服务的默认行为配置对应源码中的默认参数# 数据源优先级配置 DATA_SOURCE_PRIORITY { akshare: 1, # 优先使用AKShare tushare: 2, # 其次使用Tushare realtime: 3 # 最后使用实时聚合 } # 同步配置 SYNC_CONFIG { default_hours_back: 24, max_news_per_source: 50, batch_size: 100, retry_times: 3 }实际编码中数据源优先级通过data_sources列表的传参顺序与默认值体现个股同步默认[tushare, akshare, realtime]市场新闻同步默认[realtime]。项目还提供了NEWS_SYNC_MAX_PER_SOURCE等设置项用于控制定时同步的单源数量上限见 app/main.py 中定时任务的调用。八、性能优化策略1. 数据库优化索引优化唯一索引、单字段索引与复合索引组合覆盖按代码、时间、类别、情绪、重要性等全部查询路径配合全文索引支撑毫秒级检索批量操作入库统一走bulk_write批量 Upsert显著降低网络往返与写放大连接池通过 app/core/database.py 管理 MongoDB 连接池支撑高并发访问。2. 缓存策略系统层面配合查询缓存、数据缓存与统计缓存热点查询结果直接复用统计信息定期刷新避免重复聚合计算。3. 并发处理异步处理查询与保存均为async/await全异步实现API 层天然支持高并发批量同步同步服务支持传入多数据源并行抓取定时任务可对自选股批量调度限流控制AKShare 补丁中内置请求延迟规避数据源限流与反爬封禁Tushare 侧按积分权限自动降级。九、监控与日志日志级别约定INFO正常操作日志如同步完成: N 条保存成功新闻数据保存完成: N 条记录WARNING部分数据获取失败、数据源不可用、Tushare 权限/积分不足、日期解析失败、索引创建警告等ERROR数据库连接失败、API 调用异常、批量写入失败等。日志中大量使用 emoji 前缀 开始同步、✅ 获取成功、❌ 获取失败、 保存完成、️ 删除过期便于在日志流中快速检索关键事件。关键指标监控同步成功率来自NewsSyncStats.success_rate是数据链路健康度的核心指标查询性能服务层记录了查询耗时与返回条数可观测query_news的日志数据质量通过去重统计duplicate_skipped与保存成功率间接反映系统健康/api/news-data/health接口提供一键探测。十、故障排除常见问题排查数据库连接失败检查 MongoDB 服务状态mongod是否运行验证MONGODB_URL连接字符串与MONGODB_DB库名配置确认网络连接与 MongoDB 认证信息正确。新闻数据获取失败检查数据源 API 可用性东方财富/CCTV 接口是否可访问验证TUSHARE_TOKEN配置及新闻接口权限、积分是否充足确认网络访问权限与反爬限制AKShare 侧留意是否被限流。查询性能慢检查索引是否生效通过 MongoDBexplain()查看执行计划优化查询条件优先使用symbol publish_time等已建索引的组合考虑为热点查询增加缓存。同步数据重复确认唯一索引url title publish_time已创建验证同步服务内_deduplicate_news去重逻辑是否被正确调用对存量脏数据执行DELETE /api/news-data/cleanup或手动清理。十一、扩展开发指南添加新数据源以现有 AKShare/Tushare 提供者为模板两者均实现get_stock_news与is_available等接口见 tradingagents/dataflows/providers/china/akshare.py 与 tradingagents/dataflows/providers/china/tushare.py继承BaseProvider基类实现get_stock_news方法按_standardize_news_data的结构输出标准化字段symbol/title/content/summary/url/source/publish_time 等在 app/worker/news_data_sync_service.py 的sync_stock_news中注册新数据源分支在data_sources参数中声明新源名称即可被调度。自定义分析算法扩展情绪分析词典向_analyze_sentiment的正向/负向词表追加领域词汇或引入中文情感词库/大模型打分替代词典计数优化重要性评估规则扩充_assess_importance的高/中重要性关键词或结合消息类型如监管函、股权质押加权添加新的分类标准在_classify_news_category中增加类别分支并在文档中同步更新枚举说明实现自定义分析指标可在标准化阶段为文档追加自定义字段如热度分、舆情扩散度配合 MongoDB 聚合管道输出。十二、总结TradingAgents-CN 新闻数据系统是连接外部资讯与智能分析链路的枢纽三层架构职责清晰三路数据源互为备份规则化的智能分析分类、情绪、重要性、关键词让原始新闻在入库瞬间即完成结构化打标MongoDB 的批量 Upsert 与多索引设计保证了写入去重与毫秒级查询REST API 与 Python SDK 双入口则让上层分析器与外部调用方都能便捷消费新闻数据。配合智能降级数据库无数据自动实时抓取、容错日志与可扩展的数据源接口这套系统为股票投资分析提供了可靠、高效、可观测的新闻数据基础设施。延伸阅读本文所述架构与服务在 docs/architecture、docs/features/news/NEWS_SYNC_FEATURE.md 与 docs/features/news/NEWS_SENTIMENT_ANALYSIS.md 中有更深入的设计文档Tushare 新闻接入的专项说明可参考 docs/guides/tushare_news_integration/README.md。【免费下载链接】TradingAgents-CN基于多智能体LLM的中文金融交易框架 - TradingAgents中文增强版项目地址: https://gitcode.com/GitHub_Trending/tr/TradingAgents-CN创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 TradingAgents-CN 新闻数据系统实战指南：多数据源聚合、智能分析与高性能存储
TradingAgents-CN 新闻数据系统实战指南多数据源聚合、智能分析与高性能存储【免费下载链接】TradingAgents-CN基于多智能体LLM的中文金融交易框架 - TradingAgents中文增强版项目地址: https://gitcode.com/GitHub_Trending/tr/TradingAgents-CN新闻数据是金融交易决策的重要输入。TradingAgents-CN 新闻数据系统为 A 股市场提供了从新闻采集、智能分析到存储查询的完整链路解决方案支撑分析师智能体News Analyst的舆情研判与投资决策。阅读本文你将掌握系统的三层架构设计、AKShare/Tushare/实时聚合三类数据源的接入方式、情绪分析与重要性评估的实现原理、MongoDB 存储与索引优化策略以及完整的 REST API 与 Python SDK 调用方法可直接在项目中落地使用。一、系统定位与整体架构新闻数据系统在 TradingAgents-CN 中承担着信息输入层的角色它将东方财富、CCTV 财经、新浪财经、Tushare 及实时聚合源等多路新闻统一采集、标准化、打标分类/情绪/重要性/关键词后写入 MongoDB再由上层分析链路按需查询。系统的核心能力体现在四个方面多数据源新闻获取AKShare、Tushare、实时新闻聚合三路并进互为补充智能数据分析新闻分类、情绪分析sentiment、重要性评估importance、关键词提取高性能存储基于 MongoDB 的标准化集合设计与多索引优化支持毫秒级查询与全文检索灵活查询提供 REST API 与 Python SDK 双入口覆盖基础查询、高级查询、全文搜索、统计聚合与数据清理。三层架构设计系统采用经典的三层架构各层职责清晰、通过依赖注入解耦app/routers/news_data.py 是 API 层app/services/news_data_service.py 与 app/worker/news_data_sync_service.py 构成业务服务层数据提供层则由 AKShare、Tushare 提供者与实时新闻聚合器组成┌─────────────────────────────────────────────────────────────┐ │ RESTful API 层 │ │ app/routers/news_data.py - 新闻数据API接口 │ └─────────────────────────────────────────────────────────────┘ │ ┌─────────────────────────────────────────────────────────────┐ │ 业务服务层 │ │ app/services/news_data_service.py - 新闻数据管理服务 │ │ app/worker/news_data_sync_service.py - 新闻数据同步服务 │ └─────────────────────────────────────────────────────────────┘ │ ┌─────────────────────────────────────────────────────────────┐ │ 数据提供层 │ │ AKShare Provider - 东方财富、CCTV财经、新浪财经新闻 │ │ Tushare Provider - Tushare新闻数据 │ │ Realtime Provider - 实时新闻聚合 │ └─────────────────────────────────────────────────────────────┘从路由注册代码可以看到news_data路由在 app/main.py 中通过app.include_router(news_data.router, tags[news-data])挂载到 FastAPI 应用所有接口统一以/api/news-data为前缀见 app/routers/news_data.py 中的router APIRouter(prefix/api/news-data, tags[新闻数据])。所有接口均通过Depends(get_current_user)接入认证需要登录态才能调用。二、多数据源新闻获取1. AKShare 新闻源AKShare 提供者封装在 tradingagents/dataflows/providers/china/akshare.py 中通过get_stock_news_sync同步版返回原始 DataFrame与get_stock_news异步版返回结构化列表两个入口对外提供服务个股新闻调用东方财富的ak.stock_news_em(symbol...)接口传入 6 位 A 股代码市场新闻调用ak.news_cctv()CCTV 财经等接口获取宏观市场资讯数据字段标题、内容、摘要、链接、来源、作者、发布时间。一个值得注意的实现细节是由于stock_news_em()接口本身未设置必要的请求头AKShare 提供者在初始化时会对requests.get进行补丁见 tradingagents/dataflows/providers/china/akshare.py 中_akshare_headers_patched相关逻辑对eastmoney.com域名的请求注入浏览器 User-Agent 并加入请求延迟以规避反爬虫限制如果环境中可用还会优先使用curl_cffi模拟真实浏览器 TLS 指纹。获取新闻遇到json.JSONDecodeError时会按max_retries次重试并间隔retry_delay秒提升采集稳定性。2. Tushare 新闻源Tushare 提供者封装在 tradingagents/dataflows/providers/china/tushare.py 中其get_stock_news方法签名如下async def get_stock_news(self, symbol: str None, limit: int 10, hours_back: int 24, src: str None) - Optional[List[Dict[str, Any]]]:个股新闻与市场新闻symbol为空时获取市场新闻否则获取对应个股新闻数据字段标题、内容、来源、发布时间、重要性权限提示Tushare 新闻接口需要单独开通权限。同步服务在处理异常时会特别识别权限/permission/unauthorized积分不足/point等关键词并输出可读性更高的警告日志见 app/worker/news_data_sync_service.py 的_sync_tushare_news便于运维人员快速定位是权限问题还是网络问题。3. 实时新闻聚合实时聚合器实现在 tradingagents/dataflows/news/realtime_news.py 的RealtimeNewsAggregator类中核心入口为def get_realtime_stock_news(self, ticker: str, hours_back: int 6, max_news: int 10) - List[NewsItem]其内部按专业 API 新闻 API 搜索引擎的优先级依次尝试多个信息源最终统一封装为NewsItem包含title、content、source等字段。聚合器具备两级去重能力标题去重对标题做小写、去空白归一化后比对标题长度 ≤10 的过短新闻会被直接过滤URL/内容维度去重结合标题与链接识别重复项去重后会按publish_time倒序排序输出并通过日志输出原始 N 条 → 去重后 M 条、移除重复 X 条、标题过短 Y 条的统计信息方便观察聚合质量。三、智能数据分析分类、情绪与重要性同步服务在将新闻入库前会对每条新闻自动执行四步智能分析实现在 app/worker/news_data_sync_service.py新闻分类_classify_news_category基于标题关键词进行规则分类优先级从高到低类别标识触发关键词说明company_announcement年报、季报、业绩、财报、公告公司公告类policy_news政策、央行、监管、法规政策监管类market_news市场、行情、指数、板块市场行情类research_report研报、分析、评级、推荐研究报告类general其余兜底分类情绪分析_analyze_sentiment采用词典计数法对标题 内容全文打分正向词表增长、上涨、利好、盈利、成功、突破、创新、优秀负向词表下跌、亏损、风险、问题、困难、下滑、减少、警告规则正向词命中数 负向词命中数 →positive反之为negative相等则为neutral。情绪分析结果会写入sentiment字段并可在查询时作为过滤条件如sentimentpositive筛选利好新闻。需要说明的是这是项目内置的轻量词典方案文档中给出的情绪词典示例positive: [利好, 上涨, 增长, 盈利, 突破]、negative: [利空, 下跌, 亏损, 风险, 暴跌]、neutral: [公告, 会议, 发布, 披露, 变更]对应同一套思路读者可在此基础上按扩展情绪分析词典的路径自定义更细的规则。重要性评估_assess_importance同样基于标题关键词高重要性high重大、紧急、突发、年报、业绩、重组、收购中重要性medium公告、通知、变更、调整、计划低重要性low未命中上述关键词的一般新闻。关键词提取_extract_keywords从预置的常见金融关键词库业绩、年报、季报、增长、利润、营收、股价、投资、市场、行业、政策、监管、风险、机会、创新、发展等中扫描文本命中项最多返回 10 个关键词写入keywords数组供后续筛选与标签云统计使用。四、高性能存储MongoDB 集合设计与索引优化stock_news 集合结构新闻数据统一写入 MongoDB 的stock_news集合见 app/services/news_data_service.py 的_get_collection每条文档结构如下// stock_news 集合结构 { _id: ObjectId, symbol: 000001, // 股票代码 symbols: [000001, 000002], // 多股票代码 full_symbol: 000001.SZ, // 完整股票代码含市场后缀 market: CN, // 市场标识 title: 新闻标题, // 新闻标题 content: 新闻内容, // 新闻内容 summary: 新闻摘要, // 新闻摘要 url: https://..., // 新闻链接 source: 东方财富, // 新闻来源 author: 记者姓名, // 作者 publish_time: ISODate, // 发布时间 category: company_announcement, // 新闻类别 sentiment: positive, // 情绪分析 sentiment_score: 0.8, // 情绪得分 importance: high, // 重要性 keywords: [关键词1, 关键词2], // 关键词 data_source: akshare, // 数据源 region: CN, // 地区 created_at: ISODate, // 创建时间 updated_at: ISODate, // 更新时间 version: 1 // 数据结构版本 }标准化逻辑_standardize_news_data在入库前完成字段规整其中几个关键点多代码关联若新闻同时涉及多只股票symbols数组会收录全部代码symbol字段保留主代码当symbol存在但不在symbols中时会自动将其置为首元素见 app/services/news_data_service.py完整代码推导_get_full_symbol根据市场与代码前缀推导带后缀代码——60/68开头 →.SH00/30开头 →.SZ时间解析容错_parse_datetime依次尝试%Y-%m-%d %H:%M:%S、%Y-%m-%dT%H:%M:%S、%Y-%m-%dT%H:%M:%SZ、%Y-%m-%d四种格式解析失败时回退为当前 UTC 时间保证数据不因脏时间格式而丢弃注意标准化结构刻意不包含language字段以避免与 MongoDB 文本索引冲突源码注释中明确说明。优化索引设计服务在首次写入数据前会通过_ensure_indexes自动检查并创建索引全部使用backgroundTrue后台创建避免阻塞线上写入源码中的索引清单如下// 唯一约束防止重复新闻URL标题发布时间 db.stock_news.createIndex({url: 1, title: 1, publish_time: 1}, {unique: true, background: true}) // 股票代码 db.stock_news.createIndex({symbol: 1}, {background: true}) // 多股票代码 db.stock_news.createIndex({symbols: 1}, {background: true}) // 发布时间倒序 db.stock_news.createIndex({publish_time: -1}, {background: true}) // 股票时间复合常用查询 db.stock_news.createIndex({symbol: 1, publish_time: -1}, {background: true}) // 数据源 db.stock_news.createIndex({data_source: 1}, {background: true}) // 新闻类别 db.stock_news.createIndex({category: 1}, {background: true}) // 情绪 db.stock_news.createIndex({sentiment: 1}, {background: true}) // 重要性 db.stock_news.createIndex({importance: 1}, {background: true}) // 更新时间数据维护 db.stock_news.createIndex({updated_at: -1}, {background: true})文档中给出的15 个优化索引方案还进一步覆盖了symbolspublish_time复合索引、symbolcategorypublish_time、sentimentimportancepublish_time复合索引以及title/content/summary的三字段全文索引text类型与created_at索引可在数据量增长后按需补齐。全文索引是search_news全文搜索能力的底层支撑。索引创建失败不会阻止服务启动仅输出警告日志兼顾了可用性与健壮性。写入去重与批量写入写入采用pymongo.ReplaceOnebulk_write的批量 Upsert 策略以url title publish_time三元组作为唯一过滤条件命中则整条替换、未命中则插入见save_news_data的实现。这意味着同一新闻即使被多个数据源重复抓到也不会产生重复记录——配合唯一索引形成双保险。BulkWriteError会被捕获并转化为部分成功统计单条失败不影响整批数据落库。五、API 接口详解所有接口定义于 app/routers/news_data.py响应统一通过ok()包装为{success, message, data}结构。1. 新闻查询接口查询股票新闻智能获取优先数据库无数据时实时抓取GET /api/news-data/query/000001?hours_back24limit20categorycompany_announcement该接口的智能逻辑值得关注见query_stock_news的实现先按symbol hours_back从数据库查询若结果为空则自动调用 AKShare 同步服务provider.get_stock_news(symbol, limit)实时抓取并保存入库再重新查询返回。响应中的data_source字段会标明本次数据来自database还是realtime日志中也会输出 数据库无新闻数据实时获取的提示。高级查询多维条件组合POST /api/news-data/query Content-Type: application/json { symbol: 000001, symbols: [000001, 000002], start_time: 2024-01-01T00:00:00Z, end_time: 2024-12-31T23:59:59Z, category: company_announcement, sentiment: positive, importance: high, data_source: akshare, keywords: [业绩], limit: 50, skip: 0 }对应服务端NewsQueryParams支持的全部维度见 app/services/news_data_service.py其中symbols使用$in匹配多代码keywords会转换为$text文本查询排序默认按publish_time倒序。获取最新新闻GET /api/news-data/latest?symbol000001limit10hours_back24symbol为空时返回全市场最新新闻实现上等价于按publish_time倒序的限时查询。全文搜索GET /api/news-data/search?query银行symbol000001limit20基于 MongoDB 文本索引执行$text搜索并按textScore相关性排序返回见search_news实现。2. 新闻统计接口获取统计信息GET /api/news-data/statistics?symbol000001days_back7服务端通过 MongoDB 聚合管道一次完成分组统计见get_news_statistics$match过滤时间与代码范围后$group聚合出总量、正/负/中性情绪计数、高/中/低重要性计数并对category、data_source字段做分布统计。响应示例{ success: true, data: { symbol: 000001, days_back: 7, statistics: { total_count: 25, sentiment_distribution: { positive: 10, negative: 5, neutral: 10 }, importance_distribution: { high: 8, medium: 12, low: 5 }, categories: { company_announcement: 15, market_news: 8, industry_news: 2 }, sources: { 东方财富: 20, 新浪财经: 3, CCTV财经: 2 } } } }3. 新闻同步接口启动同步任务后台异步执行POST /api/news-data/sync/start Content-Type: application/json { symbol: 000001, data_sources: [akshare, tushare], hours_back: 24, max_news_per_source: 50 }该接口通过 FastAPIBackgroundTasks将任务放入后台执行立即返回任务已启动响应symbol为空时自动切换为市场新闻同步。data_sources支持tushare、akshare、realtime三选或组合不传则默认按[tushare, akshare, realtime]全量同步市场新闻同步默认仅realtime见 app/worker/news_data_sync_service.py 的sync_stock_news与sync_market_news实现。同步单只股票同步等待结果POST /api/news-data/sync/single?symbol000001hours_back24max_news_per_source50同步执行的版本响应中包含详细的同步统计信息{ success: true, data: { symbol: 000001, sync_stats: { total_processed: 60, successful_saves: 55, failed_saves: 5, duplicate_skipped: 10, sources_used: [tushare, akshare], duration_seconds: 12.3, success_rate: 91.7 } } }其中success_rate为successful_saves / total_processed × 100duplicate_skipped来自同步服务内部按标题 URL的去重_deduplicate_news。4. 管理接口清理过期新闻DELETE /api/news-data/cleanup?days_to_keep90按publish_time now - days_to_keep批量删除过期新闻返回删除条数。健康检查GET /api/news-data/health同时探测数据服务与同步服务的可用性返回service_status: healthy。六、Python SDK 使用示例无需走 HTTP在项目内部可直接以异步方式调用服务层。1. 获取新闻数据服务from app.services.news_data_service import get_news_data_service, NewsQueryParams # 获取服务实例全局单例 service await get_news_data_service()2. 查询新闻数据from datetime import datetime, timedelta # 查询最新新闻 latest_news await service.get_latest_news(symbol000001, limit10) # 高级查询 params NewsQueryParams( symbol000001, start_timedatetime.utcnow() - timedelta(days7), categorycompany_announcement, sentimentpositive, limit20 ) news_list await service.query_news(params) # 全文搜索 search_results await service.search_news(银行, symbol000001, limit10)3. 新闻数据同步from app.worker.news_data_sync_service import get_news_data_sync_service # 获取同步服务 sync_service await get_news_data_sync_service() # 同步股票新闻 stats await sync_service.sync_stock_news( symbol000001, data_sources[akshare], hours_back24, max_news_per_source50 ) print(f同步完成: {stats.successful_saves} 条成功保存)4. 直接保存标准化数据对于自定义抓取的数据可调用save_news_data直接入库支持传入单条 dict 或列表data_source与market参数必填saved_count await service.save_news_data( news_data[{title: ..., url: ..., publish_time: 2024-06-01 10:00:00}], data_sourceakshare, marketCN )服务还提供了同步版本save_news_data_sync内部使用同步 PyMongo 客户端get_mongo_db_sync适用于非异步上下文如定时任务、脚本。七、配置说明环境变量配置在.env文件中配置如下变量# .env 文件 TUSHARE_TOKENyour_tushare_token_here AKSHARE_TIMEOUT60 MONGODB_URLmongodb://localhost:27017 MONGODB_DBtradingagentsTUSHARE_TOKENTushare 接口令牌新闻接口需单独开通权限AKSHARE_TIMEOUTAKShare 请求超时秒数默认 60 秒MONGODB_URL/MONGODB_DBMongoDB 连接串与数据库名新闻数据存放于该库的stock_news集合。数据源与同步配置以下为同步服务的默认行为配置对应源码中的默认参数# 数据源优先级配置 DATA_SOURCE_PRIORITY { akshare: 1, # 优先使用AKShare tushare: 2, # 其次使用Tushare realtime: 3 # 最后使用实时聚合 } # 同步配置 SYNC_CONFIG { default_hours_back: 24, max_news_per_source: 50, batch_size: 100, retry_times: 3 }实际编码中数据源优先级通过data_sources列表的传参顺序与默认值体现个股同步默认[tushare, akshare, realtime]市场新闻同步默认[realtime]。项目还提供了NEWS_SYNC_MAX_PER_SOURCE等设置项用于控制定时同步的单源数量上限见 app/main.py 中定时任务的调用。八、性能优化策略1. 数据库优化索引优化唯一索引、单字段索引与复合索引组合覆盖按代码、时间、类别、情绪、重要性等全部查询路径配合全文索引支撑毫秒级检索批量操作入库统一走bulk_write批量 Upsert显著降低网络往返与写放大连接池通过 app/core/database.py 管理 MongoDB 连接池支撑高并发访问。2. 缓存策略系统层面配合查询缓存、数据缓存与统计缓存热点查询结果直接复用统计信息定期刷新避免重复聚合计算。3. 并发处理异步处理查询与保存均为async/await全异步实现API 层天然支持高并发批量同步同步服务支持传入多数据源并行抓取定时任务可对自选股批量调度限流控制AKShare 补丁中内置请求延迟规避数据源限流与反爬封禁Tushare 侧按积分权限自动降级。九、监控与日志日志级别约定INFO正常操作日志如同步完成: N 条保存成功新闻数据保存完成: N 条记录WARNING部分数据获取失败、数据源不可用、Tushare 权限/积分不足、日期解析失败、索引创建警告等ERROR数据库连接失败、API 调用异常、批量写入失败等。日志中大量使用 emoji 前缀 开始同步、✅ 获取成功、❌ 获取失败、 保存完成、️ 删除过期便于在日志流中快速检索关键事件。关键指标监控同步成功率来自NewsSyncStats.success_rate是数据链路健康度的核心指标查询性能服务层记录了查询耗时与返回条数可观测query_news的日志数据质量通过去重统计duplicate_skipped与保存成功率间接反映系统健康/api/news-data/health接口提供一键探测。十、故障排除常见问题排查数据库连接失败检查 MongoDB 服务状态mongod是否运行验证MONGODB_URL连接字符串与MONGODB_DB库名配置确认网络连接与 MongoDB 认证信息正确。新闻数据获取失败检查数据源 API 可用性东方财富/CCTV 接口是否可访问验证TUSHARE_TOKEN配置及新闻接口权限、积分是否充足确认网络访问权限与反爬限制AKShare 侧留意是否被限流。查询性能慢检查索引是否生效通过 MongoDBexplain()查看执行计划优化查询条件优先使用symbol publish_time等已建索引的组合考虑为热点查询增加缓存。同步数据重复确认唯一索引url title publish_time已创建验证同步服务内_deduplicate_news去重逻辑是否被正确调用对存量脏数据执行DELETE /api/news-data/cleanup或手动清理。十一、扩展开发指南添加新数据源以现有 AKShare/Tushare 提供者为模板两者均实现get_stock_news与is_available等接口见 tradingagents/dataflows/providers/china/akshare.py 与 tradingagents/dataflows/providers/china/tushare.py继承BaseProvider基类实现get_stock_news方法按_standardize_news_data的结构输出标准化字段symbol/title/content/summary/url/source/publish_time 等在 app/worker/news_data_sync_service.py 的sync_stock_news中注册新数据源分支在data_sources参数中声明新源名称即可被调度。自定义分析算法扩展情绪分析词典向_analyze_sentiment的正向/负向词表追加领域词汇或引入中文情感词库/大模型打分替代词典计数优化重要性评估规则扩充_assess_importance的高/中重要性关键词或结合消息类型如监管函、股权质押加权添加新的分类标准在_classify_news_category中增加类别分支并在文档中同步更新枚举说明实现自定义分析指标可在标准化阶段为文档追加自定义字段如热度分、舆情扩散度配合 MongoDB 聚合管道输出。十二、总结TradingAgents-CN 新闻数据系统是连接外部资讯与智能分析链路的枢纽三层架构职责清晰三路数据源互为备份规则化的智能分析分类、情绪、重要性、关键词让原始新闻在入库瞬间即完成结构化打标MongoDB 的批量 Upsert 与多索引设计保证了写入去重与毫秒级查询REST API 与 Python SDK 双入口则让上层分析器与外部调用方都能便捷消费新闻数据。配合智能降级数据库无数据自动实时抓取、容错日志与可扩展的数据源接口这套系统为股票投资分析提供了可靠、高效、可观测的新闻数据基础设施。延伸阅读本文所述架构与服务在 docs/architecture、docs/features/news/NEWS_SYNC_FEATURE.md 与 docs/features/news/NEWS_SENTIMENT_ANALYSIS.md 中有更深入的设计文档Tushare 新闻接入的专项说明可参考 docs/guides/tushare_news_integration/README.md。【免费下载链接】TradingAgents-CN基于多智能体LLM的中文金融交易框架 - TradingAgents中文增强版项目地址: https://gitcode.com/GitHub_Trending/tr/TradingAgents-CN创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考