3. 事件驱动引擎：让回测系统“活”起来
做量化回测，最怕什么？
我最怕的是策略逻辑和市场数据搅在一起，改一个参数就得重写整个流程。后来我接触了事件驱动架构，说白了，就是把系统拆成一个个“反应单元”——市场来了新数据，系统就触发一个事件；策略想下单，也发一个事件。各模块各干各的，互不干扰。
这一章，我们就来亲手搭建这个核心引擎。
3.1 事件循环机制：系统的“心跳”
事件驱动引擎的核心，是一个永不停止的循环。它就像人的心跳，一下一下地检查有没有新事件需要处理。
核心逻辑：
- 从事件队列里取出一个事件
- 判断事件类型
- 交给对应的处理器去执行
- 处理完，继续取下一个
我在早期版本里犯过一个低级错误——循环里忘了加退出条件。结果回测跑完了，程序还在空转，CPU 直接拉满。嗯，这里要注意：事件循环必须能优雅地终止。
class EventLoop:
def __init__(self):
self.queue = Queue()
self.running = False
def start(self):
self.running = True
while self.running:
event = self.queue.get()
if event is None: # 退出信号
break
self._dispatch(event)
def stop(self):
self.running = False
self.queue.put(None) # 塞一个退出事件你想想看，这个 None 信号是不是很巧妙？它既能让循环退出，又不会阻塞在 queue.get() 上。
3.2 事件类型定义：三类核心事件
一个回测系统里，事件其实就三大类。我习惯把它们定义成枚举，清晰又好扩展。
| 事件类型 | 触发时机 | 携带数据 |
|---|---|---|
| 市场事件 | 新 K 线/ tick 到达 | 时间、开盘价、收盘价、成交量等 |
| 信号事件 | 策略产生买卖信号 | 标的、方向、置信度 |
| 订单事件 | 下单、成交、撤单 | 订单 ID、价格、数量、状态 |
我的习惯：每个事件都带一个 timestamp 字段。为什么？因为回测里时间就是一切。没有时间戳，你根本没法对齐市场数据和策略信号。
from enum import Enum, auto
from dataclasses import dataclass
from datetime import datetime
class EventType(Enum):
MARKET = auto()
SIGNAL = auto()
ORDER = auto()
@dataclass
class Event:
type: EventType
timestamp: datetime
data: dict我曾经遇到过一个坑：信号事件和订单事件的时间戳不一致，导致回测结果完全对不上。后来我强制所有事件的时间戳都从同一个时钟源获取，问题才解决。
3.3 事件队列管理：别让事件“堵车”
事件队列，说白了就是一个先进先出的管道。但这里有个关键问题：优先级。
市场事件必须优先处理，因为它驱动着整个回测的推进。订单事件可以稍微排后，毕竟成交确认晚一两个 tick 影响不大。
避坑指南：我曾经用过一个简单的 list 当队列，结果事件一多，插入和删除的性能直线下降。后来换成了 collections.deque，性能提升了 10 倍不止。
from collections import deque
from heapq import heappush, heappop
class PriorityEventQueue:
def __init__(self):
self._market_queue = deque()
self._signal_queue = deque()
self._order_queue = deque()
def put(self, event: Event):
if event.type == EventType.MARKET:
self._market_queue.append(event)
elif event.type == EventType.SIGNAL:
self._signal_queue.append(event)
else:
self._order_queue.append(event)
def get(self) -> Event:
# 市场事件优先
if self._market_queue:
return self._market_queue.popleft()
if self._signal_queue:
return self._signal_queue.popleft()
if self._order_queue:
return self._order_queue.popleft()
return None你可能会问：为什么不用一个统一的优先级队列？嗯，我试过。但实际跑起来发现，市场事件和信号事件的优先级差距太大，混在一起反而增加了排序开销。拆成三个独立队列，反而更清爽。
3.4 完整的事件驱动流程
把上面三个部分串起来，就是一套完整的事件驱动引擎。我画了一张流程图，帮你理清关系。
整个流程其实就三步：
- 数据进来：市场数据源产生市场事件，塞进队列
- 循环调度：事件循环不断从队列里取事件，交给分发器
- 分发处理：分发器根据事件类型，扔给对应的处理器
处理器处理完后，可能会产生新的事件（比如策略处理器产生信号事件），再塞回队列。这样就形成了一个闭环。
性能优化小技巧：如果回测的数据量很大（比如 10 年 tick 级数据），可以考虑用多线程。一个线程专门负责从数据源读取并生成事件，另一个线程负责事件循环和处理。但要注意线程安全问题，我一般用 queue.Queue 来保证。
好了，事件驱动引擎的核心骨架已经搭好了。下一节我们会在这个骨架上，填上具体的策略逻辑和订单管理模块。到时候你会发现，这套架构的扩展性真的非常强。
公众号:蓝海资料掘金营，微信deep3321