第4章事件循环(Event Loop)机制

4. 事件循环(Event Loop)机制：主循环的架构，如何从队列中取出事件并分发

好，咱们今天聊聊事件循环。说白了，这就是回测引擎的「心脏」。

你想想看，整个回测系统跑起来，所有的事情——数据来了、订单成交了、信号触发了——都得有个东西统一调度。谁先谁后？怎么通知？怎么处理？这就是事件循环要解决的问题。

我个人习惯把事件循环比作一个「邮局分拣系统」。事件就是信件，队列就是邮筒，而事件循环就是那个不停从邮筒里取信、然后按地址分发的邮递员。

4.1 主循环的架构：一个永不停歇的 while

事件循环的核心结构，其实特别简单。就是一个 while True，加上一个队列。

class EventLoop:
    def __init__(self):
        self.queue = Queue()  # 事件队列
        self.handlers = {}    # 事件类型 -> 处理函数
        self.running = False

    def run(self):
        self.running = True
        while self.running:
            # 从队列取事件
            event = self.queue.get()
            if event is None:  # 退出信号
                break
            # 分发事件
            self.dispatch(event)

    def dispatch(self, event):
        handler = self.handlers.get(event.type)
        if handler:
            handler(event)
        else:
            print(f"警告：未注册的事件类型 {event.type}")

嗯，这里要注意：queue.get() 是阻塞的。什么意思？就是队列里没事件的时候，循环会停在这里等，不空转CPU。这一点很重要，我在早期版本里用过非阻塞的 get_nowait()，结果CPU直接飙到100%，吓得我赶紧改回来。

4.2 事件队列：先进先出的调度核心

事件队列用什么数据结构？我建议用 queue.Queue，它是线程安全的。如果你用 collections.deque，在多线程环境下可能会出问题。

事件本身长什么样？看这个：

from dataclasses import dataclass
from datetime import datetime
from enum import Enum

class EventType(Enum):
    MARKET_DATA = 1      # 行情数据
    SIGNAL = 2           # 交易信号
    ORDER = 3            # 订单
    FILL = 4             # 成交
    TIMER = 5            # 定时器

@dataclass
class Event:
    type: EventType
    timestamp: datetime
    data: dict

为什么用 dataclass？省代码，而且自动生成 __repr__，调试的时候特别方便。我在项目里踩过一个坑：忘了给事件加时间戳，结果回测结果对不上，查了两天才发现是事件顺序乱了。

4.3 事件分发机制：路由到正确的处理器

分发机制其实就是一个「查表」的过程。事件来了，根据 event.type 找到对应的处理函数，然后调用它。

我见过很多新手把所有的处理逻辑写在一个巨大的 if-elif 里，那代码简直没法看。正确的做法是用字典映射：

class EventDispatcher:
    def __init__(self):
        self._handlers = {}

    def register(self, event_type, handler):
        """注册事件处理器"""
        if event_type not in self._handlers:
            self._handlers[event_type] = []
        self._handlers[event_type].append(handler)

    def dispatch(self, event):
        """分发事件到所有注册的处理器"""
        handlers = self._handlers.get(event.type, [])
        for handler in handlers:
            try:
                handler(event)
            except Exception as e:
                print(f"处理事件 {event.type} 时出错: {e}")
                # 这里可以加日志，但不要吞掉异常

我的小技巧： 注册处理器的时候，可以加一个优先级参数。比如行情更新处理器优先级高，先执行；日志记录处理器优先级低，最后执行。这样能控制事件处理的顺序。

4.4 事件循环的完整流程

咱们把整个流程串起来，看看实际跑起来是什么样子：

class BacktestEngine:
    def __init__(self):
        self.event_loop = EventLoop()
        self.dispatcher = EventDispatcher()
        self._setup_handlers()

    def _setup_handlers(self):
        # 注册各种事件处理器
        self.dispatcher.register(EventType.MARKET_DATA, self.on_market_data)
        self.dispatcher.register(EventType.SIGNAL, self.on_signal)
        self.dispatcher.register(EventType.ORDER, self.on_order)
        self.dispatcher.register(EventType.FILL, self.on_fill)

    def run(self):
        # 1. 加载历史数据，生成初始事件
        self._load_data()

        # 2. 启动事件循环
        while not self.event_loop.is_empty():
            event = self.event_loop.get_next()
            self.dispatcher.dispatch(event)

            # 3. 检查是否还有数据需要推进
            if self._has_more_data():
                self._push_next_data_event()

    def on_market_data(self, event):
        """处理行情数据：更新策略，生成信号"""
        price = event.data['close']
        signal = self.strategy.on_price(price, event.timestamp)
        if signal:
            # 生成信号事件，放回队列
            signal_event = Event(
                type=EventType.SIGNAL,
                timestamp=event.timestamp,
                data={'signal': signal, 'price': price}
            )
            self.event_loop.push(signal_event)

    def on_signal(self, event):
        """处理信号：生成订单"""
        order = self.portfolio.generate_order(event.data['signal'])
        if order:
            order_event = Event(
                type=EventType.ORDER,
                timestamp=event.timestamp,
                data=order
            )
            self.event_loop.push(order_event)

我曾经踩过的坑： 在事件处理函数里又往队列里放新事件，结果忘了加退出条件，导致死循环。比如行情数据触发信号，信号触发订单，订单成交又触发新的行情... 一定要在某个环节加「不再生成新事件」的判断。

4.5 事件循环的核心流程图

下面这张图，是我用SVG画的，把整个事件循环的架构展示得清清楚楚：

4.6 实际项目中的几个关键点

讲完了理论，说说我在实际项目中遇到的一些问题：

事件积压怎么办？ 如果行情数据来得太快，队列会越堆越多。我一般会加一个 maxsize 限制，超过就丢弃旧事件或者阻塞生产者。
事件优先级怎么处理？ 有些事件需要优先处理，比如风控事件。可以用 PriorityQueue 替代普通队列。
事件丢失怎么排查？ 我习惯在每个事件上加一个 event_id，用 UUID 生成。这样日志里一查就知道哪个事件丢了。

核心要点总结：

事件循环 = while True + 队列 + 分发器
队列用 queue.Queue，保证线程安全
分发用字典映射，别写 if-else 地狱
处理器里生成新事件时，注意别死循环
加时间戳和事件ID，方便调试

嗯，事件循环这块其实不难，但细节特别多。你只要把「取事件 -> 查表 -> 处理 -> 可能生成新事件」这个闭环搞清楚了，整个回测引擎的骨架就搭起来了。

公众号:蓝海资料掘金营，微信deep3321

📚 课程目录

4. 事件循环(Event Loop)机制：主循环的架构，如何从队列中取出事件并分发

4.1 主循环的架构：一个永不停歇的 while

4.2 事件队列：先进先出的调度核心

4.3 事件分发机制：路由到正确的处理器

4.4 事件循环的完整流程

4.5 事件循环的核心流程图

4.6 实际项目中的几个关键点