3、事件驱动架构：事件循环设计、事件类型定义、事件队列管理

说到回测框架的核心，我个人觉得事件驱动架构是绕不开的一个坎。说白了，你的策略怎么跟市场数据互动？订单怎么被处理？成交怎么反馈回来？这些都得靠事件来串联。我刚开始写回测框架时，把逻辑全写在一个大循环里，结果改一个需求就要动半套代码，那叫一个痛苦。

后来我学乖了——用事件驱动。你想想看，把整个回测过程拆成一个个独立的事件，每个模块只管自己那一亩三分地，耦合度一下子就降下来了。今天我们就来聊聊这个架构怎么落地。

3.1 事件循环设计——回测的“心脏”

事件循环，说白了就是一个永不停止的 while 循环，不断从队列里取事件、分发事件、处理事件。嗯，这里要注意，回测里的事件循环跟生产环境的高并发事件循环不太一样——我们是单线程、顺序执行的，因为回测的核心是“模拟时间推进”。

我一般这样设计事件循环的主逻辑：

class EventLoop:
    def __init__(self):
        self.queue = EventQueue()
        self.handlers = {}  # 事件类型 -> 处理函数列表
        self.running = False

    def register_handler(self, event_type, handler):
        if event_type not in self.handlers:
            self.handlers[event_type] = []
        self.handlers[event_type].append(handler)

    def start(self):
        self.running = True
        while self.running:
            event = self.queue.get()
            if event is None:
                break  # 终止信号
            self._dispatch(event)

    def _dispatch(self, event):
        handlers = self.handlers.get(type(event), [])
        for handler in handlers:
            handler(event)

    def stop(self):
        self.running = False
        self.queue.put(None)  # 发送终止信号

这段代码看着简单，但有几个坑。我曾经在项目里把 get() 写成了阻塞调用，结果回测跑完最后一个事件后，程序卡死在那了。后来加了个超时机制才解决。

3.2 事件类型定义——三大核心事件

回测里的事件类型其实不需要太多，三个核心事件就能覆盖 90% 的场景：BarEvent、OrderEvent、FillEvent。我见过有人定义十几个事件类型，结果自己都搞混了。保持简单，够用就行。

3.2.1 BarEvent——市场数据的载体

BarEvent 代表一根 K 线。它承载了时间、开盘价、最高价、最低价、收盘价、成交量这些基础数据。为什么叫 Bar 不叫 Kline？我个人习惯用 Bar，因为量化圈子里 Bar 更通用，跟 Bloomberg、Reuters 的术语一致。

from dataclasses import dataclass
from datetime import datetime

@dataclass
class BarEvent:
    symbol: str          # 合约代码
    timestamp: datetime  # 时间戳
    open: float
    high: float
    low: float
    close: float
    volume: float
    # 可选字段
    open_interest: float = 0.0  # 持仓量（期货用）
    vwap: float = 0.0           # 成交量加权均价

这里有个细节：timestamp 我建议用 datetime 而不是字符串。为什么？因为排序、比较、时区转换都方便。我曾经接手过一个项目，时间字段全是字符串，每次回测前都要先做一轮字符串转 datetime，那效率，啧啧。

3.2.2 OrderEvent——策略的“手”

OrderEvent 是策略向执行模块发出的指令。它告诉系统：我想在什么时间、以什么方向、什么价格、买卖多少手。

from enum import Enum

class OrderDirection(Enum):
    BUY = "BUY"
    SELL = "SELL"

class OrderType(Enum):
    MARKET = "MARKET"    # 市价单
    LIMIT = "LIMIT"      # 限价单
    STOP = "STOP"        # 止损单

@dataclass
class OrderEvent:
    order_id: str              # 唯一订单ID
    symbol: str
    direction: OrderDirection
    order_type: OrderType
    quantity: int
    price: float = 0.0         # 限价单/止损单的价格
    timestamp: datetime = None # 下单时间

3.2.3 FillEvent——成交的“回执”

FillEvent 是模拟交易所返回的成交确认。它告诉策略：你的订单被执行了，成交了多少、什么价格、花了多少手续费。

@dataclass
class FillEvent:
    order_id: str          # 对应的订单ID
    symbol: str
    fill_price: float      # 成交均价
    fill_quantity: int     # 成交数量
    commission: float      # 手续费
    fill_time: datetime    # 成交时间
    # 滑点信息（可选）
    slippage: float = 0.0  # 实际成交价与目标价的偏差

注意 fill_price 和 price 的区别。订单里的 price 是“我想以什么价格成交”，而 fill_price 是“实际以什么价格成交”。市价单的 price 是 0，但 fill_price 一定有个值。这个区分很重要，我见过有人把这两个混用，结果回测的滑点计算全错了。

3.3 事件队列管理——别让事件“打架”

事件队列是事件循环的“缓冲区”。所有模块产生的事件都往队列里塞，事件循环再从队列里取。队列管理的好坏，直接决定了回测的稳定性和性能。

3.3.1 队列的两种实现

特性	Python 内置 queue.Queue	自定义优先级队列
线程安全	是（内置锁）	需自行实现
排序能力	FIFO 先进先出	可按时间/优先级排序
性能	中等（有锁开销）	高（无锁或轻量锁）
适用场景	多线程回测	单线程高性能回测

我个人习惯在单线程回测里用自定义优先级队列。因为回测是顺序执行的，不需要线程安全，但需要按时间排序——Bar 事件必须按时间先后处理，不能乱序。

import heapq
from datetime import datetime

class PriorityEventQueue:
    """基于堆的优先级事件队列，按时间排序"""
    def __init__(self):
        self._heap = []
        self._counter = 0  # 用于相同时间戳时的稳定排序

    def put(self, event):
        # 以时间戳为主键，counter 为次键
        heapq.heappush(self._heap, (event.timestamp, self._counter, event))
        self._counter += 1

    def get(self):
        if self._heap:
            _, _, event = heapq.heappop(self._heap)
            return event
        return None

    def peek(self):
        """查看下一个事件但不取出"""
        if self._heap:
            return self._heap[0][2]
        return None

    def is_empty(self):
        return len(self._heap) == 0

3.3.2 队列的容量管理

回测过程中，事件队列可能会暴涨。比如你跑 10 年的分钟级数据，Bar 事件就有 50 多万个。如果策略在每根 Bar 上都下单，OrderEvent 和 FillEvent 也会跟着膨胀。

我建议给队列设置一个最大容量，防止内存溢出：

class BoundedPriorityEventQueue(PriorityEventQueue):
    def __init__(self, max_size=100000):
        super().__init__()
        self.max_size = max_size

    def put(self, event):
        if len(self._heap) >= self.max_size:
            raise QueueFullError("事件队列已满，请检查是否事件消费速度跟不上")
        super().put(event)

嗯，这里要注意：队列满不一定是 bug，也可能是你的策略逻辑有问题——比如在某个时间点疯狂下单，但成交模块处理不过来。我曾经遇到过一个策略，在开盘瞬间下了 10000 个订单，直接把队列撑爆了。后来加了限流才解决。

3.4 事件驱动架构的整体流程

说了这么多，我们来看看事件驱动架构在回测里到底怎么流转的。下面这张图是我自己画的一个标准流程：

这个流程的核心思想是：所有事件都走队列，所有模块都通过事件来通信。数据源产生 BarEvent，策略消费 BarEvent 并产生 OrderEvent，执行模块消费 OrderEvent 并产生 FillEvent，FillEvent 再反馈给策略更新持仓。整个闭环就转起来了。

3.5 避坑指南

最后，分享几个我在实际项目中踩过的坑，希望能帮你少走弯路：

• 事件类型用字符串还是类？ 我建议用类。字符串虽然简单，但后期维护时根本不知道有哪些事件类型。用类的话，IDE 可以自动补全，类型检查也能帮你发现错误。
• 事件处理函数不要阻塞。 如果某个 handler 执行时间太长，整个事件循环都会被卡住。我遇到过策略里做了个复杂的机器学习预测，结果一根 Bar 处理了 5 秒，回测速度直接崩了。
• 小心事件循环的“死锁”。 比如策略在处理 BarEvent 时又往队列里塞了 100 个 OrderEvent，然后等着 FillEvent 回来才继续。但 FillEvent 需要执行模块处理 OrderEvent 才能产生，而执行模块还在等事件循环调度...这就死锁了。解决办法是：策略不要同步等待成交结果。
• 日志一定要打。 每个事件进出队列都打一条日志，回测出问题时能快速定位。我习惯用 logging.debug 级别，不影响性能，需要时再打开。