零成本抓取tick级实时行情：Python+WebSocket实战指南

在行情数据这件事上，“免费”和“实时”向来是冤家。日线数据满大街都是，分钟线也能找到不少免费接口，但一旦你开口要 tick 级行情——每笔成交都逐条推送的那种数据——几乎所有数据商都会先跟你谈钱。原因不难理解：tick 数据才是量化交易的真正口粮，量大、实时、服务器成本高，靠它收费是行业惯例。之前我帮一个做日内策略的朋友搭行情抓取服务，发现了一条被很多人忽略的路：用 Python 通过 WebSocket 订阅公共免费行情网关，不走任何付费渠道，就能拿到接近 tick 级别的逐笔成交实时推送。这篇文章，我把这条路径完完整整拆开讲一遍，数据源怎么选、代码怎么写、坑在哪里，适合正在做量化回测、短线工具开发或者行情研究的读者参考。

1. tick 级行情到底值在哪：为什么大家都在抢逐笔数据

1.1 快照与逐笔成交：三秒一张照片与全程录像

刚接触行情的朋友经常搞混两个概念：快照和 tick。

快照就是每隔几秒推一次的盘口状态，包含买一到买五、卖一到卖五的挂单价格和数量、最新成交价、累计成交量等字段。这就好比你看一场直播，每隔 3 秒按一下手机截屏，最后手里攒了一叠静止照片，两张照片之间到底发生了什么，你是完全不知道的。

tick 则完全不同，它是逐笔成交明细。每一笔真实成交都会触发一条推送消息，里面写清这笔成交的时间、价格、手数、主动买卖方向。如果说快照是照片，tick 就是全程录像，每一帧都不丢。

这个差别在实际分析中非常致命。举个例子，你想统计某只股票在某个时间段内的大单净流入。用快照数据，你只能通过相邻两个快照的成交量差值去粗略推算，精度很低；用 tick 数据，你可以直接把单笔成交量超过阈值、方向为主动买的记录筛出来求和，结果干净得多。

再比如日内回测。分钟级回测通常假设一根 K 线内部价格是线性变化的，信号出现后到价就能成交。但真实盘口是一跳一跳走的，价格可能根本没有碰到你的触发价，或者碰到了又瞬间弹回，你的委托就是成交不了。tick 级数据能暴露这类微观结构问题，回测结果才更接近实盘。

顺便说清楚一个现实：常规意义上的 Level-2 全档行情，也就是逐笔委托那边，几乎不会有免费源。大家能免费拿到的主要是逐笔成交明细（部分源可能带延迟），以及基础盘口快照。本文所说的 tick，指的是这种“成交明细级”的推送。你拿不到“挂单到撤单”的完整动作，但每一笔成交是真实的。对这个精度有个正确预期，后面所有设计才看得明白。

1.2 谁在为 tick 买单：三类真实场景

第一类是量化研究者。做日内策略、高频类策略、微观结构研究的人，必须依赖逐笔数据。没有 tick，很多信号根本算不出来，尤其是那些基于盘口微观结构衍生出来的指标，比如订单流失衡、逐笔主动买卖强度，靠快照只能算个近似值。

第二类是短线交易工具的开发者。我在做盯盘告警、盘口异动检测、主力资金流跟踪这类工具时，tick 是核心输入。一次大单扫货发生后，如果等快照更新再告警，往往已经过了好几秒。很多短线机会的时间窗口本来就不宽，晚一秒感知，差别很大。

第三类是风控系统的实现者。监控持仓个股的瞬时暴跌、瞬间放量时，快照的秒级延迟在极端行情里可能造成实质性损失。风控场景对行情的实时性、完整性要求最高，也因此愿意为数据源付出更高成本。

所以 tick 数据不是“多一种粒度”这么简单，它是很多真实场景的刚需。正因为刚需，数据商才敢收高价。这也是为什么“零成本拿到 tick 级实时行情”听起来像不可能——但免费路径确实存在，前提是你知道去哪找、怎么接。

2. 零成本从哪儿来：免费数据源的选型与取舍

2.1 三条免费路径，我帮你排过雷

先说结论：免费拿 tick，其实有三条路，各有适用场景。

第一条路最简单，用 requests 之类的库轮询公开 REST 接口，拿到的是滞后几秒到几十秒的快照，严格来说不算 tick。它适合离线回测，不适合时效性要求高的场景。

第二条路是本文要展开的：行情服务商开放的 WebSocket 接口。客户端和服务器建立一条长连接，服务端主动推送已订阅标的的行情消息。这类接口在不少公共服务平台上有，通常有限频、限制订阅数量，但胜在免费、实时性好，是“零成本抓 tick”的最优解。

第三条路是社区自建网关，比如有人把某个行情协议转成 WebSocket 转发出来，或者做一个模拟推送服务供开发测试。可控性强，但稳定性要自己保障，接入前务必确认数据来源合规，别给自己埋雷。

2.2 WebSocket 行情消息的通用骨架

无论接哪家服务，行情 WebSocket 的消息结构都大同小异。

连接层面：客户端发起 WSS 请求，随后一般先发一条鉴权消息，携带 token 或 app_id 加 app_secret；鉴权通过后再发订阅消息，指定要接收的标的代码和订阅主题（tick、快照等）。之后服务端持续推送业务消息。

业务层面：行情推送消息一般包含消息类型字段（tick/snapshot/heartbeat）、标的代码、行情本体。常见格式是 JSON，少数是 protobuf 或压缩二进制。第一次接入时，我建议优先选择 JSON 协议的服务，因为它最直观，抓包后排错也容易。

我见过有人拿到一个 protobuf 协议却没有描述文件，对着十六进制报文手工猜字段，折腾了两天最后放弃了。如果你决定接二进制协议，务必先在联调阶段确认拿到完整协议文档和示例报文，否则后面每一行解析代码都可能是猜出来的。

2.3 为什么不建议轮询

有人会问：既然是免费接口，我每秒用 requests 拉一次不就行了吗？答案是否定的，原因有两个。

第一，轮询拿到的依然是快照。行情源通常以固定间隔更新缓存字段，你再怎么高频地去请求，拿到的都是同一份数据。requests 用得多并不会把快照变成 tick，只会让本地延迟高、代码丑。

第二，高频轮询极易触发限流甚至封禁。曾见过有人用 0.5 秒一次的频率刷免费接口，半小时后 IP 直接被服务商封了。相比之下，WebSocket 是服务端主动推送，一个连接上多只股票的消息复用同一通道，资源消耗小，也不容易被判为异常行为。

所以零成本抓 tick 的正确姿势，就是拥抱长连接、拥抱推送。免费 WebSocket 行情源虽然没有付费源那么丝滑，但只要工程架构合理，完全能支撑起个人级别的量化研究和工具开发。

3. 200 行代码搭出抓取服务：连接、订阅、解析、落盘

3.1 四层架构与依赖

先把整体设计讲清楚，这个抓取服务分四层：

1. 连接层：负责 WebSocket 连接、鉴权、心跳、断线重连。
2. 订阅层：负责维护自选股列表，重连之后自动重新订阅。
3. 解码层：把服务端推送的原始报文转成统一结构的 tick。
4. 存储层：处理后的数据进入内存队列并定期落盘，供下游使用。

依赖方面，起步阶段只需要两个：websockets负责异步长连接，标准库json、logging、asyncio负责序列化和调度。如果后面订阅的股票量上来了，再考虑 uvloop、数据库这类增强，但不是现在需要操心的。

3.2 连接与订阅：把通道先搭起来

直接看代码，这是整个服务的入口部分。URL 和 token 我用了占位符，你换成实际服务商提供的就能跑。

import asyncio import json import logging import time import websockets logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", ) logger = logging.getLogger("tick_catcher") GATEWAY_URL = "wss://free-quote.example.com/ws" TOKEN = "your-token-here" WATCHLIST = ["600000", "000001", "300750"] async def send_json(ws, payload): await ws.send(json.dumps(payload)) async def subscribe(ws): await send_json(ws, { "action": "auth", "token": TOKEN, "timestamp": int(time.time()), }) await send_json(ws, { "action": "subscribe", "topic": "tick", "codes": WATCHLIST, "mode": "realtime", }) logger.info("subscribed: %s", WATCHLIST) async def reader(ws): async for raw in ws: message = json.loads(raw) tick = parse_tick(message) if tick: process_tick(tick)

这里强调几个细节：

鉴权和订阅是两条独立消息，不要合成一条。部分网关对消息顺序敏感，先 auth 后 subscribe，顺序反了会被直接拒绝。此外，订阅消息里的 mode 字段是可选的，有些源默认只推快照，必须显式声明要实时成交明细。

连接建立后，最好先发一条 ping 探活，确认服务端真正能响应再批量订阅。有些网关鉴权成功后并不立即确认，只是在下一条消息里带一个 auth_ok 标志，阅读服务端文档时要特别留意这个交互细节。

3.3 解析与去重：seq 才是金标准

服务端推送的原始报文通常很长，字段名也各有一套。我们要把它裁剪成内部统一的结构：

def parse_tick(message): if not isinstance(message, dict): return None if message.get("type") != "tick": return None try: return { "seq": int(message.get("seq", 0)), "ts": message.get("timestamp", ""), "code": message.get("code", ""), "price": float(message.get("last_price", 0.0)), "volume": int(message.get("volume", 0)), "side": message.get("side", "U"), # B 主动买 S 主动卖 } except (TypeError, KeyError, ValueError): logger.warning("parse failed: %s", message) return None

我特别把 seq 字段保留下来，它是行情服务端下发的全局递增序列号，也是数据去重的金标准。

为什么需要去重？因为断线重连之后，服务端为了补偿客户端，往往会补发断线期间的 tick。如果你不去重，存储层会混入大量重复记录，后续统计结果直接报废。去重用 set 维护一个最近窗口就可以：

SEQ_WINDOW = set() def is_dup(seq): if seq is None: return False if seq in SEQ_WINDOW: return True SEQ_WINDOW.add(seq) if len(SEQ_WINDOW) > 100_000: SEQ_WINDOW.clear() return False

这个 set 只保留最近十万条序号，窗口足够覆盖一次重连的补发量，又不会让内存膨胀。如果服务端没有提供 seq，退而求其次可以用“时间戳 + 代码 + 价格 + 手数”拼一个签名去重，但效果远不如 seq 可靠——同一秒内同价同量的两笔独立成交确实存在。

顺便说一个坑：把 Python 的 dict 转成 JSON 时，数字字段默认可能是小数或字符串，行情字段尤其是 volume 建议全部显式转成 int，避免后续聚合时类型报错。

3.4 心跳与重连：长连接的保命机制

长连接最怕一件事：连接悄悄断掉。TCP 层断开还能通过异常发现，最怕是网络中间设备把连接“假保留”，客户端收不到数据却没有任何报错。

websockets库内置了ping_interval和ping_timeout参数，我用的是 20 秒和 10 秒的组合：每 20 秒发一次 ping，10 秒内没收到 pong 就判定连接异常，触发异常退出。

真正的主循环长这样：

async def supervisor(): retry = 0 while True: try: async with websockets.connect( GATEWAY_URL, ping_interval=20, ping_timeout=10, max_size=2 ** 20, ) as ws: retry = 0 await subscribe(ws) while True: try: raw = await asyncio.wait_for(ws.recv(), timeout=90) tick = parse_tick(json.loads(raw)) if tick: process_tick(tick) except asyncio.TimeoutError: logger.warning("no data for 90s, reconnecting") break except websockets.ConnectionClosed as exc: logger.warning("connection closed: %s", exc) except Exception as exc: logger.error("unexpected error: %s", exc) retry = min(retry + 1, 6) wait = 2 ** retry logger.info("reconnect in %ss", wait) await asyncio.sleep(wait)

重连等待采用指数退避：2 秒、4 秒、8 秒、16 秒、32 秒、64 秒，封顶 64 秒。这是为了避免服务端抖动时，所有客户端同时重连造成“羊群效应”。

第二个细节是asyncio.wait_for(ws.recv(), timeout=90)。这里解决的是“假死”场景：如果 90 秒内连一条业务消息、一条 ping/pong 都没有，就主动断开并重连。这个超时值可以按服务端的惯性调整，但不要设得太长，否则行情真正断流时，你要多等很久才能发现。

第三个细节是max_size=2**20，把单条消息上限设置成 1 MB，防止某些异常报文把内存打爆。如果服务商经常推送大体积的批量消息，这个值要根据实际情况调大。

到这里，核心骨架其实已经能跑了。接下来这部分，才是我最想分享的实战内容。

4. 实测中绕不开的坑：断线、限流和交易日历

4.1 “连接正常但行情冻结”的完整排查

我第一次跑这个服务时，遇到一个特别诡异的场景：连接明明没断，日志无任何报错，但行情数据从某个时间点开始就再也不更新了。

排查链路一步步走：

先看 TCP 状态，连接 ESTABLISHED，网络层正常。再看客户端消息计数，结果是某个时刻之后，真的一条消息都没收到。第三查心跳，ping 和 pong 都正常，说明链路是活的，但服务端就是不再推业务数据。

最后调出服务端的订阅状态查询接口，才发现问题：订阅的 token 因为消息量为零被服务端判定为“僵尸订阅”，悄悄清理掉了。原来这个免费网关对订阅连接有活跃度判定，某只标的长期没有成交时，订阅会被服务端自动关闭，但连接保持，造成“假死”。

解决方案是在 reader 里维护一个最后业务消息时间，用前面写的asyncio.wait_for超时机制兜底：90 秒没收到任何东西就主动断开重连，重连后自然完成重新订阅，行情恢复。这个“寂寞检测”机制在免费行情源上几乎是必备的，因为免费源的管理策略通常比付费源激进得多。

4.2 非交易时段的空转陷阱

tick 服务在非交易时段通常不推成交，但连接保持。如果不加处理，客户端会把集合竞价消息、盘前测试数据全部当成正常 tick 存进库。

我早期没做交易日过滤，结果回测数据里混入了一批 9 点 25 分之前的集合竞价撮合数据，把开盘策略的参数全部污染了。后来改成在 process_tick 里先做交易时段判断：

from datetime import datetime def in_trading_session(dt: datetime) -> bool: if dt.weekday() >= 5: return False start_m = datetime.strptime("09:30", "%H:%M").time() end_m = datetime.strptime("11:30", "%H:%M").time() start_a = datetime.strptime("13:00", "%H:%M").time() end_a = datetime.strptime("15:00", "%H:%M").time() t = dt.time() if start_m <= t <= end_m: return True if start_a <= t <= end_a: return True return False

注意，周末可以通过 weekday 判断，但法定节假日不能只看 weekday，必须维护一份节假日表或者对接交易日历接口。A 股全年休市日不算少，少过滤一天，脏数据就多一天。

还需要考虑一点：如果我不在交易时段却硬连网关上，wait_for会每 90 秒触发一次超时重连，白白增加服务端压力。我现在的主循环会在入口先判断本地时间，不在交易时段就计算下一个开盘时间，直接睡到那时候再连。这样既不会空转，也不会打扰行情服务端。

4.3 订阅限流与标的分片规划

免费行情网关的限流通常有两层：单连接订阅数量限制和消息速率限制。

我测过某公共网关，单条连接最多订阅 30 只股票，超出之后服务端直接返回错误码。所以第一步是把自选股池控制在限额之内，别一次性塞 100 个标的进去。

更隐蔽的是第二层限制：虽然只订阅 30 只，但其中如果包含几只超级活跃股，每秒推送的消息量会非常大，免费网关会对单连接消息速率熔断。我踩过之后总结出一个经验：不要把 30 个标的全选成活跃股，尽量让活跃股和冷门股混搭在一条连接里，用多条连接把流量压力打散。

标的分片的做法也简单：按股票代码做一个哈希，模 N 之后分到 N 条连接，每条连接各自维护订阅列表。这样既规避了单连接限制，也能让整体消息吞吐更均匀。

5. 从 demo 到长期扛得住：并发、存储与扩容

5.1 三层解耦，别让 IO 拖垮事件循环

早期版本我直接在 reader 里同步写文件，行情一密集，IO 一慢，整个事件循环被拖住，消息开始积压。后来改成三阶段解耦：抓取协程只负责收消息，放入 asyncio.Queue；处理协程负责清洗去重；落盘协程按批写入。

queue = asyncio.Queue(maxsize=10000) async def producer(ws): async for raw in ws: message = json.loads(raw) if message.get("type") != "tick": continue await queue.put(message) async def consumer(): batch = [] while True: item = await queue.get() tick = parse_tick(item) if not tick or is_dup(tick["seq"]): continue batch.append(tick) if len(batch) >= 500: flush_to_disk(batch) batch.clear() def flush_to_disk(rows): with open("ticks.csv", "a", encoding="utf-8") as f: for r in rows: f.write( f"{r['seq']},{r['ts']},{r['code']}," f"{r['price']},{r['volume']},{r['side']}\n" )

asyncio.Queue(maxsize=10000)的核心价值是背压。如果下游处理不过来，producer 的 put 会被阻塞，消息不会无限堆积在内存里。对行情这类高频数据来说，宁可慢一点，也不能让内存失控。

5.2 存储选型：CSV、SQLite 还是 Parquet

存储方案完全取决于你下游的用法：

• 临时试验、快速验证：CSV 足够，一行一条 tick。要注意按日期分文件，别所有数据写进一个无限膨胀的大文件。
• 中小规模回测：SQLite 很好用。单文件、支持 SQL 查询，给 code 和 ts 建上索引后，查询效率完全够用。
• 大规模分析：Parquet 列式存储更适合，但引入了额外依赖，前期不必折腾。

我日常最常用的组合是：白天抓数据时写按日期分片的 CSV，晚上再统一导入分析数据库。这样抓取进程保持轻量，分析任务集中在夜间跑，互不干扰。

5.3 多条连接和多个进程的水平扩展

当单连接订阅额度不够时，最直接的做法是分片订阅，把股票池按哈希分成 N 组，每组连接连到同一个网关。这样做的好处是单连接的订阅数量限制被规避，消息吞吐也被打散到多个连接上。

如果标的数量非常大，单进程也扛不住，就上进程级水平扩展：一个调度进程负责分片，N 个 worker 进程各自连接各自的网关，数据按股票代码分目录落盘。Python 的 asyncio 在这个场景下省掉了大量线程同步的麻烦，每个 worker 启动同一个脚本、传入不同分片编号即可。

到这一步，这套“零成本 Python 抓取 tick 级实时行情”的方案就完整铺开了。对我个人来说，这类系统的核心价值在于：把原本被高昂行情订阅费挡在门外的 tick 数据需求，降级成“一个晚上的工作量加一台普通云主机”。只要你愿意折腾、能沉下心排掉前面那些坑，它比想象中稳得多。

最后再分享一个亲测有效的经验：全部跑通之后，先拿一两只活跃股票对照行情软件核对 10 分钟左右的数据，确认 seq 连续、无重复、成交方向与盘口一致，再放量订阅。数据质量这一关过了，后面所有策略和统计才有意义；质量不过关，再花哨的框架都是白搭。