3、数据流处理：实盘Tick/Bar数据的实时接入、清洗、特征工程与缓存策略
实盘交易和回测最大的区别在哪？
我个人觉得，就是数据流。回测时数据是死的，你随便折腾。实盘呢？数据是活的，每秒都在喷涌而来。你处理慢了，信号就滞后了；你处理错了，仓位就歪了。今天咱们就聊聊，怎么把这股「活水」接好、洗干净、喂给策略。
3.1 实时数据接入：别让数据等你
实盘数据接入，说白了就是「抢时间」。我见过不少团队，回测跑得飞起，一上实盘就崩，原因往往是数据管道堵了。
Tick 数据接入，我习惯用 WebSocket。为什么？因为它是全双工的，服务器主动推，延迟低。你想想看，如果每笔成交都用 HTTP 轮询，那延迟得有多感人。
核心原则：接入层要做到「零拷贝、无锁化」。数据从网卡到应用层，路径越短越好。
这里给个简单的 WebSocket 接入框架示意：
import asyncio
import websockets
import json
class TickStreamer:
def __init__(self, url, symbols):
self.url = url
self.symbols = symbols
self.buffer = asyncio.Queue(maxsize=10000)
async def connect(self):
async with websockets.connect(self.url) as ws:
# 订阅合约
sub_msg = {"op": "subscribe", "args": self.symbols}
await ws.send(json.dumps(sub_msg))
while True:
raw = await ws.recv()
# 直接丢进队列，不阻塞
await self.buffer.put(raw)
async def get_tick(self):
return await self.buffer.get()嗯，这里要注意：队列一定要设置最大长度。我曾经遇到过数据生产速度远大于消费速度，结果内存爆了，整个进程被 OOM Killer 干掉。那叫一个惨。
3.2 数据清洗：脏数据是策略的毒药
交易所的数据就一定干净吗？不一定。我遇到过的情况包括：
- 重复推送：同一笔成交推了两次
- 乱序到达：后发生的成交先到了
- 异常值：价格突然跳到一个离谱的数字
- 缺失数据：某几秒的 Tick 完全没收到
清洗策略其实不复杂，但必须做。我个人习惯用「三关过滤法」：
- 第一关：去重。用 (交易所时间戳 + 成交ID) 做唯一键，重复的直接扔掉。
- 第二关：排序。维护一个滑动窗口，窗口内的数据按时间戳重排。窗口大小我一般设 100ms。
- 第三关：合理性校验。价格不能超过上一笔的 ±10%，成交量不能为负，等等。
避坑指南：我曾经在排序环节吃过亏。窗口设得太小（比如 10ms），结果网络抖动时数据还是乱的。后来改成 100ms，问题就解决了。但窗口太大又会增加延迟，这个平衡点需要根据你的网络环境来调。
3.3 特征工程：从原始数据到策略信号
原始 Tick 数据，说白了就是一堆数字。你得把它变成策略能理解的特征。
Bar 的构建是第一步。我常用的几种方式：
| Bar 类型 | 构建方式 | 适用场景 |
|---|---|---|
| 时间 Bar | 固定时间窗口（如 1分钟） | 趋势跟踪、均值回归 |
| 成交量 Bar | 固定成交量（如 1000手） | 高波动品种、避免无成交时段 |
| Tick Bar | 固定笔数（如 100笔） | 微观结构分析、高频策略 |
我个人偏好成交量 Bar。为什么？因为时间 Bar 在夜盘或者流动性差的时候，可能一根 Bar 里只有几笔成交，那算出来的指标根本没法用。成交量 Bar 能保证每根 Bar 都有足够的样本量。
特征计算这块，我建议用增量更新。别每次来一个新 Tick 就把所有数据重算一遍，那太慢了。
class IncrementalEMA:
def __init__(self, alpha):
self.alpha = alpha
self.value = None
def update(self, price):
if self.value is None:
self.value = price
else:
self.value = self.alpha * price + (1 - self.alpha) * self.value
return self.value你看，EMA 的增量更新就一行公式。像布林带、RSI 这些指标，也都有对应的增量算法。你想想看，如果每次都要遍历过去 1000 根 Bar，那延迟得多高。
3.4 缓存策略：内存里的「高速公路」
数据清洗完了，特征算好了，接下来要存起来。但注意，实盘场景下，你不能每次都去查数据库。磁盘 IO 太慢了。
我一般用 三层缓存架构：
- L1 缓存：进程内内存。存最近 1 分钟的 Tick 数据，用环形缓冲区实现。读写都是 O(1)。
- L2 缓存：共享内存。存最近 1 小时的 Bar 数据，多个策略进程可以共享。我用的是 Redis，但注意要开 AOF 持久化。
- L3 缓存：本地文件。存当天的完整数据，用于盘后复盘。格式我推荐用 Parquet，压缩率高，读取快。
注意：L1 缓存一定要用预分配内存。我见过有人用 Python 的 list 动态增长，结果在行情剧烈波动时，频繁的内存分配导致 GC 暂停，策略直接错过了信号。用 array 或者 numpy 的固定大小数组，提前把内存申请好。
这里画个图，帮你理解整个数据流的走向：
你看这个流程，数据从交易所出来，经过接入、清洗、特征工程，最后落到缓存里，策略引擎直接从缓存取数据。每一步都是流水线作业，互不阻塞。
3.5 实战中的几个坑
最后分享几个我踩过的坑，希望能帮你省点时间：
- 时间同步：交易所的时间戳和你的服务器时间可能不一致。我建议用 NTP 同步，并且所有数据都统一用交易所时间戳，别用本地时间。
- 断线重连：WebSocket 一定会断。重连时要注意数据补全，别出现缺口。我一般会在重连后拉取最近 10 秒的补数数据。
- 内存泄漏：Python 的循环引用会导致内存泄漏。我习惯用 weakref 来管理缓存中的对象引用。
- 日志记录：每条数据的处理耗时都要记下来。这样出了问题，你能快速定位是哪个环节慢了。
一个小技巧：在开发阶段，我会把实盘数据录下来，存成文件。然后在回测环境里「回放」这些数据，用来测试数据管道的吞吐量和稳定性。这比直接上实盘安全多了。
数据流处理这块，说白了就是「快」和「稳」的平衡。太快了容易出错，太稳了又跟不上行情。我个人觉得，先保证数据不丢、不乱，再慢慢优化速度。毕竟，一个慢但正确的信号，比一个快但错误的信号，要值钱得多。