3. 事件与事件队列:事件驱动回测的通信中枢
事件驱动架构的核心思想是:系统的各个模块之间不直接调用,而是通过“事件”进行松耦合通信。在回测引擎中,策略模块、投资组合模块、执行模块和风控模块各自独立运行,它们通过一个统一的事件队列来收发消息。本章将详细设计事件基类、事件类型枚举以及线程安全的事件队列。
3.1 事件基类设计
所有事件都继承自同一个抽象基类 Event,该基类定义了事件的公共属性和接口。基类确保每个事件都携带时间戳和类型信息,便于队列排序和分发。
| 属性/方法 | 类型 | 说明 |
|---|---|---|
timestamp |
datetime |
事件发生的时间戳,用于回测时间轴对齐 |
event_type |
EventType |
事件类型枚举值,用于分发器路由 |
__lt__() |
方法 | 按时间戳排序,确保队列按时间顺序处理 |
from datetime import datetime
from enum import Enum, auto
class Event:
"""事件基类,所有具体事件类型的父类。"""
def __init__(self, timestamp: datetime = None):
self.timestamp = timestamp or datetime.now()
self.event_type = None # 由子类赋值
def __lt__(self, other):
"""按时间戳排序,用于优先队列。"""
return self.timestamp < other.timestamp
def __repr__(self):
return f"{self.__class__.__name__}(time={self.timestamp})"
设计要点:基类不包含业务数据,只保留时间戳和类型。具体数据由子类承载,这样事件队列可以统一处理所有事件类型,而分发器则根据
event_type 路由到对应的处理器。
3.2 事件类型枚举
事件类型枚举 EventType 定义了回测引擎中所有可能的事件种类。每个枚举值对应一个具体的事件子类,形成清晰的分类体系。
| 枚举值 | 对应事件类 | 触发时机 | 主要数据 |
|---|---|---|---|
MARKET |
MarketEvent |
每个时间片(如每分钟) | 行情快照(OHLCV) |
SIGNAL |
SignalEvent |
策略逻辑产生信号时 | 方向、标的、置信度 |
ORDER |
OrderEvent |
投资组合模块收到信号后 | 订单类型、数量、价格 |
FILL |
FillEvent |
模拟撮合引擎成交后 | 成交价、成交量、手续费 |
class EventType(Enum):
MARKET = auto() # 市场事件
SIGNAL = auto() # 信号事件
ORDER = auto() # 订单事件
FILL = auto() # 成交事件
class MarketEvent(Event):
def __init__(self, symbol: str, open: float, high: float,
low: float, close: float, volume: int,
timestamp: datetime = None):
super().__init__(timestamp)
self.event_type = EventType.MARKET
self.symbol = symbol
self.open = open
self.high = high
self.low = low
self.close = close
self.volume = volume
class SignalEvent(Event):
def __init__(self, symbol: str, direction: str,
confidence: float, timestamp: datetime = None):
super().__init__(timestamp)
self.event_type = EventType.SIGNAL
self.symbol = symbol
self.direction = direction # 'BUY' or 'SELL'
self.confidence = confidence
class OrderEvent(Event):
def __init__(self, symbol: str, order_type: str,
quantity: int, price: float = None,
timestamp: datetime = None):
super().__init__(timestamp)
self.event_type = EventType.ORDER
self.symbol = symbol
self.order_type = order_type # 'MARKET' or 'LIMIT'
self.quantity = quantity
self.price = price
class FillEvent(Event):
def __init__(self, symbol: str, fill_price: float,
fill_quantity: int, commission: float,
timestamp: datetime = None):
super().__init__(timestamp)
self.event_type = EventType.FILL
self.symbol = symbol
self.fill_price = fill_price
self.fill_quantity = fill_quantity
self.commission = commission
事件流转关系:市场数据驱动 → 策略产生信号 → 投资组合生成订单 → 模拟撮合返回成交。每个环节只关心自己需要的事件类型,模块间通过事件队列解耦。
3.3 事件队列实现
事件队列是回测引擎的“通信总线”。它需要支持多线程安全、按时间排序、以及阻塞/非阻塞两种获取模式。我们使用 Python 标准库 queue.PriorityQueue 作为底层容器,因为事件基类已经实现了 __lt__ 方法,队列会自动按时间戳排序。
import queue
from typing import Optional
class EventQueue:
"""线程安全的事件队列,按时间戳优先级排序。"""
def __init__(self):
self._queue = queue.PriorityQueue()
def put(self, event: Event):
"""向队列尾部添加一个事件。"""
self._queue.put(event)
def get(self, block: bool = True, timeout: Optional[float] = None) -> Event:
"""
从队列头部取出一个事件。
block=True 时,队列为空则阻塞等待;
block=False 时,队列为空则抛出 queue.Empty。
"""
return self._queue.get(block=block, timeout=timeout)
def empty(self) -> bool:
"""检查队列是否为空。"""
return self._queue.empty()
def qsize(self) -> int:
"""返回队列中事件的数量。"""
return self._queue.qsize()
线程安全注意事项:回测引擎通常运行在单线程事件循环中,但为了支持未来扩展(如实时交易模式),队列设计为线程安全。不要在事件处理器中长时间阻塞,否则会导致事件循环卡死。
3.4 事件驱动架构总览
下图展示了事件在回测引擎各模块之间的流动路径。箭头方向表示事件传递方向,每个模块只处理特定类型的事件,并产生新的事件放回队列。
3.5 事件循环与分发器
事件循环是回测引擎的主循环,它不断从队列中取出事件,并根据事件类型调用对应的处理器。分发器 EventDispatcher 维护一个事件类型到处理函数的映射表。
from typing import Callable, Dict
class EventDispatcher:
"""事件分发器,将事件路由到对应的处理器。"""
def __init__(self):
self._handlers: Dict[EventType, Callable] = {}
def register(self, event_type: EventType, handler: Callable):
"""注册一个事件处理器。"""
self._handlers[event_type] = handler
def dispatch(self, event: Event):
"""根据事件类型调用对应的处理器。"""
handler = self._handlers.get(event.event_type)
if handler is None:
raise ValueError(f"No handler registered for {event.event_type}")
handler(event)
class EventLoop:
"""事件循环,驱动整个回测过程。"""
def __init__(self, queue: EventQueue, dispatcher: EventDispatcher):
self.queue = queue
self.dispatcher = dispatcher
self._running = False
def start(self):
"""启动事件循环。"""
self._running = True
while self._running:
try:
event = self.queue.get(block=True, timeout=1.0)
self.dispatcher.dispatch(event)
except queue.Empty:
continue
def stop(self):
"""停止事件循环。"""
self._running = False
事件循环的两种模式:
- 回测模式:事件循环由历史数据驱动,每个时间片产生一个 MarketEvent,循环处理完所有数据后自动停止。
- 实时模式:事件循环持续运行,等待外部数据源推送新事件,适用于模拟交易或实盘对接。
3.6 事件类型与处理函数对照表
下表总结了每种事件类型对应的处理函数及其职责,方便后续章节实现时对照参考。
| 事件类型 | 处理函数 | 所属模块 | 职责描述 |
|---|---|---|---|
MARKET |
on_market() |
策略模块 | 更新策略内部状态,判断是否产生交易信号 |
SIGNAL |
on_signal() |
投资组合模块 | 根据信号和当前持仓生成具体订单 |
ORDER |
on_order() |
风控模块 | 校验订单合规性(资金、仓位限制等) |
FILL |
on_fill() |
投资组合模块 | 更新持仓、资金和交易记录 |
扩展建议:如果需要增加新的事件类型(如
NEWS 事件、ERROR 事件),只需在 EventType 枚举中添加新值,创建对应的事件子类,并在分发器中注册处理器即可,无需修改现有模块。