四、行情订阅实战：合约代码规则、订阅深度行情（MarketData）、处理行情快照与增量数据
行情订阅，是量化交易系统的心脏起搏器。没有实时、准确的行情数据，你的策略再牛也白搭。今天我们就来聊聊，怎么把交易所的行情数据稳稳地接进来。
4.1 合约代码规则：别让命名坑了你
每家交易所的合约代码规则都不一样。我刚开始做期货量化时，就吃过这个亏。以为所有交易所都是「品种+月份」的格式，结果对接某商品交易所时，发现它的主力连续合约代码里藏着字母后缀，直接导致订阅失败。
咱们先看看主流的规则：
| 交易所 | 合约代码示例 | 规则说明 |
|---|---|---|
| 中金所 | IF2403 | 品种代码（IF）+ 年份后两位（24）+ 月份（03） |
| 上期所 | rb2405 | 小写品种代码（rb）+ 年份后两位 + 月份 |
| 大商所 | m2405 | 小写品种代码 + 年份后两位 + 月份 |
| 郑商所 | SR405 | 大写品种代码（SR）+ 年份后两位 + 月份 |
| 上期能源 | sc2406 | 小写品种代码 + 年份后两位 + 月份 |
嗯，这里要注意一点：主力合约和连续合约的代码规则是独立的。比如中金所的IF主力连续合约代码是IF00，而上期所的主力连续可能是rb888。我建议你在代码里维护一个合约代码映射表，别硬编码。
避坑指南：我曾经在对接CTP时，发现某些合约在非交易时段订阅会返回错误。后来才知道，交易所对非上市合约的订阅请求会直接拒绝。所以订阅前，一定要先检查合约状态。
4.2 订阅深度行情：不只是买卖一档
深度行情（MarketData），说白了就是买卖盘口的快照。但很多新手只订阅买卖一档，这其实不够用。做高频策略或者套利策略时，你需要的是五档甚至十档的深度数据。
订阅深度行情的流程，我一般这么设计：
# 伪代码示例：订阅深度行情
class MarketDataSubscriber:
def __init__(self, api_connection):
self.api = api_connection
self.subscribed_contracts = set()
def subscribe_depth(self, contract_code, depth_level=5):
"""
订阅指定合约的深度行情
:param contract_code: 合约代码，如 'IF2403'
:param depth_level: 深度档位，默认5档
"""
# 检查合约是否已订阅
if contract_code in self.subscribed_contracts:
print(f"合约 {contract_code} 已订阅，跳过")
return
# 构建订阅请求
req = {
'type': 'SUBSCRIBE_MARKET_DATA',
'contract': contract_code,
'depth': depth_level,
'is_snapshot': True # 首次订阅需要快照
}
# 发送订阅请求
response = self.api.send_request(req)
if response['status'] == 'success':
self.subscribed_contracts.add(contract_code)
print(f"成功订阅 {contract_code}，深度 {depth_level} 档")
else:
print(f"订阅失败: {response['error_msg']}")
def unsubscribe_depth(self, contract_code):
"""取消订阅"""
if contract_code in self.subscribed_contracts:
req = {
'type': 'UNSUBSCRIBE_MARKET_DATA',
'contract': contract_code
}
self.api.send_request(req)
self.subscribed_contracts.remove(contract_code)
你想想看，为什么订阅时要指定深度档位？因为不同策略对数据精度的要求不一样。做市商策略可能需要十档数据，而趋势跟踪策略可能只看买卖一档就够了。订阅太多不必要的数据，只会增加网络带宽和内存压力。
4.3 处理行情快照与增量数据：别漏掉任何一笔
行情数据分为两种：快照（Snapshot）和增量（Incremental）。快照是某一时刻的完整盘口状态，增量是后续的变动数据。
我个人习惯的做法是：以快照为基准，用增量来更新。具体来说：
- 首次订阅时，获取一个完整的快照
- 后续收到的增量数据，在快照基础上进行加减操作
- 定期（比如每10秒）重新请求一次快照，防止累积误差
为什么会这样？因为增量数据在传输过程中可能丢包，或者顺序错乱。如果不定期校准，你的本地盘口数据就会和交易所的真实数据越差越远。
小技巧：增量数据通常包含一个序列号（Sequence Number）。你可以用这个序列号来检测是否有数据丢失。如果发现序列号不连续，立即请求一次快照进行校准。
下面是一个处理增量数据的核心逻辑：
class OrderBook:
def __init__(self, contract_code):
self.contract = contract_code
self.bids = {} # 买盘，价格 -> 数量
self.asks = {} # 卖盘，价格 -> 数量
self.last_snapshot_time = 0
def apply_snapshot(self, snapshot):
"""应用快照数据"""
self.bids.clear()
self.asks.clear()
for bid in snapshot['bids']:
self.bids[bid['price']] = bid['volume']
for ask in snapshot['asks']:
self.asks[ask['price']] = ask['volume']
self.last_snapshot_time = time.time()
def apply_incremental(self, incremental):
"""应用增量数据"""
# 处理买盘变动
for bid_change in incremental['bid_changes']:
price = bid_change['price']
volume = bid_change['volume']
if volume == 0:
# 数量为0表示删除该价位
self.bids.pop(price, None)
else:
# 更新或新增价位
self.bids[price] = volume
# 处理卖盘变动
for ask_change in incremental['ask_changes']:
price = ask_change['price']
volume = ask_change['volume']
if volume == 0:
self.asks.pop(price, None)
else:
self.asks[price] = volume
# 检查是否需要重新校准
if time.time() - self.last_snapshot_time > 10:
self.request_snapshot() # 请求新的快照
def get_best_bid_ask(self):
"""获取最优买卖价"""
if not self.bids or not self.asks:
return None
best_bid = max(self.bids.keys())
best_ask = min(self.asks.keys())
return {
'bid_price': best_bid,
'bid_volume': self.bids[best_bid],
'ask_price': best_ask,
'ask_volume': self.asks[best_ask]
}
警告：增量数据中的价格和数量，一定要用整数或定点数处理，别用浮点数。浮点数的精度问题在频繁计算时会累积成灾难。我在一个项目中就因为这个原因，导致套利策略的价差计算偏差了0.5个tick，亏了不少钱。
4.4 整体架构：一张图看懂
说了这么多，咱们用一张图来总结行情订阅的整体流程：
从这张图可以看出，行情订阅的核心就是三个环节：订阅管理、数据分发、本地维护。订阅管理器负责和交易所网关通信，数据分发器把原始数据拆成快照和增量两条路径，最后在本地OrderBook中合并成完整的盘口数据。
嗯，这里要特别强调一下：数据分发器一定要做去重和排序。因为网络延迟可能导致增量数据乱序到达，如果不处理，你的OrderBook就会乱套。我一般会在分发器里维护一个环形缓冲区，按序列号排序后再分发。
核心要点：行情订阅不是简单的「订阅-接收」两步走。它需要你处理好合约代码映射、深度档位选择、快照与增量的合并、数据校准等多个环节。任何一个环节出问题，你的策略都可能跑偏。
最后说一句，行情数据的性能优化，其实就两个方向：减少不必要的数据传输和加快本地处理速度。前者靠合理的订阅策略，后者靠高效的数据结构和算法。这两点做好了，你的行情系统就能稳稳地跑上几年。
公众号:蓝海资料掘金营，微信deep3321