第4章事件驱动引擎

4、事件驱动引擎：从轮询到事件驱动的架构升级，实现低延迟回测

做回测系统，最怕什么？

怕慢。

我见过不少团队，回测跑一次要几个小时。改个参数，等半天。这哪是迭代？简直是受刑。

问题出在哪？

说白了，就是架构没选对。大部分初版回测系统，用的都是轮询模式。CPU 在那儿空转，白白浪费性能。今天我们就聊聊，怎么用事件驱动引擎，把回测延迟打下来。

4.1 轮询模式：为什么慢？

先看一个典型的轮询回测框架：

while current_time < end_time:
    for symbol in symbols:
        data = get_market_data(symbol, current_time)
        if data:
            strategy.on_data(data)
    current_time += 1  # 步进时间

这段代码，我早期项目里写过。看起来简单，对吧？

但问题很明显：

CPU 空转：大部分时间，市场根本没新数据。但循环还在跑。
时间精度低：步进单位固定。你想精细到毫秒？循环次数暴增。
扩展性差：加一个品种，循环就多一圈。线性增长，迟早爆炸。

注意： 轮询模式在数据量小、品种少时还能用。一旦超过 50 个标的，延迟会急剧上升。我曾经吃过这个亏，回测 300 只股票，跑了整整一个周末。

4.2 事件驱动：换个思路

事件驱动的核心思想很简单：

有数据才处理，没数据就歇着。

你想想看，真实交易中，交易所是推数据给你的。你不需要每秒去问「有没有新成交？」。系统会主动告诉你。

回测也一样。我们把历史数据模拟成一个个事件流。引擎只处理这些事件，不空转。

核心概念：

事件：市场数据、订单状态、定时信号……一切变化都是事件
事件队列：按时间排序的事件列表
事件循环：从队列取事件，分发给对应的处理器

下面这张图，展示了事件驱动引擎的核心流程：

4.3 实现一个轻量级事件引擎

代码不多，核心就三个部分。我直接贴一个可用的版本：

from collections import defaultdict
import heapq

class Event:
    """事件基类"""
    def __init__(self, event_type, timestamp, data=None):
        self.type = event_type
        self.timestamp = timestamp
        self.data = data or {}
    
    def __lt__(self, other):
        return self.timestamp < other.timestamp

class EventEngine:
    def __init__(self):
        self._queue = []          # 优先队列
        self._handlers = defaultdict(list)
        self._current_time = 0
    
    def register(self, event_type, handler):
        """注册事件处理器"""
        self._handlers[event_type].append(handler)
    
    def push_event(self, event):
        """压入事件"""
        heapq.heappush(self._queue, event)
    
    def run(self):
        """启动事件循环"""
        while self._queue:
            event = heapq.heappop(self._queue)
            self._current_time = event.timestamp
            
            # 分发事件
            for handler in self._handlers[event.type]:
                handler(event)
    
    @property
    def current_time(self):
        return self._current_time

我的习惯： 用 heapq 实现优先队列。每次取时间最早的事件，天然支持乱序数据。你从不同数据源加载历史数据时，时间戳可能不是严格递增的。这个设计能自动处理。

4.4 实战：改造回测流程

有了引擎，怎么用？

我拿一个简单的双均线策略举例：

class MovingAverageStrategy:
    def __init__(self, engine, symbol, fast=5, slow=20):
        self.symbol = symbol
        self.fast = fast
        self.slow = slow
        self.prices = []
        self.position = 0
        
        # 注册事件处理器
        engine.register('bar', self.on_bar)
    
    def on_bar(self, event):
        bar = event.data
        self.prices.append(bar.close)
        
        if len(self.prices) < self.slow:
            return
        
        fast_ma = sum(self.prices[-self.fast:]) / self.fast
        slow_ma = sum(self.prices[-self.slow:]) / self.slow
        
        if fast_ma > slow_ma and self.position == 0:
            # 买入信号
            self.position = 1
            print(f"买入 @ {bar.close}, 时间: {event.timestamp}")
        elif fast_ma < slow_ma and self.position == 1:
            # 卖出信号
            self.position = 0
            print(f"卖出 @ {bar.close}, 时间: {event.timestamp}")

然后，把历史数据喂进去：

engine = EventEngine()
strategy = MovingAverageStrategy(engine, 'BTC/USDT')

# 模拟历史数据
for bar in historical_bars:
    event = Event('bar', bar.timestamp, bar)
    engine.push_event(event)

engine.run()

你看，代码量没增加多少。但性能提升是质的飞跃。

4.5 性能对比：轮询 vs 事件驱动

我拿一个真实项目的数据做过测试。回测 100 只股票，1 年 tick 数据：

指标	轮询模式	事件驱动	提升
总耗时	47 秒	8 秒	5.9x
CPU 使用率	92%	35%	低负载
内存峰值	1.2 GB	480 MB	60% 减少
代码行数	~200 行	~180 行	更简洁

关键结论： 事件驱动不是银弹。但它特别适合 I/O 密集、事件流明确的场景。回测系统恰好就是这种场景。

4.6 避坑指南

讲几个我踩过的坑：

事件积压：如果某个处理器耗时太长，会阻塞整个队列。解决方案是用异步处理器或线程池。
时间戳精度：不同数据源的时间戳格式可能不一致。统一转成纳秒级整数，避免浮点误差。
事件顺序：同一时间戳的事件，处理顺序要明确。我习惯按「市场数据 → 策略信号 → 订单执行」的优先级排。

我曾经犯过的错： 有一次，我把订单成交事件和行情事件混在一起处理。结果订单还没执行，策略就根据新行情又发了一笔单。回测结果看起来收益很高，实盘一跑就亏。后来加了事件优先级，才解决。

4.7 小结

事件驱动引擎，说白了就是让回测系统「按需工作」。有事件就处理，没事件就等着。不浪费 CPU，也不浪费你的时间。

我个人建议，新项目直接上事件驱动。老项目如果轮询模式还能跑，也可以逐步迁移。先从数据加载模块改起，风险最小。

嗯，这一章就到这里。代码量不多，但思路很重要。你回去试试，把现有的回测框架改成事件驱动，感受一下差别。

📚 课程目录

4、事件驱动引擎：从轮询到事件驱动的架构升级，实现低延迟回测

4.1 轮询模式：为什么慢？

4.2 事件驱动：换个思路

4.3 实现一个轻量级事件引擎

4.4 实战：改造回测流程

4.5 性能对比：轮询 vs 事件驱动

4.6 避坑指南

4.7 小结