第1章：实时数据流接入
做市系统里，数据流就是血液。没有实时、干净的数据，再牛的策略也是白搭。这一章，咱们就聊聊怎么把行情数据从交易所拉回来，再洗干净，喂给我们的库存计算引擎。
1.1 WebSocket行情接入
我个人习惯，做市系统的行情接入首选WebSocket。为什么？因为它轻量、全双工，延迟低。比起轮询HTTP接口，WebSocket能省下不少带宽和CPU。
我记得刚开始做这个项目时，团队里有人提议用TCP长连接自己解析协议。我直接否了——没必要重复造轮子。交易所基本都支持WebSocket，咱们直接用现成的库就行。
1.1.1 连接管理
连接WebSocket，核心就三步：握手、订阅、心跳。
import asyncio
import websockets
async def connect_exchange():
uri = "wss://api.exchange.com/ws"
async with websockets.connect(uri) as ws:
# 订阅行情
subscribe_msg = {
"type": "subscribe",
"channel": "orderbook",
"symbol": "BTC-USDT"
}
await ws.send(json.dumps(subscribe_msg))
# 持续接收数据
async for message in ws:
process_message(message)
这里有个坑：交易所的WebSocket连接经常断。我遇到过最夸张的一次，某交易所一天断了七八次。所以，重连机制必须做。
1.1.2 心跳保活
很多交易所要求客户端定期发心跳。一般是每15-30秒发一次。如果超过一定时间没收到心跳，服务器就会断开连接。
async def heartbeat(ws, interval=15):
while True:
await asyncio.sleep(interval)
try:
await ws.send(json.dumps({"type": "ping"}))
except:
break
1.2 交易所API对接
不同交易所的API风格差异很大。有的用RESTful，有的用WebSocket，还有的混合使用。我建议统一封装一层，屏蔽底层差异。
1.2.1 统一接口设计
我一般会定义一个抽象基类，所有交易所都继承它：
class ExchangeAPI(ABC):
@abstractmethod
async def subscribe_orderbook(self, symbol: str):
pass
@abstractmethod
async def subscribe_trade(self, symbol: str):
pass
@abstractmethod
async def get_account_info(self) -> dict:
pass
这样，上层策略代码就不用关心具体连的是哪个交易所。换交易所时，只需要换一个实现类就行。
1.2.2 鉴权与签名
交易接口通常需要鉴权。常见的做法是API Key + Secret Key，加上时间戳和签名。签名算法各家不同，但核心逻辑差不多：
def sign_request(params: dict, secret: str) -> str:
sorted_params = sorted(params.items())
query_string = "&".join(f"{k}={v}" for k, v in sorted_params)
signature = hmac.new(
secret.encode(),
query_string.encode(),
hashlib.sha256
).hexdigest()
return signature
1.3 数据格式解析
行情数据到了，但格式五花八门。JSON最常见，也有用Protocol Buffers的。解析速度直接影响系统延迟。
1.3.1 JSON解析优化
Python自带的json库够用，但性能一般。我推荐用orjson或ujson，速度能快3-5倍。
import orjson
def parse_orderbook(raw: bytes) -> dict:
# orjson直接解析bytes，省去decode步骤
data = orjson.loads(raw)
return {
"bids": [(float(p), float(q)) for p, q in data["bids"]],
"asks": [(float(p), float(q)) for p, q in data["asks"]],
"timestamp": data["ts"]
}
你想想看，每秒几千条行情，每条省0.1毫秒，整体延迟就能降不少。
1.3.2 二进制协议解析
有些交易所用Protobuf或自定义二进制协议。解析起来麻烦点，但带宽占用小。我做过一个项目，从JSON切到Protobuf后，带宽节省了60%。
# 假设交易所用Protobuf定义行情
import market_pb2
def parse_protobuf(raw: bytes):
orderbook = market_pb2.OrderBook()
orderbook.ParseFromString(raw)
return orderbook
1.4 数据清洗与校验
数据到了，但不能直接用。交易所偶尔会发脏数据——重复的、乱序的、甚至错误的。清洗这一步，说白了就是去伪存真。
1.4.1 去重
WebSocket偶尔会重发消息。我一般用序列号去重：
class Deduplicator:
def __init__(self):
self.seen = set()
self.max_size = 10000
def is_duplicate(self, seq_id: int) -> bool:
if seq_id in self.seen:
return True
self.seen.add(seq_id)
if len(self.seen) > self.max_size:
# 清理旧数据
self.seen.clear()
return False
1.4.2 乱序处理
行情数据到达顺序可能和生成顺序不一致。我遇到过这种情况：先收到t+1时刻的数据，后收到t时刻的。如果不处理，库存计算会出错。
class OrderBookReconstructor:
def __init__(self):
self.buffer = []
self.last_seq = 0
def add_update(self, seq: int, update: dict):
if seq <= self.last_seq:
return # 丢弃旧数据
self.buffer.append((seq, update))
self.buffer.sort(key=lambda x: x[0]) # 按序列号排序
# 处理连续的数据
while self.buffer and self.buffer[0][0] == self.last_seq + 1:
_, update = self.buffer.pop(0)
self.apply_update(update)
self.last_seq += 1
1.4.3 数据校验
校验规则其实不复杂，但必须做：
- 价格不能为负数或零
- 数量不能为负数
- 时间戳不能是未来的
- 买卖盘口不能交叉（买一价 < 卖一价）
def validate_orderbook(data: dict) -> bool:
if data["bids"][0][0] >= data["asks"][0][0]:
return False # 盘口交叉，数据异常
if data["timestamp"] > time.time() * 1000 + 1000:
return False # 时间戳异常
return True
1.5 整体架构
说了这么多，咱们看看整体数据流长什么样：
这张图展示了数据从交易所到库存计算引擎的完整路径。每一步都有它的作用，缺一不可。
1.6 性能优化要点
做市系统对延迟极其敏感。我总结了几条优化经验：
| 优化点 | 做法 | 效果 |
|---|---|---|
| 内存分配 | 预分配对象池，避免频繁GC | 延迟降低30% |
| 序列化 | 用orjson代替标准json | 解析速度提升3倍 |
| 网络IO | 使用异步IO，避免阻塞 | 吞吐量提升5倍 |
| 数据校验 | 轻量校验，不做复杂计算 | 延迟增加<1% |
嗯，数据流接入这块就聊到这儿。说白了，核心就是三件事：连得上、读得懂、洗得净。把这三点做好，后面的库存计算才能稳如磐石。