4. 事件循环与调度器：主循环设计、事件分发机制、定时器事件、异步事件处理

好，咱们今天聊聊回测引擎的心脏——事件循环与调度器。说实话，我见过太多回测系统跑着跑着就崩了，或者时序错乱，十有八九是这块没设计好。你想想看，回测的本质是什么？就是在模拟市场的时间流逝，把历史数据当成实时流喂给策略。这个模拟过程，全靠事件循环撑着。

4.1 主循环设计：别让CPU空转

主循环说白了就是一个while True。但怎么设计这个循环，学问大了。我个人习惯把主循环分成三个层次：

• 时间推进层：决定下一个事件什么时候发生
• 事件获取层：从队列里拿出下一个要处理的事件
• 事件处理层：把事件分发给对应的处理器

我在项目中遇到过一个问题：有人把主循环写成死等，CPU占用直接拉满。低延迟回测最忌讳这个。正确的做法是让循环在无事可做时休眠，或者用条件变量等待。

class EventLoop:
    def __init__(self):
        self._running = False
        self._event_queue = PriorityQueue()
        self._timer_heap = []  # 最小堆，存定时器事件
        
    def run(self):
        self._running = True
        while self._running:
            # 1. 先检查定时器
            self._process_timers()
            # 2. 从队列取事件，设置超时
            try:
                event = self._event_queue.get(timeout=0.001)
                self._dispatch(event)
            except Empty:
                continue  # 没事件就继续检查定时器
                
    def stop(self):
        self._running = False

4.2 事件分发机制：路由的艺术

事件来了，发给谁？这就是分发机制要解决的问题。我见过最粗暴的做法是用if-else链，事件类型一多，代码就成了一团浆糊。

更好的做法是用注册表模式。每个处理器声明自己关心什么事件，分发器维护一张映射表。嗯，这里要注意：事件类型应该用枚举或者字符串常量，别用魔法数字。

class EventDispatcher:
    def __init__(self):
        self._handlers = defaultdict(list)
        
    def register(self, event_type, handler):
        self._handlers[event_type].append(handler)
        
    def dispatch(self, event):
        handlers = self._handlers.get(event.type, [])
        for handler in handlers:
            handler(event)
            
    # 支持优先级
    def register_with_priority(self, event_type, handler, priority=0):
        heapq.heappush(self._handlers[event_type], (priority, handler))

4.3 定时器事件：时间就是金钱

回测里定时器事件太常见了：每天收盘前5分钟清仓、每周一重新平衡、每月初调参数……这些都需要精确的定时触发。

我曾经踩过一个坑：用time.sleep来做定时，结果回测速度慢得像蜗牛。后来改用最小堆管理定时器，效率提升了几十倍。

定时器事件的设计要点：

• 绝对时间 vs 相对时间：我建议统一用绝对时间（时间戳），避免时区问题
• 一次性 vs 周期性：周期性定时器要记录下次触发时间，而不是简单重复
• 精度控制：回测不需要纳秒级精度，毫秒级就够了。别为了精度牺牲性能

class TimerEvent:
    def __init__(self, trigger_time, callback, interval=None):
        self.trigger_time = trigger_time  # 绝对时间戳
        self.callback = callback
        self.interval = interval  # None表示一次性，否则是周期
        
    def __lt__(self, other):
        return self.trigger_time < other.trigger_time
        
class TimerScheduler:
    def __init__(self):
        self._heap = []
        
    def add_timer(self, timer_event):
        heapq.heappush(self._heap, timer_event)
        
    def process_due_timers(self, current_time):
        while self._heap and self._heap[0].trigger_time <= current_time:
            timer = heapq.heappop(self._heap)
            timer.callback()
            # 如果是周期性定时器，重新加入堆
            if timer.interval:
                timer.trigger_time += timer.interval
                heapq.heappush(self._heap, timer)

4.4 异步事件处理：别让一个事件卡死全局

回测里最怕什么？一个事件处理时间过长，导致后续事件全部延迟。比如策略里做了个复杂的计算，或者数据库查询超时。这时候就需要异步处理。

我个人建议用线程池来处理耗时操作，但要注意线程安全。事件循环本身是单线程的，分发事件很快，但处理事件可以扔到线程池里跑。

import concurrent.futures

class AsyncEventLoop(EventLoop):
    def __init__(self, max_workers=4):
        super().__init__()
        self._executor = concurrent.futures.ThreadPoolExecutor(max_workers)
        self._async_events = set()  # 标记哪些事件类型需要异步
        
    def mark_async(self, event_type):
        self._async_events.add(event_type)
        
    def _dispatch(self, event):
        if event.type in self._async_events:
            # 异步处理，不阻塞主循环
            self._executor.submit(self._handle_event, event)
        else:
            # 同步处理，保证顺序
            self._handle_event(event)
            
    def _handle_event(self, event):
        handlers = self._handlers.get(event.type, [])
        for handler in handlers:
            handler(event)

4.5 整体架构图

下面这张图展示了事件循环与调度器的完整工作流程。我画这张图的时候，特意把数据流向标清楚了，方便你理解各个组件怎么配合。

好了，事件循环与调度器的核心设计就这些。你想想看，其实整个回测引擎就是围绕这个循环转的。把主循环设计好了，事件分发清楚了，定时器管理精确了，异步处理到位了，你的回测系统就稳了八成。剩下的就是填业务逻辑了。