1. 事件驱动架构基础

好，咱们直接进入正题。事件驱动架构，说白了就是一套「发生了什么事，就触发什么动作」的机制。我刚开始接触这个概念时，觉得它不就是个回调函数吗？后来在实盘系统里踩过坑才明白——没那么简单。

什么是事件驱动架构

想象一下你坐在交易大厅里。行情来了，你下单；成交了，你更新仓位；风控触发了，你暂停交易。每一件事都触发一个动作，这就是事件驱动。

传统程序是「你问我答」的模式：

# 轮询模式 - 低效
while True:
    price = get_price()
    if price < threshold:
        execute_order()

事件驱动是「你来了我响应」：

# 事件驱动模式 - 高效
def on_price_update(price):
    if price < threshold:
        execute_order()

event_bus.subscribe('price_update', on_price_update)

区别在哪？前者在空转，后者在等待。你想想看，回测跑十年数据，轮询模式得空转多少次？

为什么回测需要事件驱动

这个问题我问过不少同行。有人觉得「我就用for循环遍历K线，不也挺好？」嗯，对于简单策略确实够用。但一旦涉及以下场景，你就知道事件驱动的价值了：

• 多资产同时交易：A股和港股的开盘时间不同，用循环你得自己管理时间轴
• 订单执行细节：限价单可能部分成交、可能滑点，事件驱动能精确模拟
• 外部事件干扰：比如新闻公告、分红除权，这些不是按K线节奏来的

我在项目中遇到过一件事。有个同事用循环回测跑得好好的，换到实盘就崩了。为什么？因为实盘里订单成交是异步的，他的循环逻辑根本处理不了「先下单、后成交、中间行情变了」这种场景。后来改成事件驱动，问题迎刃而解。

核心组件拆解

事件驱动架构有三个核心零件：事件、事件队列、事件处理器。咱们一个一个说。

1. 事件（Event）

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

事件类型	触发时机	携带数据
BarEvent	新K线生成	时间、开盘价、收盘价、成交量...
TickEvent	新逐笔成交	时间、价格、成交量、买卖方向...
OrderEvent	策略下单	标的、方向、数量、价格类型...
FillEvent	订单成交	成交价、成交量、手续费...
SignalEvent	策略产生信号	标的、方向、置信度...

我个人习惯把事件设计成不可变对象。为什么？因为事件在队列里传递时，谁改了它都会出问题。我曾经被一个同事改了事件里的价格字段，排查了两天才发现——从那以后，所有事件我都用 namedtuple 或者 dataclass(frozen=True)。

from dataclasses import dataclass
from datetime import datetime

@dataclass(frozen=True)
class BarEvent:
    symbol: str
    timestamp: datetime
    open: float
    high: float
    low: float
    close: float
    volume: float

2. 事件队列（Event Queue）

事件队列就是事件的「传送带」。它负责把事件从生产者送到消费者手里。在回测里，这个队列必须是有序的——时间不能乱。

我见过有人用 Python 的 queue.Queue 做回测队列。嗯，这里要注意：queue.Queue 是线程安全的，但回测通常是单线程的。用它的代价是性能损耗。我个人推荐用 collections.deque，或者干脆用 heapq 实现一个优先级队列——按时间戳排序。

import heapq
from dataclasses import dataclass, field
from typing import Any

@dataclass(order=True)
class PrioritizedEvent:
    timestamp: float
    event: Any = field(compare=False)

class EventQueue:
    def __init__(self):
        self._queue = []
    
    def push(self, event):
        heapq.heappush(self._queue, event)
    
    def pop(self):
        return heapq.heappop(self._queue) if self._queue else None
    
    def peek(self):
        return self._queue[0] if self._queue else None
    
    def is_empty(self):
        return len(self._queue) == 0

3. 事件处理器（Event Handler）

事件处理器就是「谁来处理这个事件」。在回测架构里，每个组件都是一个处理器：

• 数据处理器：把原始行情转成标准事件
• 策略处理器：收到行情事件，产生交易信号
• 风控处理器：检查信号是否合规
• 执行处理器：把信号转成订单，模拟成交
• 组合处理器：更新持仓和资金

每个处理器只关心自己订阅的事件类型。这种解耦设计，说白了就是「各扫门前雪」。策略不用管订单怎么成交，执行不用管策略怎么想的——出了问题也好排查。

class EventHandler:
    def __init__(self, event_bus):
        self.event_bus = event_bus
        self.event_bus.subscribe('BarEvent', self.on_bar)
    
    def on_bar(self, event):
        # 子类实现具体逻辑
        raise NotImplementedError

架构全景图

说了这么多，咱们用一张图把整个流程串起来。下面这张 SVG 展示了事件在回测系统中的流转路径：

这张图里，数据从左边进来，变成事件，进队列，按时间顺序分发，处理器们各司其职。处理过程中产生的新事件（比如策略下单）会重新进入队列。整个系统就像一条流水线——每个环节只做一件事，但组合起来就能模拟出完整的交易生命周期。

好了，事件驱动架构的基础就这些。说白了就是三件事：定义事件、排队、处理。但就是这三件事，决定了你的回测系统能不能跑得稳、跑得快、跑得真。下一节咱们会深入事件队列的实现细节，包括如何处理不同时间粒度的事件对齐问题。

---