3. 事件与事件队列

事件驱动架构的核心思想是：系统的各个模块之间不直接调用，而是通过“事件”进行松耦合通信。在回测引擎中，策略模块、投资组合模块、执行模块和风控模块各自独立运行，它们通过一个统一的事件队列来收发消息。本章将详细设计事件基类、事件类型枚举以及线程安全的事件队列。

3.1 事件基类设计

所有事件都继承自同一个抽象基类 Event，该基类定义了事件的公共属性和接口。基类确保每个事件都携带时间戳和类型信息，便于队列排序和分发。

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})"

3.2 事件类型枚举

事件类型枚举 EventType 定义了回测引擎中所有可能的事件种类。每个枚举值对应一个具体的事件子类，形成清晰的分类体系。

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

3.6 事件类型与处理函数对照表

下表总结了每种事件类型对应的处理函数及其职责，方便后续章节实现时对照参考。