4、事件类型定义：OrderEvent、TradeEvent、BookUpdateEvent、ErrorEvent、自定义事件

事件驱动架构的核心是什么？说白了，就是「事件」本身。

你想想看，整个交易系统就像一条流水线。订单进来是一个事件，成交了是一个事件，盘口变了又是一个事件。每个事件都带着自己的数据，在系统的各个模块之间流转。我做了这么多年量化，见过太多系统因为事件定义混乱而崩掉的案例。

今天我们就来聊聊，在订单簿事件驱动框架里，事件类型到底该怎么定义。

4.1 事件基类：所有事件的爸爸

我个人习惯，先定义一个事件基类。所有具体的事件类型都继承它。这样有个好处——统一接口，方便扩展。

from dataclasses import dataclass
from datetime import datetime
from typing import Any, Dict, Optional

@dataclass
class Event:
    """事件基类"""
    event_id: str
    timestamp: datetime
    event_type: str
    data: Optional[Dict[str, Any]] = None
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            'event_id': self.event_id,
            'timestamp': self.timestamp.isoformat(),
            'event_type': self.event_type,
            'data': self.data
        }

嗯，这里要注意。我用的是 dataclass，不是普通类。为什么？因为 dataclass 自动生成 __init__、__repr__ 这些方法，省事。而且数据类在事件传递时更清晰，不容易出幺蛾子。

4.2 OrderEvent：订单的生命周期

订单事件，是交易系统里最核心的事件之一。从下单、撤单到订单状态变更，每一步都是一个 OrderEvent。

我在项目中遇到过一个问题：一开始只定义了「订单创建」和「订单成交」两个事件，结果后来要加「部分成交」和「订单修改」，代码改得我头皮发麻。所以，我建议你一开始就把事件类型定义全。

@dataclass
class OrderEvent(Event):
    """订单事件"""
    order_id: str
    symbol: str
    side: str  # 'buy' or 'sell'
    order_type: str  # 'limit', 'market', etc.
    price: float
    quantity: float
    filled_quantity: float = 0.0
    status: str = 'pending'  # pending, open, filled, cancelled, rejected
    
    def __post_init__(self):
        self.event_type = 'order'

核心事件类型枚举：

  • ORDER_CREATED - 订单创建
  • ORDER_FILLED - 订单完全成交
  • ORDER_PARTIALLY_FILLED - 订单部分成交
  • ORDER_CANCELLED - 订单撤销
  • ORDER_REJECTED - 订单被拒绝
  • ORDER_MODIFIED - 订单修改

4.3 TradeEvent：成交才是硬道理

TradeEvent 和 OrderEvent 有什么区别？很多人搞混。

OrderEvent 描述的是「订单」的状态变化。而 TradeEvent 描述的是「一笔成交」——也就是买方和卖方匹配成功的那一刻。

你可以这么理解：一个订单可能产生多笔成交（比如冰山订单），每一笔成交都是一个独立的 TradeEvent。

@dataclass
class TradeEvent(Event):
    """成交事件"""
    trade_id: str
    order_id: str
    symbol: str
    side: str
    price: float
    quantity: float
    trade_time: datetime
    maker_order_id: Optional[str] = None
    taker_order_id: Optional[str] = None
    
    def __post_init__(self):
        self.event_type = 'trade'

避坑指南： 我曾经在记录成交事件时，忘了区分 maker 和 taker。结果做回测时，手续费计算全错了。记住，maker 和 taker 的手续费率通常不一样，这个字段一定要保留。

4.4 BookUpdateEvent：盘口变了

BookUpdateEvent 是订单簿更新的核心事件。每当有新的订单进入、撤单或者成交，盘口数据就会变化，这时候就需要触发 BookUpdateEvent。

我建议把 BookUpdateEvent 设计成增量更新模式。什么意思？就是只传递变化的部分，而不是整个盘口数据。这样能大幅减少网络传输和内存开销。

@dataclass
class BookUpdateEvent(Event):
    """订单簿更新事件"""
    symbol: str
    bids: List[Tuple[float, float]]  # (price, quantity) 列表
    asks: List[Tuple[float, float]]
    update_type: str  # 'snapshot' or 'delta'
    sequence: int  # 序列号，用于检测数据连续性
    
    def __post_init__(self):
        self.event_type = 'book_update'

注意： 增量更新模式下，一定要带序列号。我见过一个系统，因为网络丢包导致盘口数据错乱，整个策略亏了三天才发现。序列号就是你的「数据完整性检查器」。

4.5 ErrorEvent：出错了怎么办

没有不出错的系统。关键是出错后怎么处理。

ErrorEvent 不是用来「记录日志」的，而是用来「触发恢复机制」的。比如交易所连接断了、数据解析失败、订单被拒——这些都应该抛出 ErrorEvent。

@dataclass
class ErrorEvent(Event):
    """错误事件"""
    error_code: str
    error_message: str
    source: str  # 错误来源，如 'exchange', 'parser', 'strategy'
    severity: str  # 'info', 'warning', 'critical'
    recoverable: bool = True  # 是否可恢复
    
    def __post_init__(self):
        self.event_type = 'error'

我个人习惯把 ErrorEvent 的 severity 分成三级：

  • info - 小问题，比如某次数据延迟，记录一下就行
  • warning - 需要关注，比如连续三次重连失败
  • critical - 必须停机，比如内存溢出、核心数据损坏

4.6 自定义事件：框架的扩展性

框架做得再好，也架不住业务需求千奇百怪。所以，一定要留出自定义事件的口子。

我的做法是定义一个 CustomEvent，里面放一个 event_name 字段，用户自己定义事件名称和数据内容。

@dataclass
class CustomEvent(Event):
    """自定义事件"""
    custom_type: str  # 用户自定义的事件名称
    payload: Dict[str, Any]  # 任意数据
    
    def __post_init__(self):
        self.event_type = f'custom_{self.custom_type}'

举个例子。假设你要监控系统延迟，可以定义一个 LatencyEvent：

# 使用自定义事件
latency_event = CustomEvent(
    event_id='lat_001',
    timestamp=datetime.now(),
    custom_type='latency_check',
    payload={
        'exchange_latency_ms': 12.5,
        'processing_latency_ms': 3.2,
        'total_latency_ms': 15.7
    }
)

4.7 事件类型关系图

下面这张图，展示了我们定义的所有事件类型之间的关系。你看一眼就能明白整个事件体系的架构。

事件类型体系结构 Event（基类） OrderEvent 订单状态变化 TradeEvent 成交记录 BookUpdateEvent 盘口更新 ErrorEvent 异常处理 CustomEvent（扩展） ORDER_CREATED ORDER_FILLED ORDER_CANCELLED ORDER_REJECTED 实线：继承关系 | 虚线：扩展关系 | 子事件示例仅展示 OrderEvent 部分

4.8 事件注册与分发

事件定义好了，怎么用？我习惯用一个事件注册表来管理所有事件类型。

class EventRegistry:
    """事件注册表"""
    _events: Dict[str, type] = {}
    
    @classmethod
    def register(cls, event_type: str, event_class: type):
        cls._events[event_type] = event_class
    
    @classmethod
    def get(cls, event_type: str) -> type:
        return cls._events.get(event_type)
    
    @classmethod
    def create_event(cls, event_type: str, **kwargs) -> Event:
        event_class = cls.get(event_type)
        if event_class is None:
            raise ValueError(f"未知事件类型: {event_type}")
        return event_class(**kwargs)

# 注册事件
EventRegistry.register('order', OrderEvent)
EventRegistry.register('trade', TradeEvent)
EventRegistry.register('book_update', BookUpdateEvent)
EventRegistry.register('error', ErrorEvent)

小技巧： 我建议在系统启动时，自动扫描所有事件类并注册。这样新增事件类型时，不用手动修改注册代码。用 Python 的 __init_subclass__ 或者元类都能实现。

4.9 总结一下

事件类型定义这件事，看着简单，但坑不少。我总结几个要点：

  • 基类要稳：event_id、timestamp、event_type 这三个字段必须有
  • 类型要全：订单、成交、盘口、错误，一个都不能少
  • 扩展要活：留好自定义事件的接口，别把自己框死
  • 注册要自动：别手动一个个注册，用自动化方式

事件类型定义好了，整个事件驱动框架的骨架就有了。接下来就是怎么让这些事件在系统里高效流转——那是下一节的内容了。

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

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