第一章 事件驱动引擎：事件循环机制、事件类型定义、事件队列管理

大家好，我是老张。今天咱们聊聊回测系统里最核心的骨架——事件驱动引擎。

说实话，我刚开始做量化交易系统时，也踩过不少坑。最早我写回测，就是简单的循环遍历K线，每个tick算一下持仓盈亏。结果呢？策略稍微复杂点，比如同时处理多个品种、不同时间周期的信号，代码就乱成一锅粥了。

后来我才明白，回测系统本质上就是个事件驱动的状态机。你想想看，真实交易中，我们不就是根据行情事件、订单事件、时间事件来做决策的吗？回测也一样。

1.1 事件循环机制——系统的“心脏”

事件循环，说白了就是一个永不停止的while循环。它不停地问：有没有新事件？有就处理，没有就等着。

我习惯把事件循环比作餐厅的传菜员。厨师（数据源）做好菜（事件），传菜员（事件循环）端给对应的客人（事件处理器）。客人吃完了，传菜员再回来等下一道菜。

核心代码其实很简单：

class EventLoop:
    def __init__(self):
        self._running = False
        self._queue = EventQueue()
    
    def run(self):
        self._running = True
        while self._running:
            event = self._queue.get()
            if event is None:
                continue
            self._dispatch(event)
    
    def stop(self):
        self._running = False

嗯，这里要注意：get()方法一定要有超时机制。我曾经在早期版本里用了阻塞队列，结果回测跑完想停都停不下来，只能强制杀进程。后来改成了带超时的非阻塞获取，优雅多了。

关键点：事件循环必须支持优雅关闭。别让用户等得心焦。

1.2 事件类型定义——给事件“贴标签”

事件类型，就是给每个事件一个身份标识。我见过有人用字符串，有人用枚举。我个人强烈推荐用枚举，原因很简单：

  • 字符串容易拼写错误，比如 "MARKET_DATA" 写成 "MARKET_DATA "（多了个空格），排查起来想死的心都有
  • 枚举有IDE自动补全，写起来爽
  • 枚举可以附加额外信息，比如优先级

来看看我项目里常用的定义：

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()  # 撤单
    ORDER_REJECT = auto()  # 拒单
    
    # 系统事件
    TIMER = auto()         # 定时器
    SHUTDOWN = auto()      # 关闭信号
    ERROR = auto()         # 错误事件

为什么要有 ERROR 事件？我记得有一次回测跑了一整天，结果因为某个数据源异常，整个系统直接崩溃了。从那以后，我强制要求所有异常都必须包装成事件，由事件循环统一处理。这样至少能优雅地记录日志、保存中间状态。

小技巧：事件类型最好按功能分组，比如行情类、订单类、系统类。这样在事件分发时，可以快速路由到对应的处理器。

1.3 事件队列管理——别让事件“堵车”

事件队列，就是事件的缓冲区。它解决了一个核心问题：生产者和消费者的速度不匹配。

比如行情数据源可能一秒产生1000个tick，但策略处理一个tick需要5毫秒。如果没有队列，策略根本跟不上。有了队列，数据源只管往队列里塞，策略按自己的节奏处理。

我常用的队列实现有两种：

类型 特点 适用场景
FIFO队列 先进先出，保证时序 普通回测，按时间顺序处理
优先级队列 按优先级出队 需要紧急处理的事件（如风控信号）

大多数情况下，FIFO就够了。但如果你做的是高频回测，我建议用优先级队列。为什么？因为风控事件必须优先处理。我曾经见过一个策略，因为行情事件堆积太多，风控信号被堵在后面，结果回测爆仓了——虽然只是模拟，但看着也心疼啊。

来看看队列的核心实现：

import heapq
from threading import Lock

class EventQueue:
    def __init__(self, priority=False):
        self._queue = []
        self._priority = priority
        self._lock = Lock()
    
    def put(self, event):
        with self._lock:
            if self._priority:
                # 优先级队列，按事件类型的优先级入队
                heapq.heappush(self._queue, (event.type.value, event))
            else:
                self._queue.append(event)
    
    def get(self, timeout=0.1):
        with self._lock:
            if self._queue:
                if self._priority:
                    return heapq.heappop(self._queue)[1]
                else:
                    return self._queue.pop(0)
        return None

避坑指南：我曾经在队列里直接用了Python的list做FIFO，pop(0)的时间复杂度是O(n)。回测数据量一大，性能直接崩了。后来改用collections.deque，O(1)的复杂度，舒服多了。

1.4 完整的事件驱动流程

把上面三个部分串起来，就是完整的事件驱动流程。我画了张图，方便你理解：

数据源 行情/订单/定时 产生事件 事件队列 FIFO / 优先级 线程安全 取出事件 事件循环 while running 分发事件 路由 行情处理器 更新K线/指标 订单处理器 更新持仓/资金 风控处理器 检查风险指标 策略 决策

流程其实很清晰：

  1. 数据源产生事件（行情、订单、定时等）
  2. 事件进入队列，等待处理
  3. 事件循环从队列中取出事件
  4. 根据事件类型，路由到对应的处理器
  5. 处理器更新状态，或触发策略决策
  6. 策略决策产生新事件（如订单），重新进入队列

你看，这就是个闭环。数据进来，策略出去，周而复始。

核心思想：事件驱动引擎的本质，就是把“什么时候做什么事”这个复杂问题，拆解成“发生什么事就处理什么事”的简单模型。每个事件都是独立的，互不干扰。这样你的策略逻辑就清晰了，调试也方便。

1.5 实际项目中的经验

最后分享几个我在实战中总结的经验：

  • 事件要轻量：事件对象里只放必要的数据，别把整个DataFrame塞进去。我见过有人把100MB的行情数据当事件传，结果队列直接OOM。
  • 队列要有容量限制：设置最大长度，超过就丢弃旧事件或阻塞生产者。不然内存会爆。
  • 日志要详细：每个事件处理前后都打日志。回测出问题时，日志就是你的救命稻草。
  • 性能测试不能少：用真实数据量压一下，看看事件循环能不能扛住。我一般要求100万事件/秒的处理能力。

好了，事件驱动引擎的核心就这些。说白了，它就是回测系统的骨架。骨架搭好了，后面加策略、加指标、加风控，都顺理成章。

下一章咱们聊聊数据源和回测引擎的对接，到时候见。

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

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