2. 正文内容做量化或者盯盘工具的人绕不开一个坎实时行情数据从哪里来。网上能搜到的免费源不少但真正要接到自己的系统里能稳定跑、字段全、延迟可控的还得自己动手封装一层。我之前整理过一系列股票数据API的笔记前几篇讲了基础行情、K线、财务数据这些偏“静态”的内容。这篇专门说实时交易数据——也就是分时价格、买卖五档、最新成交、涨跌停价这些每秒钟都在变的东西。无论你是做自动交易、盘中异动监控还是只想给自选股页面加一个“接近实时”的刷新逻辑这部分都绕不开。先说清楚一个概念实时交易数据和K线数据完全是两码事。K线是聚合后的结果——1分钟K线就是把那一分钟里的成交做了一次归纳而实时交易数据是原始的、逐笔或逐秒的状态快照包含当前价、涨跌幅、成交量、买卖盘口等。后者的特点是频率高、字段杂、对接口稳定性极其敏感。这篇文章会从数据源选型、字段解析、轮询策略到本地缓存更新完整走一遍我实际落地过的方案。适合有一定编程基础、准备自己搭行情模块的开发者。如果你只是想拿现成库用也可以参考中间的字段说明来理解你手上那套数据的含义。1.1 实时行情数据的核心组成先拆解一次完整的实时快照到底包含什么。以沪深A股为例一次标准的分时快照一般包含以下内容。时间戳精确到秒甚至毫秒是判断数据新鲜度的关键。很多接口会同时给“交易所时间”和“本地接收时间”两者之差就是你这条链路的真实延迟。价格信息最新价是快照里最重要的字段但光有最新价不够。涨幅、振幅、今开、昨收、最高、最低这些相对价格能帮你判断当前状态尤其是涨幅是绝大多数监控策略的第一过滤条件。成交量信息总成交量、总成交额再加上换手率可以用来估算活跃度。盘口信息买一至买五的委托价和委托量、卖一至卖五的委托价和委托量。这是判断短期买卖力量对比的关键很多盘口异动策略都依赖它。涨跌停价格计算出来之后可以识别哪些股票正在封板以及封单量有多大。这两项数据在普通的K线接口里拿不到但在一份合格的实时快照里必须要有。这些字段组合起来才形成了一张可用的“实时快照”。只给一个最新价说实话连像样的监控都做不了。1.2 数据源选型的现实考量实时数据的获取方案主流就那么几条路第三方数据商的HTTP轮询接口、WebSocket推送、还有一些中间层方案。我在实际选型时主要看四个维度。第一是字段完整性。有些免费接口只给最新价和涨跌幅盘口要付费才能拿这类源基本直接淘汰因为后续每加一个需求就要换源代价太大。第二是延迟一致性。轮询模式下单次请求拿到的是一个快照这个快照本身是某个时间点的完整状态这没问题。但如果你同时请求多只股票每只股票的刷新时间点不一样做横向比较时就要特别小心。第三是限流策略。免费或低成本的接口都有每秒或每分钟的请求上限超了要么被断开要么返回错误码。第四是数据准确性。不仅是数字对不对还包括字段是否对齐——比如除权除息日、停牌状态的标记是否及时更新。我遇到过几次股票已经停牌平台接口还在返回旧价如果监控策略没有做停牌过滤就会产生虚假信号。综合下来我的建议是能用WebSocket就用WebSocket条件受限、必须走HTTP轮询那就要做好快照时间戳的记录与缓存分层。下面主要讲轮询方案的完整落地因为这个方案适配性最强适合大多数个人开发者和中小团队。1.3 为什么轮询方案依然值得做扎实很多人觉得轮询是“低端”做法实际并非如此。推送协议确实能做到秒级甚至毫秒级通知但要维持一个长连接涉及到心跳、重连、序列号对齐、数据包完整性校验等一堆问题。个人开发者为了维护一个长连接付出的时间和精力可能比优化一次HTTP轮询大得多。轮询带有一个天然优点它是无状态的。每次请求拿到一个完整快照断网、超时、重试的成本都很低。程序重启后只要重新拉一次快照状态就恢复不需要做任何断点续传。这正是很多自用监控脚本选它的原因。当然轮询的劣势也很明显。请求频率太高会给服务器造成压力也容易被限流频率太低又拿不到“连续”的行情变化。所以轮询频率需要根据你的实际场景去平衡。做日内的分钟级监控3到5秒一次完全够用。做高频的盘口异动捕捉1秒一次是上限再快就得考虑推送方案。我在自己的系统里设置了可配置的轮询间隔平时跑3秒盘中重点盯盘的时候切到1秒。2. 字段设计与快照结构2.1 快照字段的精简与保留原则接口返回的原始数据字段可能有三四十个。但实际落到自己的存储或日志里不需要全收。收太多会撑爆磁盘收太少后续分析时发现缺字段又要重新拉。我的做法是分两套一套是完整快照用于盘中实时监控和盘后复盘一套是精简快照用于长时间运行的趋势统计。完整快照字段一般包括时间戳、最新价、涨跌幅、涨跌额、今开、昨收、最高、最低、总成交量、总成交额、换手率、量比、买一量、买一价、买二量、买二价、买三量、买三价、买四量、买四价、买五量、买五价、卖一量、卖一价、卖二量、卖二价、卖三量、卖三价、卖四量、卖四价、卖五量、卖五价、涨停价、跌停价。精简快照则只保留时间戳、最新价、涨跌幅、成交量、成交额、换手率、封单量买一量或卖一量、涨跌停状态。选择登录条件时优先保留能直接参与计算的字段而不要保留需要二次加工才能用上的字段。比如“封单量”就可以直接从买一量里算出来不需要单独存一份省空间也省逻辑。2.2 快照的时间对齐策略一个经常被忽视的问题不同股票的快照时间戳并不一致。即便你在同一秒内发起多只股票的请求返回的快照时间也可能相差几百毫秒甚至几秒。直接拿这些快照做对比会得出错误结论。我采用的做法是给每条快照打上两个时间exchange_time交易所时间戳和 local_time本地接收时间戳。做策略判断时用exchange_time做对齐做链路质量监控时用两者之差。如果持续发现某只股票的exchange_time远早于本地时间说明这路数据已经明显滞后要么换源要么在程序里降低它的优先级。举个实际例子我盘中监控代码里有一个买入信号模块它会对比同一秒内多只股票的涨幅和成交量变化。刚开始直接拿local_time排序去选目标结果出现了一些明明涨幅相近、但实际报价时间差了1秒多的“假领先股”。改成用exchange_time对齐后这个问题就消失了。2.3 封单量的计算逻辑做涨停板监控的人最关心的就是封单量。不同数据平台返回字段不一样有的平台直接给封单金额有的平台只给买一量。其实封单量的计算很简单股票涨停时买一位置的委托量就是封单量股票跌停时卖一位置的委托量就是封单量。需要注意的一个细节是“封单量”和“封单金额”不一定对得上。因为涨停价的委托量以手为单位金额则需要用当前涨停价去乘。有的平台返回的封单字段已经被券商或数据商做过处理含义可能不同。我在第一次对接时就被坑过某平台的“封单量”直接给的是金额我以为是对手数导致监控逻辑一度把封单金额当成手数来算比例结果出现离谱的封单强度判断。所以拿到一个新接口第一步永远是抽样验证字段含义而不是直接读进逻辑。验证方法很简单把涨停价乘以买一量手数乘以100股看是否和平台的封单金额字段对得上。对不上就去翻接口文档里的单位说明。3. 实战搭建从请求到本地缓存3.1 数据源接入与基础封装我以最常见的HTTP接口为例模拟一个数据源GET /api/stock/realtime传入股票代码列表返回快照数组。下面是关键请求过程的伪代码。import time import requests class RealtimeClient: def __init__(self, token, base_url): self.token token self.base_url base_url self.session requests.Session() self.session.headers.update({ Authorization: fBearer {self.token} }) def fetch_snapshot(self, codes): payload {codes: codes, fields: all} for attempt in range(3): try: resp self.session.get( f{self.base_url}/api/stock/realtime, paramspayload, timeout5 ) if resp.status_code 200: data resp.json() if data.get(code) ! 0: raise RuntimeError(f接口业务错误: {data.get(msg)}) return data[data] except Exception as exc: print(f请求失败: {exc}, 第{attempt 1}次重试) time.sleep(1) return None这段代码做了三件事复用Session减少TCP握手开销设置超时避免请求卡死失败自动重试最多三次。为什么重试间隔设为1秒实测中瞬时网络抖动往往在1秒内恢复间隔太短容易连续打到同一个故障点间隔太长则会让监控数据出现明显空窗。3.2 本地快照表的设计拿到数据后不能直接扔给业务逻辑。我建议先在本地维护一份“快照表”相当于给实时数据加一层缓存。这样业务代码读数据永远是从内存里读而不是实时发HTTP请求既稳定又快速。结构上我用一个字典就够了键是股票代码值是快照对象。为了支持多线程读需要加锁避免脏读。演示代码如下。import threading class SnapshotStore: def __init__(self): self._lock threading.RLock() self._snapshots {} def update(self, snapshots): with self._lock: for snap in snapshots: code snap[code] self._snapshots[code] snap def get(self, code): with self._lock: return self._snapshots.get(code) def get_all(self): with self._lock: return self._snapshots.copy()这种缓存模式的优点是业务逻辑与数据源解耦。后续想切换WebSocket推送只需要修改数据接收端把推送来的快照同样写进update()业务层完全不用动。3.3 轮询调度与主循环实现主循环负责定时拉取快照、解析数据、更新缓存。这里要注意两个细节轮询间隔的精确控制以及请求股票池的分组。股票池比较大的情况下一次HTTP请求带太多代码容易超时或者被限流。我一般把50到100只股票分成一组分多次请求。循环调度伪代码如下。import time class MarketPoller: def __init__(self, client, store, code_pool, interval3): self.client client self.store store self.code_pool code_pool self.interval interval def split_pool(self, size80): for i in range(0, len(self.code_pool), size): yield self.code_pool[i:i size] def run_once(self): for group in self.split_pool(): data self.client.fetch_snapshot(group) if data: snapshots self.parse_snapshots(data) self.store.update(snapshots) def loop(self): while True: start time.time() self.run_once() elapsed time.time() - start sleep_time max(0, self.interval - elapsed) time.sleep(sleep_time)这里我特别强调用间隔减去实际耗时来调整sleep时间而不是固定睡3秒。原因是网络请求耗时不定如果固定睡3秒实际周期可能是3.2秒或者3.5秒长期运行下来监控节奏会偏移。用动态补偿能让整体周期稳定在设定值附近。3.4 延迟与新鲜度统计一个实时数据模块是否合格不能靠感觉要看量化指标。我每次拉取完快照后都会把每条数据的交易所时间减去本地接收时间算出一个差值记为链路延迟。我在本地统计数据时做了三档统计延迟小于500ms的占比、500ms到1秒的占比、大于1秒的占比同时记录最大值。如果某个时段连续出现延迟大于1秒的情况说明数据源或网络链路出现瓶颈就需要考虑切换到备用源或者降低轮询频率。我一直把这个统计放在配置文件里每50次请求输出一次日志摘要。不要小看这个功能它能帮你快速定位“程序没报错但数据却明显延迟”的诡异问题。4. 盘中常见问题与排查手册4.1 股票停牌、退市等状态标记实时数据中最容易踩坑的其实是状态字段。不少平台接口会在正常行情之外返回一个交易状态标识比如正常、停牌、退市整理、封闭期等。我见过有人写的监控脚本没有判断状态直接拿停牌股票的最新价去计算涨跌幅变化然后误报了一次“价格异动”。正确做法是每次快照更新时先检查状态字段。如果股票处于停牌状态就不要更新它的价格信息或者单独标记为“无有效行情”。我的本地快照表里专门加了一个status字段停牌时会把有效价保留为上一交易日的收盘价并把is_trading标记为False。业务层看到这个标记就会跳过所有实时价格判断。4.2 集合竞价期间的异常数据A股9:15到9:25是集合竞价阶段这个窗口的行情比较特殊。9:20之前可以撤单9:20之后就不能撤单了。因此这个阶段看到的买卖盘口变化剧烈很多委托量是试探性的。我的经验是如果做的是连续竞价策略建议直接把9:25之前的实时快照排除在信号计算之外。因为集合竞价的撮合逻辑和连续竞价完全不同价格发现机制也不一样用它训练出来的信号只是噪声。如果你做的是早盘高开策略那要单独处理这个阶段——把9:25这一刻的匹配量视为开盘成交量而不是把它归到某根K线里。4.3 请求超时与限流应对免费数据源的限流策略各不相同。有的按QPS限有的按总请求次数限还有的按单个IP限。我遇到过最严苛的是同一秒内同一个key最多允许5次请求超过就直接封1分钟。应对限流首先要做好日志。每次请求响应里如果有限流标识必须记录下来并自动降低轮询频率。我的策略是连续触发限流3次轮询间隔自动从3秒提高到5秒10分钟后再降回来。这种自适应频率调整比固定间隔稳得多。其次要做好漏数据的补偿。如果某一轮轮询因为限流没能拉到某只股票的快照业务层尽量使用上一轮缓存的快照同时记录一个“数据缺失”标记。最忌讳的做法是直接抛异常中断主循环那样会造成整片监控窗口完全空白。4.4 异常值过滤与脏数据清理实时数据偶尔会出现明显的脏数据比如最新价突然从10元跳到100元然后又跳回10元。这种情况多见于数据源内部故障或网络丢包导致的字段错位。我的过滤策略很简单维护一个最近20个有效快照的均价如果最新价偏离均价超过一定比例比如30%就先判定为可疑值。可疑值不会直接丢弃而是等下一次快照来验证——如果下一次快照恢复在均价附近那这次就是脏数据如果连续两次快照都偏离那可能真的发生异动。这个“延迟确认”逻辑能同时照顾到避免误报和捕捉真实异动。实测下来这个策略对处理偶发的单点跳变非常有效并且不会增加多少代码量。4.5 交易日与时间段的判断这个问题看似基础但很多人忽略。凌晨、周末、节假日数据源返回的内容可能是上一交易日的收盘快照也可能是空数据。如果程序不判断交易时段很容易把深夜的“零成交”当成流动性枯竭或者把节假日的旧数据当成最新行情。我采用一个简单的交易日判断方式内置一张交易日表每年更新一次同时再结合实时的快照成交额判断如果连续N次轮询成交额都为0但状态字段又显示正常就标记为疑似非交易时段。这样即使交易日表更新不及时也不会误判。4.6 内存与日志的长期运行隐患实时数据模块往往要连续跑一整个交易日中途不重启。这个场景下两个隐患最值得注意。一个是内存涨不停。如果每次快照更新都在某个列表里append跑一天就会累积几万条记录。我的做法是只保留最近N次完整快照比如最近1000次超过就自动丢弃最老的另外把不常用的字段放到单独的归档文件里盘中不删盘后统一清洗。另一个是日志膨胀。如果每条快照都打日志一个交易日能写几百MB文件。我建议把日志级别分开正常轮询日志只记录摘要和统计信息比如“完成一轮快照拉取共XXX只平均延迟XXX毫秒”只有触发异动或错误时才记录详细快照内容。5. 通知与业务集成的衔接5.1 条件触发器的设计缓存更新和通知触发是两个独立的环节。我会在快照更新后单独跑一组触发器判断是否满足通知条件。触发器一般分为几类价格突破类比如最新价突破某个自选的阈值涨速异动类比如连续几秒内涨幅快速拉升封单变化类比如涨停封单量突然减少。每个触发器接收快照返回一个布尔值。这样不同监控任务之间相互独立新增策略只需要添加一个新的触发器函数无需改核心逻辑。这段设计听起来简单但对实时数据模块的可维护性帮助很大。我在早期版本中把监控条件和数据更新耦合在一起每次加需求都要动主循环改出过好几次低级错误。拆开后清爽很多。5.2 通知通道与去重通知通道我平时主要用两种即时通讯机器人适合盘中告警邮件适合盘后汇总统计。无论用哪种都要处理一个问题重复通知。举个例子价格突破10元后如果触发器每秒都判断成立那每分钟会发送几十条告警。解决方式是加一个通知去重机制同一只股票同一类事件在一定时间内只通知一次。时间窗口可以配置比如突破类事件窗口设为1分钟封单减少事件窗口设为5分钟。去重逻辑用字典就能实现键是“股票代码事件类型”值是最新通知时间。每次触发前比对时间窗口超过窗口才算新事件。5.3 与交易执行模块的联动如果你把实时数据模块接到自动交易系统那延迟和数据准确性要求会更高。我的建议是务必加一道“人工复核”开关默认状态下只输出提示不直接下单。即便你完全相信自己的策略代码盘口数据在极端行情下也可能失真。比如瞬间大单砸出很深的卖坑然后又快速恢复这时候策略看到的“极好价格”实际上并不存在。一个正确的做法是在买卖信号生成后额外请求一次独立数据源做二次确认确认价格差在合理范围内才执行。这个双源确认逻辑虽然增加了一点延迟但在真金白银的场景下非常值得。6. 复盘与后续扩展6.1 盘中日志的盘后分析实时数据不仅用于盘中盘后的价值同样巨大。每次交易日结束后我会把当天记录的完整快照做一次压缩归档。这样后续做回测、统计涨停次日表现、分析资金流向习惯时不用再去数据商拉历史分时数据自己手里就有一份第一手素材。归档格式我选择最简单的CSV加上压缩按股票代码分目录一天的数据压缩后也就几MB。关键是字段保持原始快照完整不要删任何字段因为做回测时你永远不知道哪个字段会成为关键特征。6.2 从轮询升级到推送的平滑切换当你的监控股票数量上升到几百只以后HTTP轮询的瓶颈会越来越明显拉取一次全量快照的时间可能已经超过轮询间隔。这时就该考虑升级到WebSocket推送了。好消息是因为之前已经把数据接收与业务逻辑解耦升级过程比想象中轻松。我只需要写一个新的接收器把服务端推送的JSON快照解析后调用store.update()写入缓存主流程完全不变。轮询代码保留为兜底用来在推送异常时自动补拉形成双通道冗余。这种架构既享受了推送的低延迟又不怕推送链路故障导致的行情空白。升级过程中的几个小问题比如推送消息丢序、重复推送也是在缓存层做了幂等处理后就解决了。6.3 多市场扩展最后提一个扩展方向如果你后续要接入基金、可转债、港美股等市场实时快照的字段结构会有差异。比如可转债有“转股溢价率”、基金有“单位净值估算”、美股盘前盘的交易时段不同。设计缓存结构时尽量保持一个基础快照基类再对各市场做扩展字段避免为每个市场各写一套独立逻辑。我在扩展可转债数据时深有体会由于一开始的字段结构设计得够灵活新增一个市场只花了不到半天时间核心代码变动很小。这算是早期架构设计带来的长期红利。实际操作中遇到的具体坑和心得我基本都写在对应章节里了。你如果正在搭自己的实时数据模块最关键的还是想清楚数据新鲜度要求、股票池规模、触发策略复杂度这三个变量。它们决定了你该选轮询还是推送、轮询间隔设多少、日志要怎么取舍。后面如果有新的实战心得我再继续补充分享。