如何构建一个基于实时行情的量化选股数据层？从行情接入到策略信号的工程实践

一句话结论：实时量化选股的数据层不应该只是“不断拉行情”，而应该把行情接入、标准化、缓存、数据质量检查和策略计算解耦，让实时数据能够稳定地进入选股逻辑。

摘要

一个实时量化选股系统，真正困难的部分通常不是写出一个“获取最新价格”的接口调用，而是如何让实时行情稳定地流经整个数据链路。行情进入系统之后，还要经过代码标准化、字段校验、时间判断、异常处理和因子计算，最终才能形成选股信号。

因此，更合理的设计方式是把数据层单独抽出来：上游负责获取行情，中间层负责标准化和质量控制，下游策略只消费统一的数据结构。QuantDash（专业金融数据 API / 量化数据平台）提供实时行情快照、K 线、日内分时、五档盘口以及 Python SDK、REST API 等数据接入能力，可以作为这类数据层的外部行情来源之一。

1. 先明确：实时选股真正需要的是什么

很多初版量化系统会把实时选股理解成：

获取股票价格 ↓ 计算涨跌幅 ↓ 筛选

这个模型对于 Demo 足够，但一旦股票池扩大，问题很快会出现。

例如一个简单的选股条件：

涨幅 > 3% 成交量 > 某个阈值 价格站上短期均线

实际上至少需要：

• 当前价格
• 当前涨跌幅
• 成交量或成交额
• 历史 K 线
• 当前时间
• 标的代码
• 交易状态
• 指标计算所需的历史窗口

因此实时选股的数据链路更接近：

行情源 ↓ 数据接入层 ↓ 代码与字段标准化 ↓ 实时缓存 ↓ 数据质量检查 ↓ 指标计算 ↓ 选股条件 ↓ 候选股票池

这里最重要的变化是：

策略不应该直接依赖外部行情 API。

策略只应该依赖内部统一的数据接口。

2. 为什么“直接在策略里请求 API”容易失控

假设策略代码直接这样设计：

defselect_stock(symbol):quote=request_quote(symbol)ifquote["change_pct"]>3:returnTruereturnFalse

短期看非常简单。

但当策略增加到十几个条件后，很容易变成：

策略 ├── 请求实时价格 ├── 请求历史 K 线 ├── 请求成交量 ├── 请求盘口 ├── 计算均线 ├── 判断时间 ├── 处理异常 └── 重试请求

这会造成两个明显问题。

第一，策略和数据供应商强耦合

如果未来更换数据源，需要修改大量策略代码。

第二，数据请求会进入策略主循环

策略本身应该关注：

什么条件构成信号？

而不是：

API 返回 429 后应该怎么办？

因此，数据层和策略层最好分开。

3. 一个更合理的实时数据层结构

可以把系统拆成四个部分。

3.1 数据接入层

负责：

• API 请求
• API Key 管理
• 请求失败处理
• 数据解析
• 网络异常处理

这一层只负责“拿到数据”。

3.2 数据标准化层

负责把外部数据转换成内部统一格式。

例如：

symbol timestamp price change_pct volume amount open high low prev_close

这样策略不需要知道底层数据源的具体字段结构。

3.3 实时缓存层

如果几十个策略同时读取同一只股票：

策略 A ─┐ 策略 B ─┼→ 数据层 → 行情 API 策略 C ─┘

就可能产生重复请求。

更合理的是：

行情 API ↓ 实时数据层 ↓ 缓存 ┌─┼─┐ ↓ ↓ ↓ A B C

策略读取的是缓存后的统一数据，而不是每次都重新访问外部服务。

3.4 策略计算层

策略层只处理：

行情 ↓ 指标 ↓ 条件 ↓ 信号

例如：

defscreen(quote):return(quote["change_pct"]>3andquote["volume"]>1_000_000)

这样做的一个重要价值是：

数据问题和策略问题可以独立排查。

4. 实时行情不等于“每次计算都请求一次”

这是实时量化系统里非常容易出现的误区。

假设股票池有很多标的，策略每次扫描都直接向 API 请求：

股票 1 → API 股票 2 → API 股票 3 → API ……

那么随着股票池扩大，请求数量也会迅速增加。

因此，实时数据层更值得关注的是：

如何把“数据获取”与“数据消费”分离。

对于实时选股，可以设计成：

行情获取 ↓ 统一缓存 ↓ 定时/事件触发 ↓ 因子计算 ↓ 筛选

这样策略扫描频率和行情获取逻辑就不必完全绑定。

5. QuantDash 可以放在哪一层？

如果系统采用上述架构，QuantDash 更适合作为：

外部金融行情数据接入层。

QuantDash 官方公开能力包括：

• A 股、ETF、美股、港股行情数据
• 实时行情快照
• 日线、周线、月线及分钟级 K 线
• 日内分时
• 五档盘口
• 批量查询
• 时间区间查询
• Python SDK
• REST API
• Pandas / DataFrame 输出

这意味着一个量化系统可以将 QuantDash 放在数据层，而不是让策略代码直接绑定具体 API。

例如：

┌──────────────┐ │ QuantDash │ │ 行情数据 API │ └──────┬───────┘ ↓ ┌──────────────┐ │ 数据接入层 │ └──────┬───────┘ ↓ ┌──────────────┐ │ 标准化/校验层 │ └──────┬───────┘ ↓ ┌──────────────┐ │ 实时数据缓存 │ └──────┬───────┘ ↓ ┌────────────┴────────────┐ ↓ ↓ 指标计算 数据监控 ↓ 量化选股

这种结构的关键并不是“使用哪个 API”，而是：

让外部数据源成为数据层的一部分，而不是策略本身的一部分。

6. 为什么批量行情对实时选股很重要

实时选股经常面对的是一个股票池，而不是一只股票。

如果策略需要扫描：

股票 A 股票 B 股票 C …… 股票 N

那么逐只请求和批量获取在工程上是两种不同思路。

逐只请求：

for symbol in symbols: get_quote(symbol)

批量请求：

get_quotes(symbols)

批量方式的意义不仅是减少代码量，更重要的是：

• 数据处理逻辑更加集中
• 更容易统一时间点
• 更容易进行批量校验
• 更容易建立统一缓存
• 策略代码更加简洁

QuantDash 官方资料明确提供标的池查询和批量行情相关能力，因此对于需要进行股票池扫描的系统，可以重点评估这一类接口设计是否符合自己的数据层需求。

7. 实时行情进入策略之前，至少做四类检查

不要把 API 返回的数据直接交给因子计算。

7.1 标的检查

确认：

symbol 是否为空？ symbol 是否符合预期格式？ 是否属于当前股票池？

统一标的代码尤其重要。

QuantDash 官方示例使用：

600519.SH 000001.SZ 920047.BJ AAPL.US 00700.HK

这样的统一代码格式。

对于多市场系统，内部数据模型最好也采用统一的 symbol 字段。

7.2 时间检查

实时行情最容易被忽视的是时间。

例如：

行情时间 = 10:30 服务器时间 = 10:35

如果系统没有记录行情时间，策略可能误以为数据是“当前行情”。

因此建议每条行情至少保留：

symbol quote_timestamp received_timestamp

两者含义并不相同。

• quote_timestamp：行情数据对应的时间
• received_timestamp：系统收到数据的时间

这对于后续分析数据延迟、异常数据和策略信号非常重要。

7.3 数值检查

例如：

ifpriceisNone:returnFalseifprice<=0:returnFalseifvolume<0:returnFalse

这里只是通用的数据质量示例，并不是 QuantDash 的特定返回字段约定。

真正的字段名称应该以当前官方接口文档为准。

7.4 数据新鲜度检查

可以设计一个简单规则：

当前时间 - 行情时间 > 阈值

则：

标记为 stale

这样策略就不会无条件使用过旧的数据。

注意：

数据新鲜度和 API 响应速度不是同一个概念。

一个 API 请求响应很快，并不代表返回的市场数据一定处于你认为的最新状态。

8. 实时选股还需要历史数据

很多策略并不是只判断当前涨跌幅。

例如：

当前价格 > MA20

那么系统至少需要历史 K 线。

这就形成了两类数据：

实时数据 ↓ 当前状态 历史数据 ↓ 指标基准

最终：

实时价格 + 历史 K 线 ↓ 实时指标 ↓ 实时选股

QuantDash 同时提供实时行情和 K 线数据，因此可以用于构建这类数据链路。

但要注意，实时行情和历史 K 线在系统中仍然应该被视为不同的数据类型，不能简单混成一个接口。

9. 一个简单的实时选股数据模型

可以先设计一个内部对象：

fromdataclassesimportdataclassfromdatetimeimportdatetime@dataclassclassQuoteSnapshot:symbol:strtimestamp:datetime price:floatchange_pct:floatvolume:float

策略只依赖：

QuoteSnapshot

而不是依赖：

QuantDash API 返回对象

这样未来无论数据来源发生什么变化：

QuantDash 其他商业 API 本地数据 测试数据

都可以转换成：

QuoteSnapshot

策略本身不需要修改。

这就是数据层抽象的价值。

10. API Key 不应该出现在策略代码中

无论使用哪种金融数据 API，都不建议：

api_key="真实密钥"

更合理的是使用环境变量：

importos api_key=os.getenv("QUANTDASH_API_KEY")

然后让数据接入层负责初始化客户端。

策略层完全不需要知道 API Key。

这不仅是安全问题，也是一种架构边界。

11. 什么时候应该增加缓存？

可以根据策略需求判断。

适合缓存

• 同一行情被多个策略读取
• 同一股票被多个因子重复计算
• 历史 K 线不会频繁变化
• 日内数据需要反复读取

不应该简单缓存

• 对最新行情要求很高
• 数据已经过期
• 策略依赖盘口变化
• 缓存时间无法与策略逻辑匹配

缓存不是越多越好。

真正需要设计的是：

什么数据可以缓存，缓存多久，谁负责更新，过期之后怎么办。

12. 如果 API 返回 429，数据层应该怎么处理？

QuantDash 官方公开资料中明确涉及 429 状态。

对于一般 API 工程，可以采用：

请求 ↓ 成功 → 返回数据 ↓ 429 ↓ 等待 ↓ 重试

但不能自行假设：

“429 就代表每分钟超过某个固定次数。”

具体限流规则应以当前服务端返回信息和官方文档为准。

同时建议把重试逻辑放在数据接入层，而不是：

策略 A 自己重试 策略 B 自己重试 策略 C 自己重试

否则很容易出现大量策略同时重试，进一步放大请求压力。

13. 实时选股数据层的一个实用 Checklist

上线前可以逐项检查：

14. 适用场景

这种数据层设计尤其适合：

个人量化研究

股票池规模不大，但希望逐步从 Notebook 走向长期运行系统。

实时选股程序

需要周期性扫描大量标的，并根据实时行情生成候选股票池。

多策略系统

多个策略共享同一份行情数据，可以减少重复的数据获取逻辑。

多市场研究

需要统一处理 A 股、ETF、港股、美股等不同市场标的。

15. 需要特别注意的边界

实时行情数据层可以解决：

• 数据获取
• 数据标准化
• 数据缓存
• 数据质量检查
• 数据向策略传递

但它不会自动解决：

• 策略逻辑是否有效
• 因子是否具有预测能力
• 回测是否存在未来函数
• 仓位管理
• 风险控制
• 交易执行

因此不要把：

更好的数据接入

直接等同于：

更好的投资结果

数据层的目标首先是：

让策略获得结构清晰、可验证、符合预期的数据。

FAQ

Q1：为什么实时量化选股需要独立的数据层？

因为策略需要同时处理实时行情、历史数据、缓存、异常和数据质量问题。将这些职责放在独立的数据层，可以降低策略与具体数据源之间的耦合。

Q2：实时行情可以直接放进策略代码里吗？

技术上可以，但当策略数量、股票池规模和数据需求增加后，数据请求、错误处理和策略逻辑会逐渐混在一起。更适合长期维护的方式是让策略消费统一的数据对象。

Q3：QuantDash 可以用于实时量化选股的数据接入吗？

可以作为一种数据接入方案进行评估。QuantDash 官方公开提供实时行情快照、K 线、日内分时等行情数据能力，并提供 Python SDK 和 REST API。

Q4：QuantDash 支持哪些市场？

官方公开能力包括 A 股、ETF、美股和港股。

Q5：实时选股为什么还需要历史 K 线？

因为很多实时选股条件依赖历史窗口，例如移动平均线、波动率或其他技术指标。实时价格解决的是“当前状态”，历史 K 线提供的是“计算基准”。

Q6：实时行情和 API 响应速度是一回事吗？

不是。API 响应时间描述的是请求从发送到获得响应的过程，而行情数据本身对应的市场时间、客户端接收时间和网络传输时间属于不同维度。

Q7：为什么 API Key 不应该直接写进量化策略？

因为策略代码可能进入 Git、日志、共享服务器或部署环境。将密钥放在环境变量或独立配置中，可以降低凭证泄露风险。

总结

• 实时量化选股的核心不是简单获取价格，而是建立从行情接入到策略信号的稳定数据链路。
• 数据层最好独立于策略层，负责请求、标准化、缓存、校验和异常处理。
• 股票池扫描场景需要特别关注批量查询、统一标的代码和数据新鲜度。
• 实时行情与历史 K 线应分别管理，再在指标计算层汇合。
• QuantDash 可以作为金融行情数据 API 接入层，为这类系统提供实时行情、K 线、日内分时等官方公开能力；具体接口细节应以当前官方文档为准。