第4章事件驱动模型

一、事件驱动模型：WebSocket 流式数据

做量化交易的朋友都知道，订单簿数据是高频交易的核心。但有个问题——全量数据太大了。你想想看，每秒几千笔成交，每笔都传全量订单簿，带宽和CPU都扛不住。所以，业界普遍采用增量更新。

增量更新的核心，就是事件驱动模型。说白了，就是「有变化才通知」。我刚开始接触这个思路时，觉得挺简单，但真正落地才发现坑不少。今天咱们就聊聊WebSocket流式数据、事件类型定义，以及序列化反序列化这些事儿。

核心思想： 事件驱动模型 = 状态变化 → 事件 → 通知 → 处理。不是轮询，是推送。

1.1 WebSocket 流式数据的特点

WebSocket 和 HTTP 最大的区别是什么？HTTP 是「你问一句，我答一句」。WebSocket 是「我主动告诉你」。在订单簿场景下，交易所一旦有变化，就通过 WebSocket 把增量数据推过来。

我个人习惯把 WebSocket 流式数据分为三类：

快照数据（Snapshot）：刚连接时，交易所给你一份完整的订单簿。这是基准。
增量数据（Delta）：后续的每次变化，只传变化的部分。比如「卖一价从 100 变成 101」。
心跳数据（Heartbeat）：防止连接断开，定期发个空包。

我在项目中遇到过一个问题：快照和增量数据到达的顺序可能乱掉。比如你先收到增量，后收到快照。那怎么办？嗯，这里要注意——必须等快照到位后，才能处理增量。否则你的订单簿就是错的。

避坑指南： 我曾经因为没处理好快照和增量的顺序，导致回测数据全错了。后来加了个状态机：WAITING_SNAPSHOT → SNAPSHOT_RECEIVED → PROCESSING_DELTA。只有状态为 SNAPSHOT_RECEIVED 时，才处理增量。

1.2 事件类型定义

事件类型怎么定义？说白了就是给每种变化起个名字。常见的订单簿事件类型有：

事件类型	含义	示例
`NEW`	新增一个价格档位	原来没有 100.5 这个价，现在有了
`UPDATE`	更新现有档位的数量	卖一数量从 100 变成 200
`DELETE`	删除一个价格档位	买一档被吃光了，消失
`CLEAR`	清空整个订单簿	很少见，但某些交易所会发

你可能会问：为什么不用 INSERT 而用 NEW？其实都可以，但 NEW 更直观。我个人习惯用枚举类来定义：

from enum import Enum

class OrderBookEventType(Enum):
    NEW = "new"
    UPDATE = "update"
    DELETE = "delete"
    CLEAR = "clear"

这样写的好处是——类型安全。你不会手滑写成 "updat" 这种错别字。我曾经在日志里看到过 "updat"，排查了半天才发现是拼写错误。从那以后，所有事件类型我都用枚举。

1.3 序列化与反序列化

事件定义好了，怎么在网络上传输？答案是序列化。把 Python 对象变成字节流，发出去。对方收到后，再反序列化回来。

常见的序列化格式有：

JSON：可读性好，但体积大，解析慢。适合低频场景。
MessagePack：比 JSON 小，解析快。我比较推荐。
Protocol Buffers：体积最小，但需要定义 .proto 文件。适合超高频。
FlatBuffers：零拷贝，极快。但学习成本高。

我个人习惯用 MessagePack。为什么？因为它在「开发效率」和「性能」之间取得了平衡。JSON 太慢，Protobuf 太麻烦。MessagePack 刚刚好。

来看一个具体的例子。假设我们有一个增量事件：

{
    "type": "update",
    "side": "bid",
    "price": 100.5,
    "size": 200,
    "timestamp": 1699000000000
}

用 MessagePack 序列化：

import msgpack

event = {
    "type": "update",
    "side": "bid",
    "price": 100.5,
    "size": 200,
    "timestamp": 1699000000000
}

packed = msgpack.packb(event)
print(f"序列化后大小: {len(packed)} 字节")
# 输出: 序列化后大小: 28 字节

# 反序列化
unpacked = msgpack.unpackb(packed)
print(unpacked)
# 输出: {'type': 'update', 'side': 'bid', 'price': 100.5, 'size': 200, 'timestamp': 1699000000000}

你看，28 个字节就搞定了。如果用 JSON，至少 50 字节以上。别小看这几十字节，每秒几万笔数据，差距就出来了。

注意： 浮点数精度问题。订单簿的价格和数量，建议用整数表示（比如价格乘以 10000 变成整数）。否则序列化反序列化过程中，可能出现 100.5 变成 100.4999999 的情况。我踩过这个坑，回测时发现订单簿对不上，查了两天才找到原因。

1.4 事件处理流程

好了，现在事件类型定义好了，序列化也搞定了。那整个流程是什么样的？我画了一张图，你看一眼就明白了：

流程其实不复杂：交易所通过 WebSocket 推送数据 → 事件接收器拿到原始字节流 → 反序列化成 Python 对象 → 事件解析器判断事件类型 → 事件处理器更新订单簿。

但这里有个细节：事件处理器怎么知道当前是快照还是增量？我一般会在事件里加一个字段 event_type，值为 "snapshot" 或 "delta"。这样解析器一看就知道该怎么处理。

小技巧： 事件处理器里，我习惯用 match-case 语法（Python 3.10+）。比一堆 if-elif 清晰多了：

match event["type"]:
    case "new":
        order_book.add_level(event["side"], event["price"], event["size"])
    case "update":
        order_book.update_level(event["side"], event["price"], event["size"])
    case "delete":
        order_book.remove_level(event["side"], event["price"])
    case "clear":
        order_book.clear()
    case _:
        logger.warning(f"未知事件类型: {event['type']}")

1.5 性能优化建议

最后，聊几个性能优化的点。这些都是我在实际项目中踩过的坑：

内存池：频繁创建和销毁事件对象，GC 压力很大。我建议用对象池，复用事件对象。
批量处理：不要来一个事件就更新一次订单簿。攒一批，比如 10 毫秒处理一次。吞吐量能提升 3-5 倍。
零拷贝：如果使用 FlatBuffers，反序列化时不需要拷贝数据，直接读取。适合极致性能场景。
异步处理：WebSocket 接收和事件处理用不同的协程。接收只管收，处理只管处理。互不阻塞。

嗯，今天就聊到这儿。事件驱动模型是订单簿增量更新的基石。搞懂了它，后面的内容就顺了。

📚 课程目录

一、事件驱动模型：WebSocket 流式数据

1.1 WebSocket 流式数据的特点

1.2 事件类型定义

1.3 序列化与反序列化

1.4 事件处理流程

1.5 性能优化建议