第1章事件驱动架构概述

一、事件驱动架构概述

做量化回测这些年，我接触过不少架构模式。说实话，事件驱动架构是我个人最偏爱的一种。为什么？因为它跟真实市场的运作方式太像了。

你想想看，真实市场里发生了什么？订单来了、成交了、行情变了——这些都是一个个事件。事件驱动架构，说白了就是模拟这种自然状态。

什么是事件驱动架构

事件驱动架构，核心就一句话：系统响应事件，而不是主动轮询。

传统程序是「你问我答」——你调用一个函数，它返回结果。事件驱动是「你告诉我，我处理」——事件发生了，系统自动触发对应的处理逻辑。

举个例子。你写一个简单的回测：

# 传统方式：循环处理每个时间点
for bar in bars:
    if bar.timestamp == '2024-01-01 09:30:00':
        handle_open(bar)
    elif bar.timestamp == '2024-01-01 09:31:00':
        handle_next(bar)
    # ... 每个时间点都要检查

换成事件驱动：

# 事件驱动方式：事件来了就处理
def on_bar(bar):
    # 直接处理，不用判断时间
    update_indicators(bar)
    check_signals(bar)

# 事件队列自动分发
event_queue.push(BarEvent('2024-01-01 09:30:00', data))
event_queue.push(BarEvent('2024-01-01 09:31:00', data))

看出区别了吗？传统方式里，你得自己控制「什么时候该做什么」。事件驱动里，事件本身告诉你「该做什么了」。

核心要点：事件驱动架构的本质是「控制反转」——不是你的代码去拉数据，而是数据推给你的代码。

为什么用于回测

我在做第一个实盘系统时，用的就是事件驱动。后来做回测，发现天然适配。原因有三：

回测就是模拟真实市场。市场里每一笔成交、每一个报价都是事件。回测用事件驱动，逻辑跟实盘一模一样。代码写一遍，两边都能跑。
解耦能力强。策略、风控、成交、记录——每个模块只关心自己感兴趣的事件。加一个新模块，不用改旧代码。我曾经在回测系统里加了一个「滑点分析器」，只注册了一个事件监听，半小时搞定。
性能可优化。事件队列可以批量处理、可以异步、可以并行。我优化过一个回测系统，把单线程事件处理改成多线程，速度提升了4倍。

我的经验：事件驱动回测最大的好处是「回测即实盘」。你写的策略代码，在回测和实盘里跑的是同一套逻辑。少了很多「回测赚钱实盘亏」的坑。

核心组件

事件驱动架构就三个核心东西：事件、事件队列、事件处理器。我一个个说。

1. 事件（Event）

事件就是「发生了什么事」。在回测里，常见的事件有：

Bar事件：新的K线来了
Tick事件：新的行情报价来了
订单事件：你下了个单
成交事件：订单成交了
信号事件：策略发出了买卖信号

每个事件就是一个对象，包含事件类型和数据。我习惯这样定义：

class Event:
    """事件基类"""
    def __init__(self, event_type, timestamp, data):
        self.type = event_type      # 事件类型，比如 'BAR', 'TICK'
        self.timestamp = timestamp  # 事件发生时间
        self.data = data            # 事件携带的数据

# 具体事件
class BarEvent(Event):
    def __init__(self, timestamp, open, high, low, close, volume):
        super().__init__('BAR', timestamp, {
            'open': open,
            'high': high,
            'low': low,
            'close': close,
            'volume': volume
        })

2. 事件队列（Event Queue）

事件队列就是事件的「中转站」。所有事件先放到队列里，然后按顺序分发给处理器。

队列的核心要求：先进先出。回测里时间顺序很重要，不能乱。

from queue import Queue
from collections import deque

class EventQueue:
    """事件队列 - 用deque实现，性能更好"""
    def __init__(self):
        self.queue = deque()
    
    def push(self, event):
        self.queue.append(event)
    
    def pop(self):
        if self.queue:
            return self.queue.popleft()
        return None
    
    def is_empty(self):
        return len(self.queue) == 0

注意：我曾经犯过一个错——用Python的list当队列，用pop(0)取事件。数据量小没问题，但回测几百万个事件时，性能直接崩了。list的pop(0)是O(n)操作，deque的popleft()是O(1)。

3. 事件处理器（Event Handler）

事件处理器就是「处理事件的人」。每个处理器注册自己关心的事件类型，事件来了就调用对应的处理函数。

class EventHandler:
    """事件处理器基类"""
    def __init__(self):
        self.handlers = {}  # 事件类型 -> 处理函数
    
    def register(self, event_type, handler_func):
        """注册事件处理函数"""
        self.handlers[event_type] = handler_func
    
    def handle(self, event):
        """处理事件"""
        handler = self.handlers.get(event.type)
        if handler:
            handler(event)
        else:
            # 不关心的事件，忽略
            pass

# 具体处理器
class StrategyHandler(EventHandler):
    def __init__(self, strategy):
        super().__init__()
        self.strategy = strategy
        # 注册关心的事件
        self.register('BAR', self.on_bar)
        self.register('TICK', self.on_tick)
    
    def on_bar(self, event):
        """处理Bar事件"""
        signal = self.strategy.calculate(event.data)
        if signal:
            # 产生信号事件，放回队列
            event_queue.push(SignalEvent(event.timestamp, signal))
    
    def on_tick(self, event):
        """处理Tick事件"""
        # 更新策略状态
        self.strategy.update_market(event.data)

整体架构图

下面这张图展示了事件驱动回测的核心流程。我画得比较简洁，但该有的都有了。

这张图展示了完整的事件流：数据源产生原始事件 → 事件队列暂存 → 分发器按类型分发 → 各处理器处理 → 处理器可能产生新事件放回队列。如此循环，直到所有事件处理完毕。

一个完整的回测循环

最后，我写一个最简单的回测主循环，让你看看这些组件怎么配合：

class BacktestEngine:
    """回测引擎"""
    def __init__(self):
        self.event_queue = EventQueue()
        self.handlers = []
    
    def add_handler(self, handler):
        self.handlers.append(handler)
    
    def run(self, data_source):
        # 1. 加载数据，生成初始事件
        for bar in data_source.load_bars():
            event = BarEvent(bar.timestamp, bar.open, bar.high, 
                           bar.low, bar.close, bar.volume)
            self.event_queue.push(event)
        
        # 2. 事件处理循环
        while not self.event_queue.is_empty():
            event = self.event_queue.pop()
            
            # 3. 分发给所有处理器
            for handler in self.handlers:
                handler.handle(event)
            
            # 4. 处理器可能产生新事件
            # 新事件已经在处理器内部push到队列了
        
        print("回测完成")

嗯，这就是事件驱动架构的核心。说白了就是：事件来了，处理它；处理过程中可能产生新事件，继续处理。简单，但强大。

避坑指南：我曾经在事件循环里忘记处理「事件风暴」——某个处理器疯狂产生新事件，导致队列无限增长。后来加了一个最大事件数限制，超过就报错。建议你也加上。

📚 课程目录

一、事件驱动架构概述

什么是事件驱动架构

为什么用于回测

核心组件

1. 事件（Event）

2. 事件队列（Event Queue）

3. 事件处理器（Event Handler）

整体架构图

一个完整的回测循环