4、协议解析入门：二进制数据与字节序、struct模块的使用、解析固定长度消息头

好，咱们进入实战环节了。

前面几章我们聊了行情数据的整体架构，也看了网络传输的底层逻辑。但有个问题一直悬着——数据到了我们手里，怎么把它读懂？

说白了，交易所发过来的就是一串二进制流。0和1的排列组合。你看着是乱码，但人家有严格的规矩。这个规矩，就是协议。

我刚开始做量化那会儿，第一次拿到深交所的行情快照，打开一看全是乱码。当时我还以为是网络传输出了问题，折腾了半天。后来才明白，是我自己没按规矩来解析。

嗯，今天我们就来搞定这件事。

4.1 二进制数据长什么样？

先看个最简单的例子。假设我们要传输一个整数 12345。

用文本协议，它会被编码成字符串 "12345"，占5个字节。用二进制协议，它就是一个16位的整数，占2个字节。

哪个效率高？显然是二进制。这也是为什么所有交易所的行情协议都选择二进制格式——快，省带宽。

但二进制有个麻烦：字节序。

字节序（Endianness） 指的是多字节数据在内存中的排列顺序。

  • 大端序（Big-Endian）：高位字节在前，低位字节在后。像我们写数字一样自然。
  • 小端序（Little-Endian）：低位字节在前，高位字节在后。Intel x86 架构的默认方式。

举个例子，整数 0x12345678：

字节序 内存排列（从低地址到高地址）
大端序 12 34 56 78
小端序 78 56 34 12

我在项目中遇到过最坑的一次：从某家期货公司拿到的行情数据，文档写的是大端序，结果实际发过来是小端序。解析出来的价格全是天文数字。排查了整整一个下午，最后发现是文档写错了。

避坑指南：拿到任何协议文档，第一件事就是确认字节序。不要相信文档，要拿实际数据验证。

4.2 Python 的 struct 模块

Python 里处理二进制数据，最趁手的工具就是 struct 模块。它能把 Python 的数据类型和 C 语言的结构体互相转换。

核心就两个函数：

  • pack(fmt, v1, v2, ...)：把 Python 值打包成二进制字节串
  • unpack(fmt, buffer)：从二进制字节串中解包出 Python 值

其中 fmt 是格式字符串，它决定了怎么打包/解包。

来看个例子：

import struct

# 打包一个整数和一个浮点数
data = struct.pack('>if', 12345, 3.14)
print(data)  # b'\x00\x00\x30\x39@H\xf5\xc3'

# 解包回来
val1, val2 = struct.unpack('>if', data)
print(val1, val2)  # 12345 3.140000104904175

注意那个 '>' 符号，它表示大端序。如果是小端序，用 '<'。

常用格式字符：

字符C 类型Python 类型字节数
ccharbytes (长度1)1
hshortint2
iintint4
qlong longint8
ffloatfloat4
ddoublefloat8
schar[]bytes指定长度

4.3 解析固定长度消息头

几乎所有行情协议，都会有一个固定长度的消息头。消息头里通常包含：

  • 消息类型（2字节或4字节）
  • 消息体长度（2字节或4字节）
  • 序列号（4字节或8字节）
  • 时间戳（4字节或8字节）

我以某交易所的行情协议为例，它的消息头结构如下：

字段类型字节数说明
MsgTypeuint162消息类型，大端序
BodyLengthuint162消息体长度，大端序
SeqNouint324序列号，大端序
SendTimeuint648发送时间（纳秒），大端序

消息头一共16个字节。解析代码很简单：

import struct

HEADER_FORMAT = '>HHIQ'  # 2+2+4+8 = 16 字节
HEADER_SIZE = struct.calcsize(HEADER_FORMAT)

def parse_header(data):
    """解析固定长度消息头"""
    if len(data) < HEADER_SIZE:
        raise ValueError(f'数据长度不足，需要{HEADER_SIZE}字节，实际{len(data)}字节')
    
    msg_type, body_length, seq_no, send_time = struct.unpack(HEADER_FORMAT, data[:HEADER_SIZE])
    
    return {
        'msg_type': msg_type,
        'body_length': body_length,
        'seq_no': seq_no,
        'send_time': send_time,
        'header_size': HEADER_SIZE
    }

你看，就这么几行代码。但实际项目中，我会加一些校验逻辑：

def parse_header_safe(data):
    """带校验的消息头解析"""
    if len(data) < HEADER_SIZE:
        return None, f'数据长度不足，需要{HEADER_SIZE}字节'
    
    try:
        header = parse_header(data)
    except struct.error as e:
        return None, f'struct解包失败: {e}'
    
    # 校验消息体长度是否合理
    if header['body_length'] > 1024 * 1024:  # 超过1MB，明显不合理
        return None, f'消息体长度异常: {header["body_length"]}'
    
    # 校验序列号是否连续（可选）
    # 这里可以维护一个全局的 last_seq_no 做对比
    
    return header, None

我曾经在生产环境遇到过一个问题：某次行情数据被截断了，消息头里的 body_length 指向了一个巨大的值。结果程序直接申请了 2GB 内存，然后 OOM 崩溃了。从那以后，我所有解析代码里都加了长度校验。

4.4 完整解析流程

有了消息头，解析完整消息就很简单了：

def parse_message(data):
    """解析完整消息"""
    header, err = parse_header_safe(data)
    if err:
        return None, err
    
    total_size = header['header_size'] + header['body_length']
    if len(data) < total_size:
        return None, f'数据不完整，需要{total_size}字节，实际{len(data)}字节'
    
    body = data[header['header_size']:total_size]
    
    # 根据消息类型，解析不同的消息体
    if header['msg_type'] == 1:
        return parse_market_data(body, header)
    elif header['msg_type'] == 2:
        return parse_order_book(body, header)
    else:
        return None, f'未知消息类型: {header["msg_type"]}'

这个模式很通用。你想想看，不管后面消息体多复杂，只要消息头固定，解析流程都是一样的：

  1. 先读固定长度的消息头
  2. 从消息头里拿到消息体长度
  3. 再读消息体
  4. 根据消息类型做不同的处理

4.5 知识体系图

下面这张图，把本章的核心逻辑串起来了：

二进制协议解析核心流程 原始二进制数据流 字节序判断（大端/小端） struct.unpack 解析消息头 MsgType(2B) | BodyLength(2B) | SeqNo(4B) | SendTime(8B) 解析后的结构化数据

这张图其实就说了三件事：

  • 拿到数据，先别急着解包，搞清楚字节序
  • 用 struct，按格式字符串解包消息头
  • 拿到消息头，就知道后面怎么处理了

我个人习惯，在写任何行情解析代码之前，都会先画这么一张图。把流程理清楚了，代码写起来就顺了。

小技巧：调试二进制数据时，用 hexdump 或者 Python 的 binascii.hexlify 把数据打印成十六进制，肉眼比对协议文档，能快速定位问题。

好了，二进制解析的入门就到这里。下一节我们会深入消息体的解析，看看那些复杂的行情数据结构怎么处理。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →