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）

基于标题关键词进行规则分类，优先级从高到低：

情绪分析（_analyze_sentiment）

采用词典计数法对"标题 + 内容"全文打分：

• 正向词表：增长、上涨、利好、盈利、成功、突破、创新、优秀；
• 负向词表：下跌、亏损、风险、问题、困难、下滑、减少、警告；
• 规则：正向词命中数 > 负向词命中数 →positive；反之为negative；相等则为neutral。

情绪分析结果会写入sentiment字段，并可在查询时作为过滤条件（如sentiment=positive筛选利好新闻）。需要说明的是，这是项目内置的轻量词典方案；文档中给出的情绪词典示例（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开头 →.SH，00/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自动检查并创建索引（全部使用background=True后台创建，避免阻塞线上写入），源码中的索引清单如下：

// 唯一约束：防止重复新闻（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 个优化索引"方案还进一步覆盖了symbols+publish_time复合索引、symbol+category+publish_time、sentiment+importance+publish_time复合索引以及title/content/summary的三字段全文索引（text类型）与created_at索引，可在数据量增长后按需补齐。全文索引是search_news全文搜索能力的底层支撑。索引创建失败不会阻止服务启动（仅输出警告日志），兼顾了可用性与健壮性。

写入去重与批量写入

写入采用pymongo.ReplaceOne+bulk_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_back=24&limit=20&category=company_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?symbol=000001&limit=10&hours_back=24

symbol为空时返回全市场最新新闻，实现上等价于按publish_time倒序的限时查询。

全文搜索

GET /api/news-data/search?query=银行&symbol=000001&limit=20

基于 MongoDB 文本索引执行$text搜索，并按textScore相关性排序返回（见search_news实现）。

2. 新闻统计接口

获取统计信息

GET /api/news-data/statistics?symbol=000001&days_back=7

服务端通过 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?symbol=000001&hours_back=24&max_news_per_source=50

同步执行的版本，响应中包含详细的同步统计信息：

{ "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 × 100，duplicate_skipped来自同步服务内部按"标题 + URL"的去重（_deduplicate_news）。

4. 管理接口

清理过期新闻

DELETE /api/news-data/cleanup?days_to_keep=90

按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(symbol="000001", limit=10) # 高级查询 params = NewsQueryParams( symbol="000001", start_time=datetime.utcnow() - timedelta(days=7), category="company_announcement", sentiment="positive", limit=20 ) news_list = await service.query_news(params) # 全文搜索 search_results = await service.search_news("银行", symbol="000001", limit=10)

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( symbol="000001", data_sources=["akshare"], hours_back=24, max_news_per_source=50 ) 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_source="akshare", market="CN" )

服务还提供了同步版本save_news_data_sync（内部使用同步 PyMongo 客户端get_mongo_db_sync），适用于非异步上下文（如定时任务、脚本）。

七、配置说明

环境变量配置

在.env文件中配置如下变量：

# .env 文件 TUSHARE_TOKEN=your_tushare_token_here AKSHARE_TIMEOUT=60 MONGODB_URL=mongodb://localhost:27017 MONGODB_DB=tradingagents

• TUSHARE_TOKEN：Tushare 接口令牌，新闻接口需单独开通权限；
• AKSHARE_TIMEOUT：AKShare 请求超时秒数（默认 60 秒）；
• MONGODB_URL/MONGODB_DB：MongoDB 连接串与数据库名，新闻数据存放于该库的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接口提供一键探测。

十、故障排除

常见问题排查

1. 数据库连接失败

   • 检查 MongoDB 服务状态（mongod是否运行）；
   • 验证MONGODB_URL连接字符串与MONGODB_DB库名配置；
   • 确认网络连接与 MongoDB 认证信息正确。

2. 新闻数据获取失败

   • 检查数据源 API 可用性（东方财富/CCTV 接口是否可访问）；
   • 验证TUSHARE_TOKEN配置及新闻接口权限、积分是否充足；
   • 确认网络访问权限与反爬限制（AKShare 侧留意是否被限流）。

3. 查询性能慢

   • 检查索引是否生效（通过 MongoDBexplain()查看执行计划）；
   • 优化查询条件，优先使用symbol + publish_time等已建索引的组合；
   • 考虑为热点查询增加缓存。

4. 同步数据重复

   • 确认唯一索引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）：

1. 继承BaseProvider基类，实现get_stock_news方法；
2. 按_standardize_news_data的结构输出标准化字段（symbol/title/content/summary/url/source/publish_time 等）；
3. 在 app/worker/news_data_sync_service.py 的sync_stock_news中注册新数据源分支；
4. 在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