4、事件驱动引擎：事件循环、事件类型定义、事件处理器注册

好，咱们今天聊聊事件驱动引擎。这东西说白了，就是回测框架的「心脏」。没有它，你的策略就是个死程序，动不了。

我个人习惯把事件驱动引擎比作一个「消息中转站」。市场产生数据，引擎收到后，分发给对应的策略模块。策略做出决策，引擎再把这个决策转给执行模块。一环扣一环。

4.1 事件循环——引擎的永动机

事件循环，就是那个一直转啊转的循环。它不停问：「有事件来了吗？有吗？有吗？」

我在项目中遇到过一个问题：循环跑得太快，CPU 占用率飙到 100%。后来加了 sleep 控制，才压下来。你想想看，高频回测里，时间精度是微秒级的，但 CPU 也不能当电暖器用啊。

核心逻辑其实很简单：

class EventLoop:
    def __init__(self):
        self._running = False
        self._queue = []  # 事件队列

    def start(self):
        self._running = True
        while self._running:
            if self._queue:
                event = self._queue.pop(0)
                self._dispatch(event)
            else:
                # 没事件时，稍微歇一歇
                time.sleep(0.001)  # 1ms 的休眠

    def stop(self):
        self._running = False

    def put_event(self, event):
        self._queue.append(event)

嗯，这里要注意：pop(0) 在 Python 里是 O(n) 的操作。数据量大了会卡。我建议用 collections.deque 替代，它的 popleft 是 O(1) 的。

4.2 事件类型定义——给事件贴标签

事件不能乱糟糟的。你得给它们分门别类。就像快递到了，你得知道这是文件还是生鲜，对吧？

我一般用枚举来定义事件类型：

from enum import Enum, auto

class EventType(Enum):
    # 市场数据事件
    TICK = auto()          # 逐笔成交
    BAR = auto()           # K线数据
    ORDER_BOOK = auto()    # 订单簿快照

    # 交易事件
    ORDER_NEW = auto()     # 新订单
    ORDER_FILL = auto()    # 订单成交
    ORDER_CANCEL = auto()  # 订单撤销

    # 系统事件
    TIMER = auto()         # 定时器
    SHUTDOWN = auto()      # 关闭信号

每个事件类型对应一个数据结构。比如 TICK 事件：

@dataclass
class TickEvent:
    event_type: EventType = EventType.TICK
    symbol: str = ''
    price: float = 0.0
    volume: int = 0
    timestamp: int = 0

为什么用 dataclass？因为省事。你不用写一堆 __init__ 方法。而且它自带 __repr__，调试时直接 print 就能看到所有字段。

4.3 事件处理器注册——谁该干活？

事件来了，谁处理？你不能让所有模块都去抢。得有个「注册表」，告诉引擎：这个事件类型，由这个函数处理。

说白了，就是观察者模式。我早期做回测框架时，用的是硬编码的 if-else：

# 别学我，这是反面教材
def dispatch(self, event):
    if event.type == 'TICK':
        self.handle_tick(event)
    elif event.type == 'BAR':
        self.handle_bar(event)
    # ... 写了几十个 elif

后来发现，每加一个新策略，就得改 dispatch 函数。太蠢了。现在我用注册表：

class EventDispatcher:
    def __init__(self):
        self._handlers = {}  # {EventType: [handler1, handler2, ...]}

    def register(self, event_type: EventType, handler):
        """注册事件处理器"""
        if event_type not in self._handlers:
            self._handlers[event_type] = []
        self._handlers[event_type].append(handler)

    def unregister(self, event_type: EventType, handler):
        """注销事件处理器"""
        if event_type in self._handlers:
            self._handlers[event_type].remove(handler)

    def dispatch(self, event):
        """分发事件到所有注册的处理器"""
        handlers = self._handlers.get(event.event_type, [])
        for handler in handlers:
            handler(event)

这样，策略模块只需要调用 register(EventType.TICK, my_tick_handler) 就行了。引擎完全不知道策略内部怎么处理，解耦得干干净净。

4.4 完整的事件驱动流程

咱们把上面三个部分串起来，看看整体长什么样：

class EventEngine:
    def __init__(self):
        self.loop = EventLoop()
        self.dispatcher = EventDispatcher()

    def start(self):
        self.loop.start()

    def stop(self):
        self.loop.stop()

    def put(self, event):
        self.loop.put_event(event)

    def register(self, event_type, handler):
        self.dispatcher.register(event_type, handler)

    def _dispatch(self, event):
        self.dispatcher.dispatch(event)

使用起来也很直观：

engine = EventEngine()

# 注册一个 tick 处理器
def on_tick(tick_event):
    print(f"收到 tick: {tick_event.symbol} @ {tick_event.price}")

engine.register(EventType.TICK, on_tick)

# 模拟市场数据
tick = TickEvent(symbol='BTC/USDT', price=50000.0, volume=1.5, timestamp=1234567890)
engine.put(tick)

# 启动引擎
engine.start()

你想想看，这个架构是不是很干净？市场数据源只管往引擎里塞事件，策略只管注册自己的处理器。两边互不干扰。

4.5 事件驱动引擎的核心流程图

下面这张图，是我用 SVG 画的。它展示了事件从产生到被处理的完整路径：

从这张图你能看到，数据从左边进来，经过队列缓冲，事件循环不断取出，交给分发器。分发器查注册表，找到对应的处理器列表，挨个调用。最后策略逻辑执行。

整个流程是单向的、清晰的。没有循环依赖，没有回调地狱。这就是事件驱动架构的魅力所在。

好了，事件驱动引擎的核心就这些。记住三个关键词：循环、类型、注册。把这三样东西搭好，你的回测框架就有了坚实的底座。