第二章：数据源与接入
好，咱们直接进入正题。订单簿系统要跑起来，第一步就是搞定数据源。说白了，你得知道从哪儿拿数据，怎么拿，拿到手的是什么格式。这一步要是没做好，后面所有的高频策略、低延迟优化全是空中楼阁。
2.1 交易所API的三种形态
我个人习惯把交易所API分成三类：REST、WebSocket、以及FIX协议。咱们做量化交易，接触最多的就是前两种。
2.1.1 REST接口
REST接口，你可以理解成“你问一句，它答一句”。每次请求都得建立连接、发请求、等响应、断开连接。这玩意儿延迟高，但胜在简单可靠。我一般用它来做三件事：
- 获取历史数据：比如拉取过去24小时的成交记录
- 查询账户状态：看看余额、持仓、挂单情况
- 提交订单：虽然WebSocket也能下单，但REST更稳妥
嗯，这里要注意：REST接口有频率限制。我曾经在某个小交易所上吃过亏，一个循环里忘了加sleep，直接触发了限流，IP被封了半小时。那半小时里策略完全停摆，亏了不少。
2.1.2 WebSocket流
WebSocket就不一样了。它是一条长连接，交易所主动往你这边推数据。延迟低，实时性好。做订单簿的实时更新，必须用WebSocket。
你想想看，如果订单簿每秒变化几百次，你用REST去轮询，那延迟得有多高？等你拿到数据，市场早变样了。
我常用的WebSocket订阅包括：
- 深度流（Depth Stream）：订单簿的增量更新
- 成交流（Trade Stream）：实时成交数据
- K线流（Kline Stream）：实时K线更新
2.2 数据格式解析
交易所返回的数据，主流格式就两种：JSON和二进制。JSON可读性好，但解析慢；二进制快，但得自己写解析器。
2.2.1 JSON格式
大部分交易所的REST接口和WebSocket消息都用JSON。举个例子，Binance的深度数据长这样：
{
"lastUpdateId": 1027024,
"bids": [
["4.00000000", "431.00000000"],
["3.98000000", "200.00000000"]
],
"asks": [
["4.00000200", "12.00000000"],
["4.00000300", "100.00000000"]
]
}这里bids是买单，asks是卖单。每个数组里第一个是价格，第二个是数量。解析起来很简单，Python里用json.loads()就行。但要注意，高频场景下频繁解析JSON会成为瓶颈。
2.2.2 二进制格式
一些专业级的交易所（比如Coinbase Pro、Deribit）会提供二进制格式的数据流。说白了，就是直接用字节流传输数据，省去了序列化和反序列化的开销。
举个例子，一个订单簿的增量更新可能只有几十个字节：
struct DepthUpdate {
uint64_t sequence; // 8字节
uint32_t side; // 4字节，0=买，1=卖
double price; // 8字节
double quantity; // 8字节
}; // 总共28字节同样的数据用JSON表示，至少得100多字节。而且解析二进制数据，C++里直接内存拷贝就行，比JSON快一个数量级。
2.3 数据接入的架构设计
好了，现在你知道怎么拿数据、怎么解析了。但实际系统里，数据源不止一个。你可能同时接入多个交易所，每个交易所又有多个数据流。怎么组织？
我个人习惯用“数据网关”模式。每个交易所对应一个网关进程，网关负责：
- 建立并维护WebSocket连接
- 解析原始数据
- 做初步的校验和清洗
- 把标准化后的数据推给下游
下面这张图是我常用的架构：
你看，每个交易所的数据先经过各自的网关做标准化处理，然后统一推送到消息队列。订单簿引擎从消息队列里消费数据，这样就把数据源和业务逻辑解耦了。
2.4 数据校验与清洗
数据拿到手，不能直接用。交易所偶尔会发错数据，或者网络传输过程中出现乱码。我见过最离谱的一次，某个交易所的WebSocket流里突然冒出一条价格是负数的订单。你想想看，如果策略没做校验，直接拿这个负数价格去算，结果会怎样？
我常用的校验规则：
- 价格合理性：价格必须大于0，且不能超过当前市场价的10倍
- 数量合理性：数量必须大于0，且不能超过某个阈值
- 时间戳单调性：后一条数据的时间戳不能比前一条还早
- 序列号连续性：如果交易所提供了序列号，检查是否连续，不连续说明有丢包
2.5 实战：Python接入WebSocket深度流
最后，给一个简单的Python示例。假设我们要接入Binance的BTC/USDT深度流：
import asyncio
import websockets
import json
async def subscribe_depth():
uri = "wss://stream.binance.com:9443/ws/btcusdt@depth20"
async with websockets.connect(uri) as ws:
print("连接成功，开始接收数据...")
while True:
try:
data = await asyncio.wait_for(ws.recv(), timeout=10)
parsed = json.loads(data)
# 这里把parsed传给订单簿引擎
print(f"收到更新，序列号: {parsed['lastUpdateId']}")
except asyncio.TimeoutError:
# 10秒没收到数据，发个ping
await ws.ping()
print("发送心跳...")
if __name__ == "__main__":
asyncio.run(subscribe_depth())这段代码很简单，但实际生产环境里，你还需要处理重连、多路复用、数据缓冲等逻辑。嗯，这些内容我们后面章节会详细讲。