2. 事件驱动架构设计

任务目标

构建一个最小可运行的事件驱动回测框架核心，支持市场事件（BarEvent）、订单事件（OrderEvent）、成交事件（FillEvent）三类事件，并通过中央事件队列与事件循环驱动状态流转。本案例聚焦事件循环与事件分发机制，不涉及策略逻辑与撮合细节。

核心实现

from enum import Enum, auto
from queue import Queue
from dataclasses import dataclass
from typing import Callable, Dict, List

# ---------- 事件类型定义 ----------
class EventType(Enum):
    MARKET = auto()   # 市场行情事件（如新K线）
    ORDER = auto()    # 订单事件（策略发出买卖请求）
    FILL = auto()     # 成交事件（订单被撮合成交）

# ---------- 事件基类与具体事件 ----------
@dataclass
class Event:
    type: EventType
    timestamp: float

@dataclass
class MarketEvent(Event):
    symbol: str
    price: float
    volume: int
    def __init__(self, timestamp: float, symbol: str, price: float, volume: int):
        super().__init__(EventType.MARKET, timestamp)
        self.symbol = symbol
        self.price = price
        self.volume = volume

@dataclass
class OrderEvent(Event):
    symbol: str
    quantity: int
    is_buy: bool
    def __init__(self, timestamp: float, symbol: str, quantity: int, is_buy: bool):
        super().__init__(EventType.ORDER, timestamp)
        self.symbol = symbol
        self.quantity = quantity
        self.is_buy = is_buy

@dataclass
class FillEvent(Event):
    symbol: str
    quantity: int
    price: float
    is_buy: bool
    def __init__(self, timestamp: float, symbol: str, quantity: int, price: float, is_buy: bool):
        super().__init__(EventType.FILL, timestamp)
        self.symbol = symbol
        self.quantity = quantity
        self.price = price
        self.is_buy = is_buy

# ---------- 事件处理器注册表 ----------
class EventBus:
    def __init__(self):
        self._handlers: Dict[EventType, List[Callable]] = {
            et: [] for et in EventType
        }

    def register(self, event_type: EventType, handler: Callable):
        self._handlers[event_type].append(handler)

    def dispatch(self, event: Event):
        for handler in self._handlers[event.type]:
            handler(event)

# ---------- 事件循环引擎 ----------
class EventEngine:
    def __init__(self):
        self.queue = Queue()
        self.bus = EventBus()
        self._running = False

    def start(self):
        self._running = True
        while self._running:
            if not self.queue.empty():
                event = self.queue.get()
                self.bus.dispatch(event)
            else:
                # 无事件时主动退出，避免死循环
                self._running = False

    def stop(self):
        self._running = False

    def put(self, event: Event):
        self.queue.put(event)

# ---------- 最小可运行示例 ----------
if __name__ == "__main__":
    engine = EventEngine()

    # 注册各事件类型的处理器
    def on_market(event: MarketEvent):
        print(f"[{event.timestamp:.1f}] 行情 {event.symbol}: {event.price} x {event.volume}")
        # 模拟策略：价格低于100则发出买入订单
        if event.price < 100:
            engine.put(OrderEvent(event.timestamp, event.symbol, 100, is_buy=True))

    def on_order(event: OrderEvent):
        print(f"[{event.timestamp:.1f}] 订单 {'买入' if event.is_buy else '卖出'} {event.symbol} {event.quantity}股")
        # 模拟撮合：立即以当前价成交
        engine.put(FillEvent(event.timestamp, event.symbol, event.quantity, 99.5, event.is_buy))

    def on_fill(event: FillEvent):
        print(f"[{event.timestamp:.1f}] 成交 {'买入' if event.is_buy else '卖出'} {event.symbol} {event.quantity}股 @ {event.price}")

    engine.bus.register(EventType.MARKET, on_market)
    engine.bus.register(EventType.ORDER, on_order)
    engine.bus.register(EventType.FILL, on_fill)

    # 注入初始市场事件
    engine.put(MarketEvent(1.0, "AAPL", 98.5, 1000))
    engine.put(MarketEvent(2.0, "AAPL", 101.0, 1000))

    engine.start()

运行方式

python event_engine.py

验证方法与通过条件

运行后观察控制台输出，应满足以下全部条件即判定通过：

• 第一个行情事件（价格98.5）触发买入订单，随后产生成交记录；
• 第二个行情事件（价格101.0）不触发订单（价格≥100）；
• 输出顺序严格为：行情→订单→成交→行情，证明事件按队列顺序逐条处理；
• 程序在队列清空后自动退出，无死循环。

预期输出

[1.0] 行情 AAPL: 98.5 x 1000
[1.0] 订单 买入 AAPL 100股
[1.0] 成交 买入 AAPL 100股 @ 99.5
[2.0] 行情 AAPL: 101.0 x 1000

关键设计说明

• 事件队列：使用Python标准库queue.Queue，保证线程安全，事件按FIFO顺序处理；
• 事件分发：EventBus按事件类型路由到已注册的处理器，实现事件源与处理逻辑解耦；
• 循环终止条件：当队列为空时自动停止，避免无限空转。实际回测中可改为等待外部数据源持续注入事件；
• 扩展性：新增事件类型只需扩展EventType枚举并注册对应处理器，无需修改引擎核心。