📚 课程目录第4章数据采集实战

4. 数据采集实战

在Tick数据工程流水线中，采集层是整条链路的“第一公里”。它直接面对交易所或行情源，承担着把原始逐笔数据稳定、完整、低延迟地搬进本地系统的任务。一旦采集器出现丢包、阻塞或断线未恢复，后续的清洗、存储与因子计算都会建立在残缺数据之上。本节聚焦用Python构建一个高吞吐、低延迟且具备断线重连能力的Tick采集器，重点讲清原理、流程与工程落地细节。

核心原理：吞吐、延迟与可靠性的三角平衡

Tick采集器的本质是一个持续运行的I/O密集型服务，它同时受三个指标约束：吞吐量（每秒能处理多少条行情）、延迟（从行情源发出到本地落盘的时间）、可靠性（断线、丢包、异常时能否自愈）。三者往往互相牵制：为了高吞吐而批量写入，可能增加单条延迟；为了低延迟而逐条处理，又可能拖垮吞吐。

关键设计原则有三条。第一，I/O与业务解耦：网络接收、解析、落盘应分属不同阶段，用队列缓冲，避免慢速磁盘阻塞网络读取。第二，异步优先：Python的asyncio配合websockets或aiohttp能在单线程内高效管理大量连接与心跳。第三，状态可恢复：采集器必须记录“已收到的最新序号/时间戳”，断线后据此判断是否需要补数据或告警。

高吞吐不等于高延迟容忍。真正的低延迟采集器，是在保证不丢数据的前提下，把“接收—解析—入队”这条关键路径压到最短，把批量落盘、压缩、上报等重活放到旁路异步完成。

方法流程：从连接到落盘的完整链路

一个可用的Tick采集器通常包含以下阶段，建议按此顺序实现并逐段压测：

连接与鉴权：建立WebSocket/TCP连接，完成登录、订阅合约、设置心跳间隔。
接收循环：持续读取原始报文，记录接收时间戳，立即投入内存队列。
解析与校验：将JSON/二进制报文解析为结构化Tick，校验字段完整性与序号连续性。
缓冲与批量：用asyncio.Queue或环形缓冲区暂存，按条数或时间窗口批量出队。
落盘与转发：写入Parquet/CSV或消息队列，同时更新“最新序号”检查点。
监控与重连：心跳超时、异常断开时触发重连，并依据检查点决定是否补采。

下面是一个精简的异步采集骨架，体现“接收—入队—批量落盘”的解耦结构：

import asyncio, json, time
import websockets

QUEUE = asyncio.Queue(maxsize=100000)
CHECKPOINT = {"last_seq": 0, "last_ts": 0}

async def receiver(uri, subscribe_msg):
    while True:
        try:
            async with websockets.connect(uri, ping_interval=20) as ws:
                await ws.send(json.dumps(subscribe_msg))
                async for raw in ws:
                    recv_ts = time.time()
                    await QUEUE.put((recv_ts, raw))
        except Exception as e:
            print(f"[reconnect] {e}, retry in 3s")
            await asyncio.sleep(3)

async def parser_and_writer(batch_size=500, flush_interval=0.2):
    batch, last_flush = [], time.time()
    while True:
        recv_ts, raw = await QUEUE.get()
        tick = json.loads(raw)
        tick["recv_ts"] = recv_ts
        batch.append(tick)
        CHECKPOINT["last_seq"] = tick.get("seq", CHECKPOINT["last_seq"])
        now = time.time()
        if len(batch) >= batch_size or now - last_flush >= flush_interval:
            await write_batch(batch)
            batch, last_flush = [], now

async def write_batch(batch):
    # 实际项目替换为 Parquet / Kafka / Redis 写入
    print(f"flush {len(batch)} ticks, last_seq={CHECKPOINT['last_seq']}")

运行方式建议用asyncio.gather同时拉起接收与写入两个协程，并用uvloop替换默认事件循环以降低调度开销：

python -m pip install websockets uvloop
# 启动
python collector.py --uri wss://feed.example.com/ws --symbols BTCUSDT,ETHUSDT

断线重连机制：状态检查点与退避策略

断线是常态而非异常。重连机制要解决三个问题：何时重连、重连后从哪继续、如何避免风暴。推荐做法是：心跳超时或读循环抛出异常即进入重连；重连成功后先比对服务端最新序号与本地CHECKPOINT，若存在缺口则触发补采或标记告警；重连间隔采用指数退避并加随机抖动，避免大量客户端同时冲击行情源。

机制	作用	推荐参数
心跳检测	及时发现半开连接	ping_interval=20s，超时2次判定断开
指数退避	抑制重连风暴	1s→2s→4s→8s，上限30s
随机抖动	打散重连时间	退避时间±20%
序号检查点	判断是否丢数据	每条更新last_seq，落盘时持久化
队列背压	防止内存溢出	maxsize限制，满时告警或降级丢弃

把CHECKPOINT定期写入磁盘或Redis，而不是只放内存。进程崩溃重启后，采集器才能知道“我上次收到哪一条”，从而决定是继续还是补采。

案例场景与误区复盘

假设某量化团队采集某交易所逐笔成交，初期用同步requests轮询REST接口，延迟高且频繁触发限频。改造为WebSocket异步采集后，吞吐从每秒数百条提升到数万条。但上线后遇到两个问题：一是落盘用同步写CSV，磁盘抖动时队列迅速积压；二是断线后直接重连，未检查序号，导致缺口未被发现。修复方式是引入批量异步落盘与序号检查点，并对缺口触发补采任务。

常见误区包括：

把解析放在接收循环里：复杂解析阻塞网络读取，应尽快入队。
无界队列：内存无限增长，最终OOM，必须设置maxsize并定义背压策略。
重连无退避：断线后疯狂重连，被服务端封禁。
忽略时间戳：只存交易所时间，不存本地接收时间，无法评估端到端延迟。
不做监控：没有队列深度、重连次数、缺口告警，故障不可见。

把上述结构落地后，采集器就具备了“高吞吐不丢、低延迟可测、断线可恢复”的基本素质。真正决定成败的往往不是代码复杂度，而是对背压、检查点和退避这些工程细节的坚持。