企业官网建设流程全解析

TradingAgents-CN 实时行情入库服务优化指南：智能频率控制与三源接口轮换实战解析

【免费下载链接】TradingAgents-CN基于多智能体LLM的中文金融交易框架 - TradingAgents中文增强版项目地址: https://gitcode.com/GitHub_Trending/tr/TradingAgents-CN

实时行情入库服务（Quotes Ingestion Service）是 TradingAgents-CN 中文金融交易框架中负责把全市场近实时行情定时采集并写入 MongoDB（market_quotes集合）的核心模块。本文基于仓库中的 优化总结文档 与 配置指南，结合app/services/quotes_ingestion_service.py等源码实现，系统讲解本次优化的背景、方案、配置项、工作流程与升级路径。读完本文，你将掌握如何按 Tushare 权限等级（免费/付费）设置采集频率、如何启用 Tushare → AKShare 东方财富 → AKShare 新浪财经的三源轮换、如何用环境变量完成不同场景的落地部署，以及如何通过日志验证服务运行状态。

优化背景：原有方案面临的四大问题

优化前的实时行情入库服务存在以下四个直接影响可用性的设计缺陷，它们是本次全部改动的出发点：

1. 默认 30 秒采集频率过高：Tushare 免费用户每小时只能调用 2 次rt_k接口，而 30 秒采集意味着每小时 120 次调用，远超配额上限，直接导致免费用户服务不可用。
2. AKShare 只使用单一接口：仅调用东方财富接口stock_zh_a_spot_em，未使用新浪财经接口stock_zh_a_spot。频繁且集中地请求单一接口，容易触发反爬机制被封锁 IP。
3. BaoStock 无实时行情接口：BaoStock 本身不提供实时行情能力，但旧代码仍尝试调用，白白浪费调度资源与等待时间。
4. 无智能频率控制：免费用户与付费用户共用同一套配置，付费用户的高权限得不到利用，免费用户则极易超限。

优化方案：五处核心改动与源码实现

1. 调整默认采集频率：30 秒 → 360 秒

修改位于 app/core/config.py。默认采集间隔由 30 秒调整为 360 秒（6 分钟），并在配置描述中明确给出不同用户的推荐区间：

QUOTES_INGEST_INTERVAL_SECONDS: int = Field( default=360, description="实时行情采集间隔（秒）。默认360秒（6分钟），免费用户建议>=300秒，付费用户可设置5-60秒" )

效果：每小时采集 10 次，其中 Tushare 最多被调用 2 次（不超限），免费用户可正常使用，且 6 分钟粒度已能满足大多数行情展示与分析场景。

2. 为 AKShare 适配层添加新浪财经接口

修改位于 app/services/data_sources/akshare_adapter.py。get_realtime_quotes方法新增source参数，支持在东方财富与新浪财经之间切换：

def get_realtime_quotes(self, source: str = "eastmoney"): """ 获取全市场实时快照 Args: source: "eastmoney"（东方财富）或 "sina"（新浪财经） """ if source == "sina": df = ak.stock_zh_a_spot() # 新浪财经接口 else: df = ak.stock_zh_a_spot_em() # 东方财富接口

源码中值得注意的细节是列名兼容层：两个接口返回的 DataFrame 列名并不一致（如代码列可能是代码/code/symbol/股票代码，价格列可能是最新价/现价/price等），适配层通过候选列名依次探测的方式完成归一化，例如：

code_col = next((c for c in ["代码", "code", "symbol", "股票代码"] if c in df.columns), None) price_col = next((c for c in ["最新价", "现价", "最新价(元)", "price", "最新", "trade"] if c in df.columns), None)

效果：支持两个 AKShare 接口轮换使用，降低被封 IP 的风险，提高服务可靠性。

3. 实现三种接口轮换机制

修改位于 app/services/quotes_ingestion_service.py。服务内部维护轮换源列表与索引，_get_next_source每次调用返回下一个数据源：

轮换顺序：

1. Tusharert_k
2. AKShare 东方财富
3. AKShare 新浪财经

def _get_next_source(self) -> Tuple[str, Optional[str]]: """获取下一个数据源（轮换机制）""" if not settings.QUOTES_ROTATION_ENABLED: # 未启用轮换，使用默认优先级 return "tushare", None # 轮换逻辑：0=Tushare, 1=AKShare东方财富, 2=AKShare新浪财经 current_source = self._rotation_sources[self._rotation_index] self._rotation_index = (self._rotation_index + 1) % len(self._rotation_sources) if current_source == "tushare": return "tushare", None elif current_source == "akshare_eastmoney": return "akshare", "eastmoney" else: # akshare_sina return "akshare", "sina"

当QUOTES_ROTATION_ENABLED=false时，则回退到默认优先级（Tushare 优先）的取数策略。效果：三种接口轮流使用，避免单一接口被限流，提高整体稳定性。

4. 添加 Tushare 调用次数限制（滑动时间窗口）

对应实现为 quotes_ingestion_service.py 中的_can_call_tushare。免费用户通过一个基于 deque 的滑动窗口记录最近一小时的调用时间戳，超过_tushare_hourly_limit（默认 2）则跳过本次 Tushare 调用：

def _can_call_tushare(self) -> bool: """判断是否可以调用 Tushare rt_k 接口""" if self._tushare_has_premium: return True # 付费用户不限制 # 免费用户：检查每小时调用次数 now = datetime.now(self.tz) one_hour_ago = now - timedelta(hours=1) # 清理1小时前的记录 while self._tushare_call_times and self._tushare_call_times[0] < one_hour_ago: self._tushare_call_times.popleft() # 检查是否超过限制 if len(self._tushare_call_times) >= self._tushare_hourly_limit: logger.warning("⚠️ Tushare rt_k 接口已达到每小时调用限制，跳过本次调用") return False return True

每次 Tushare 调用成功后，通过_record_tushare_call将调用时间追加进队列：

def _record_tushare_call(self) -> None: """记录 Tushare 调用时间""" self._tushare_call_times.append(datetime.now(self.tz))

效果：免费用户每小时最多调用 2 次，超过限制自动跳过并降级到 AKShare，不影响服务正常运行。

5. 自动检测 Tushare 付费权限

对应实现为 quotes_ingestion_service.py 中的_check_tushare_permission。服务首次运行时，会通过TushareAdapter尝试调用rt_k(ts_code='000001.SZ')探测权限：

def _check_tushare_permission(self) -> bool: """检测 Tushare rt_k 接口权限""" try: adapter = TushareAdapter() df = adapter._provider.api.rt_k(ts_code='000001.SZ') if df is not None and not getattr(df, 'empty', True): logger.info("✅ 检测到 Tushare rt_k 接口权限（付费用户）") self._tushare_has_premium = True else: logger.info("⚠️ Tushare rt_k 接口无权限（免费用户）") self._tushare_has_premium = False except Exception as e: if "权限" in str(e) or "permission" in str(e): self._tushare_has_premium = False return self._tushare_has_premium

值得说明的是，源码中还处理了两种边界情况：当 Tushare 适配器整体不可用（adapter.is_available()返回 False，例如未配置 Token）时直接跳过检测；当异常信息中包含"权限"/"permission"/"没有访问"等关键词时判定为免费用户，否则按一般性失败降级处理。检测结果只会执行一次（_tushare_permission_checked标记），并在run_once中输出对应建议日志：

• 付费用户：提示可将QUOTES_INGEST_INTERVAL_SECONDS设置为 5-60 秒以充分利用权限；
• 免费用户：提示当前每小时最多 2 次调用及当前采集间隔。

新增配置项全览

本次优化在 app/core/config.py 中新增/调整了 7 个核心配置项（以下为.env写法，括号内为默认值与说明）：

# ======================================== # 实时行情入库服务配置 # ======================================== # 是否启用实时行情入库服务 QUOTES_INGEST_ENABLED=true # 采集间隔（秒） # - 免费用户建议: 300-600 秒（5-10分钟） # - 付费用户建议: 5-60 秒 # - 默认: 360 秒（6分钟） QUOTES_INGEST_INTERVAL_SECONDS=360 # 休市期/启动兜底补数 QUOTES_BACKFILL_ON_STARTUP=true QUOTES_BACKFILL_ON_OFFHOURS=true # ======================================== # 接口轮换和限流配置 # ======================================== # 启用接口轮换机制 # - true: 轮流使用 Tushare/AKShare东方财富/AKShare新浪财经 # - false: 按默认优先级使用（Tushare > AKShare） QUOTES_ROTATION_ENABLED=true # Tushare rt_k 接口每小时调用次数限制 # - 免费用户: 2 次（Tushare 官方限制） # - 付费用户: 可设置更高（如 1000） QUOTES_TUSHARE_HOURLY_LIMIT=2 # 自动检测 Tushare rt_k 接口权限 # - true: 首次运行自动检测，付费用户会收到提示 # - false: 不检测，按配置运行 QUOTES_AUTO_DETECT_TUSHARE_PERMISSION=true

对应的 Pydantic 定义（含类型与约束）位于 app/core/config.py，其中：

• QUOTES_INGEST_INTERVAL_SECONDS：int，默认360，描述中明确"免费用户建议 >=300 秒，付费用户可设置 5-60 秒"；
• QUOTES_ROTATION_ENABLED：bool，默认True；
• QUOTES_TUSHARE_HOURLY_LIMIT：int，默认2；
• QUOTES_AUTO_DETECT_TUSHARE_PERMISSION：bool，默认True。

另外，Tushare 基础配置（TUSHARE_TOKEN、TUSHARE_ENABLED、TUSHARE_TIER）与行情服务联动：TUSHARE_TOKEN留空或QUOTES_TUSHARE_HOURLY_LIMIT=0时，即可实现"纯 AKShare"模式。

工作流程：免费用户与付费用户的时间轴

免费用户（6 分钟采集一次，默认配置）

时间轴（每6分钟）： 00:00 → Tushare rt_k（第1次调用） 06:00 → AKShare 东方财富 12:00 → AKShare 新浪财经 18:00 → Tushare rt_k（第2次调用） 24:00 → AKShare 东方财富 30:00 → AKShare 新浪财经 36:00 → Tushare rt_k（第3次调用，但超过限制，跳过） 36:00 → AKShare 东方财富（自动降级） 42:00 → AKShare 新浪财经 48:00 → Tushare rt_k（第4次调用，但超过限制，跳过） 48:00 → AKShare 东方财富（自动降级） 54:00 → AKShare 新浪财经 60:00 → 新的一小时开始，Tushare 限制重置

说明：每小时 10 次采集，Tushare 最多调用 2 次（不超限），其余 8 次使用 AKShare，自动降级不影响服务。

付费用户（30 秒采集一次）

# 修改配置 QUOTES_INGEST_INTERVAL_SECONDS=30 QUOTES_TUSHARE_HOURLY_LIMIT=1000

时间轴（每30秒）： 00:00 → Tushare rt_k 00:30 → AKShare 东方财富 01:00 → AKShare 新浪财经 01:30 → Tushare rt_k 02:00 → AKShare 东方财富 02:30 → AKShare 新浪财经 ...

说明：每小时 120 次采集，Tushare 调用 40 次（不超限），充分利用付费权限的同时仍保持轮换以提高可靠性。

交易时段判断与兜底补数

源码中_is_trading_time（quotes_ingestion_service.py）负责判断是否处于可采集时段：仅工作日（周一至周五）的 9:30-11:30 与 13:00-15:30（收盘后含 30 分钟缓冲期，确保拿到收盘价）才执行采集。非交易时段则根据QUOTES_BACKFILL_ON_OFFHOURS决定是否执行兜底补数backfill_last_close_snapshot_if_needed；应用启动时（QUOTES_BACKFILL_ON_STARTUP=true）也会在 app/main.py 的启动流程中先创建索引并补数：

# 启动期：若需要在休市时补充上一交易日收盘快照 if settings.QUOTES_BACKFILL_ON_STARTUP: qi = QuotesIngestionService() await qi.ensure_indexes() await qi.backfill_last_close_snapshot_if_needed()

补数逻辑分两条路径（quotes_ingestion_service.py）：market_quotes集合为空时从stock_daily_quotes历史数据集合导入最新交易日收盘数据；集合非空但trade_date落后于最新交易日时，则通过DataSourceManager.get_realtime_quotes_with_fallback()（app/services/data_sources/manager.py）拉取近实时快照填充。

入库细节：字段与批量写入

采集到的行情通过_bulk_upsert（quotes_ingestion_service.py）以pymongo.UpdateOne批量 upsert 写入market_quotes集合，字段包括code（6 位）、symbol、close、pct_chg、amount、volume、open、high、low、pre_close、trade_date、updated_at。代码统一经过_normalize_stock_code标准化（处理sz000001/sh600036等前缀与不足 6 位的补零），并在集合上建立code唯一索引与updated_at索引（ensure_indexes）。每次执行还会把同步状态（成功与否、数据源、记录数、间隔等）写入quotes_ingestion_status集合，可通过get_sync_status查询。

性能对比：优化前后

优化前

优化后

分场景配置方案

场景 1：免费用户（推荐配置）

# 6分钟采集一次，每小时10次 QUOTES_INGEST_ENABLED=true QUOTES_INGEST_INTERVAL_SECONDS=360 QUOTES_ROTATION_ENABLED=true QUOTES_TUSHARE_HOURLY_LIMIT=2 QUOTES_AUTO_DETECT_TUSHARE_PERMISSION=true

说明：每小时采集 10 次，Tushare 最多调用 2 次（不超限），其余 8 次使用 AKShare，避免被限流或封 IP。

场景 2：Tushare 付费用户（高频采集）

# 30秒采集一次，每小时120次 QUOTES_INGEST_ENABLED=true QUOTES_INGEST_INTERVAL_SECONDS=30 QUOTES_ROTATION_ENABLED=true QUOTES_TUSHARE_HOURLY_LIMIT=1000 # 付费用户限制更高 QUOTES_AUTO_DETECT_TUSHARE_PERMISSION=true

说明：充分利用付费权限，30 秒更新一次接近实时，仍启用轮换提高可靠性。

场景 3：只使用 AKShare（无 Tushare Token）

# 5分钟采集一次，只使用 AKShare QUOTES_INGEST_ENABLED=true QUOTES_INGEST_INTERVAL_SECONDS=300 QUOTES_ROTATION_ENABLED=true QUOTES_TUSHARE_HOURLY_LIMIT=0 # 禁用 Tushare QUOTES_AUTO_DETECT_TUSHARE_PERMISSION=false # 不配置 Tushare Token TUSHARE_TOKEN=

说明：完全依赖 AKShare（免费），东方财富与新浪财经接口轮换，避免 Tushare 相关错误。

场景 4：极简配置（使用默认值）

# 只需启用服务，其他使用默认值 QUOTES_INGEST_ENABLED=true

说明：默认 6 分钟采集一次，自动检测 Tushare 权限，自动轮换接口。

运行监控与日志解读

查看任务状态

• 前端：系统配置 → 定时任务管理 → 实时行情入库服务；
• 后端日志：

# 查看实时日志 tail -f logs/app.log | grep "行情入库" # 查看轮换日志 tail -f logs/app.log | grep "使用.*接口获取实时行情"

关键日志示例

成功采集（Tushare）：

📊 使用 Tushare rt_k 接口获取实时行情 ✅ 行情入库完成 source=tushare, matched=5440, modified=5440

接口轮换（AKShare 东方财富）：

📊 使用 AKShare eastmoney 接口获取实时行情 ✅ AKShare eastmoney 获取到 5440 只股票的实时行情 ✅ 行情入库完成 source=akshare_eastmoney, matched=5440, modified=5440

Tushare 限流自动降级：

⚠️ Tushare rt_k 接口已达到每小时调用限制 (2次)，跳过本次调用，使用 AKShare 备用接口 📊 使用 AKShare sina 接口获取实时行情 ✅ 行情入库完成 source=akshare_sina, matched=5440, modified=5440

首次运行权限检测（付费用户）：

🔍 首次运行，检测 Tushare rt_k 接口权限... ✅ 检测到 Tushare rt_k 接口权限（付费用户） ✅ 检测到 Tushare 付费权限！建议将 QUOTES_INGEST_INTERVAL_SECONDS 设置为 5-60 秒以充分利用权限

首次运行权限检测（免费用户）：

🔍 首次运行，检测 Tushare rt_k 接口权限... ⚠️ Tushare rt_k 接口无权限（免费用户） ℹ️ Tushare 免费用户，每小时最多调用 2 次 rt_k 接口。当前采集间隔: 360 秒

升级步骤与验证

从旧版本升级到包含本次优化的版本，可参考以下流程（完整细节见 配置指南）：

步骤 1：更新代码到包含优化版本的分支。

步骤 2：更新.env配置（新配置项均可选，不写则使用默认值）：

QUOTES_INGEST_INTERVAL_SECONDS=360 QUOTES_ROTATION_ENABLED=true QUOTES_TUSHARE_HOURLY_LIMIT=2 QUOTES_AUTO_DETECT_TUSHARE_PERMISSION=true

步骤 3：重启后端服务：

uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

步骤 4：验证——查看后端日志确认权限检测与接口轮换正常，访问前端任务管理页面查看任务状态。

常见问题（FAQ）

Q1：为什么默认是 6 分钟，不是 30 秒？Tushare 免费用户每小时只能调用 2 次rt_k接口，30 秒采集会立即超限导致服务不可用；6 分钟是平衡实时性与配额限制的最佳选择。

Q2：我是付费用户，如何设置高频采集？修改.env文件设置QUOTES_INGEST_INTERVAL_SECONDS=30与QUOTES_TUSHARE_HOURLY_LIMIT=1000，然后重启后端服务。

Q3：如何禁用 Tushare，只使用 AKShare？不配置TUSHARE_TOKEN（留空），或设置QUOTES_TUSHARE_HOURLY_LIMIT=0。

Q4：接口轮换是什么意思？第 1 次采集用 Tushare，第 2 次用 AKShare 东方财富，第 3 次用 AKShare 新浪财经，第 4 次回到 Tushare，循环往复。

Q5：如何查看当前使用的是哪个接口？查看后端日志，搜索使用.*接口获取实时行情。

Q6：AKShare 会被封 IP 吗？6 分钟采集一次被封概率很低；启用轮换机制后东方财富与新浪财经交替使用进一步降低风险，即使被封也会自动切换到另一个接口。

Q7：如何手动触发采集？前端：系统配置 → 定时任务管理 → 实时行情入库服务 → 立即执行；API：POST /api/scheduler/jobs/quotes_ingestion_service/trigger。

影响范围与总结

本次优化涉及 app/core/config.py、app/services/data_sources/akshare_adapter.py、app/services/quotes_ingestion_service.py 三个源码文件，影响所有依赖实时行情的功能：前端股票行情展示、自选股列表与 AI 分析报告。

核心改进：

• 默认 6 分钟采集，免费用户友好；
• 三种接口轮换，避免限流与封 IP；
• 自动检测 Tushare 权限，智能调整调用策略；
• 付费用户可配置 5-60 秒高频采集，接近实时行情。

升级建议：免费用户直接使用默认配置；付费用户设置 30-60 秒高频采集；仅依赖 AKShare 的用户禁用 Tushare 即可。日常运维建议定期查看后端日志，关注接口轮换与限流日志，并根据实际行情时效需求动态调整QUOTES_INGEST_INTERVAL_SECONDS。

【免费下载链接】TradingAgents-CN基于多智能体LLM的中文金融交易框架 - TradingAgents中文增强版项目地址: https://gitcode.com/GitHub_Trending/tr/TradingAgents-CN