事件循环与队列：量化交易系统的“心脏”

做量化交易开发这些年，我越来越觉得，一个系统的核心不是策略有多复杂，而是它怎么处理“事件”。说白了，行情来了你得接，信号到了你得执行，风控触发了你得立刻停。这些动作如果乱成一锅粥，再好的策略也白搭。

今天我们就聊聊事件循环和队列。这是事件驱动型系统的基石。我个人习惯把这两样东西比作交易系统的“心脏”和“血管”——事件循环负责泵血，队列负责让血液有序流动。

Python事件循环基础：别让CPU空转

先说说事件循环。你想想看，一个交易系统大部分时间在干嘛？等。等行情数据，等成交回报，等定时任务。如果让CPU一直轮询检查，那效率太低了。事件循环就是来解决这个问题的——它让程序在“没事干”的时候休眠，有事件来了再醒过来处理。

Python里最基础的事件循环，其实可以用一个简单的while循环模拟：

import time
import random

class SimpleEventLoop:
    def __init__(self):
        self.events = []
        self.running = False
    
    def add_event(self, event):
        self.events.append(event)
        print(f"事件已加入: {event}")
    
    def run(self):
        self.running = True
        while self.running:
            if self.events:
                event = self.events.pop(0)
                self.handle_event(event)
            else:
                # 没有事件时，让出CPU
                time.sleep(0.01)
    
    def handle_event(self, event):
        print(f"处理事件: {event}")
        # 模拟处理耗时
        time.sleep(random.uniform(0.1, 0.5))
    
    def stop(self):
        self.running = False

# 使用示例
loop = SimpleEventLoop()
loop.add_event("行情数据到达")
loop.add_event("信号触发")
loop.add_event("订单成交回报")
# loop.run()  # 实际运行时会阻塞

嗯，这里要注意。上面这个例子虽然简单，但它暴露了一个问题：如果事件处理太慢，新事件就会堆积。我在早期做回测系统时就踩过这个坑——行情处理不过来，导致信号延迟，回测结果完全失真。

避坑指南： 我曾经在实盘环境中用类似的简单循环处理行情，结果某次行情剧烈波动时，事件队列暴涨到几万条，系统直接卡死。后来我才意识到，事件循环必须配合队列使用，而且要有背压机制。

queue.Queue的使用：让事件有序流动

Python标准库里的queue.Queue，说白了就是一个线程安全的先进先出队列。它解决了两个核心问题：一是多线程环境下的数据竞争，二是生产者-消费者模型的解耦。

我一般这样用：

import queue
import threading
import time

# 创建一个事件队列
event_queue = queue.Queue(maxsize=1000)

def producer():
    """模拟行情推送"""
    for i in range(10):
        event = {
            'type': 'tick',
            'symbol': 'BTC/USDT',
            'price': 50000 + i * 10,
            'timestamp': time.time()
        }
        try:
            event_queue.put(event, timeout=1)
            print(f"生产事件: {event['type']} - {event['price']}")
        except queue.Full:
            print("队列满了，丢弃事件")
        time.sleep(0.1)

def consumer():
    """模拟策略处理"""
    while True:
        try:
            event = event_queue.get(timeout=5)
            print(f"消费事件: {event['type']} - 价格: {event['price']}")
            # 模拟处理
            time.sleep(0.2)
            event_queue.task_done()
        except queue.Empty:
            print("队列空了，等待中...")
            break

# 启动生产者和消费者
t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer)
t1.start()
t2.start()
t1.join()
t2.join()

这里有个细节：maxsize参数很关键。我见过不少新手不设上限，结果内存被撑爆。设个合理的上限，配合timeout参数，就能优雅地处理流量高峰。

我的经验： 在实盘系统中，我会把队列大小设为预期峰值流量的2-3倍。比如行情每秒最多1000条，队列就设2000-3000。这样既能缓冲突发流量，又不会占用太多内存。

多线程事件处理：别让一个任务拖垮整个系统

单线程处理事件有个致命问题：如果某个事件处理耗时很长，后面的所有事件都得等着。这在量化交易里是不能接受的——行情延迟几毫秒，可能就错过最佳买卖点。

多线程方案很直接：一个线程负责接收事件，多个工作线程负责处理。我常用的模式是线程池：

from concurrent.futures import ThreadPoolExecutor
import queue
import time

class MultiThreadEventHandler:
    def __init__(self, num_workers=4):
        self.event_queue = queue.Queue()
        self.executor = ThreadPoolExecutor(max_workers=num_workers)
        self.running = False
    
    def start(self):
        self.running = True
        # 启动事件分发线程
        self.dispatch_thread = threading.Thread(target=self._dispatch)
        self.dispatch_thread.start()
    
    def _dispatch(self):
        while self.running:
            try:
                event = self.event_queue.get(timeout=1)
                # 提交到线程池处理
                self.executor.submit(self._process_event, event)
            except queue.Empty:
                continue
    
    def _process_event(self, event):
        """实际的事件处理逻辑"""
        print(f"线程 {threading.current_thread().name} 处理: {event}")
        time.sleep(0.1)  # 模拟处理
    
    def add_event(self, event):
        self.event_queue.put(event)
    
    def stop(self):
        self.running = False
        self.executor.shutdown()

# 使用
handler = MultiThreadEventHandler(4)
handler.start()
for i in range(10):
    handler.add_event(f"事件-{i}")
time.sleep(2)
handler.stop()

为什么用线程池而不是直接开线程？因为创建线程是有开销的。线程池复用线程，避免了频繁创建销毁。我做过测试，在同样负载下，线程池比每次新建线程快30%以上。

核心要点： 事件处理要分优先级。行情处理、风控检查这类高优先级任务，应该用独立线程或更高优先级的队列。日志记录、数据存储这类低优先级任务，可以放在后台慢慢处理。

事件优先级：重要的事先做

不是所有事件都同等重要。举个例子：

  • P0（最高优先级）：风控事件、止损触发、系统错误
  • P1：行情数据、信号生成
  • P2：订单状态更新、成交回报
  • P3：日志记录、统计指标计算

Python的queue.PriorityQueue就是干这个的。它会根据优先级自动排序，优先级高的先出队：

import queue

class PriorityEvent:
    def __init__(self, priority, event_type, data):
        self.priority = priority  # 数值越小优先级越高
        self.event_type = event_type
        self.data = data
    
    def __lt__(self, other):
        return self.priority < other.priority
    
    def __repr__(self):
        return f"[P{self.priority}] {self.event_type}: {self.data}"

# 创建优先级队列
pq = queue.PriorityQueue()

# 添加不同优先级的事件
pq.put(PriorityEvent(1, "风控", "止损触发"))
pq.put(PriorityEvent(3, "日志", "系统启动完成"))
pq.put(PriorityEvent(0, "错误", "内存不足"))
pq.put(PriorityEvent(2, "行情", "BTC 50000"))

# 按优先级处理
while not pq.empty():
    event = pq.get()
    print(f"处理: {event}")

输出结果会是：

处理: [P0] 错误: 内存不足
处理: [P1] 风控: 止损触发
处理: [P2] 行情: BTC 50000
处理: [P3] 日志: 系统启动完成

你看，错误事件虽然最后加入，但因为优先级最高，最先被处理。这在实盘中至关重要——系统出错了，你得先处理错误，而不是继续处理行情。

注意： 优先级队列不是万能的。如果高优先级事件持续涌入，低优先级事件可能永远得不到处理。这就是所谓的“饥饿”问题。我的做法是给每个优先级设置一个最大处理时间，超过时间就强制切换到低优先级任务。

知识体系总览

说了这么多，我们来梳理一下整个事件驱动系统的核心逻辑。下面这张图能帮你快速理解：

事件驱动型量化交易系统核心架构 事件源 行情/订单/定时/风控 事件队列 queue.Queue PriorityQueue 事件循环 while True 事件分发 多线程事件处理器 线程1 线程2 线程3 线程4 处理结果 信号生成 / 订单执行 / 日志记录 数据来源 交易所API/数据库/定时器 缓冲与排序 FIFO / 优先级 核心调度器 最终输出

这张图展示了一个完整的事件驱动流程。事件从各个源头发起，进入队列排队，事件循环不断从队列中取出事件，分发给多线程处理器，最后输出结果。每个环节都有它的设计考量——队列解决并发问题，优先级解决重要性问题，多线程解决性能问题。

我个人觉得，理解这个流程比背代码更重要。因为不同的交易场景，你需要调整的只是参数和具体实现，但这个架构骨架是通用的。

一个小建议： 刚开始做事件驱动系统时，别急着上复杂的框架。先用queue.Queue + threading写一个简单的原型，跑通流程。等遇到性能瓶颈了，再考虑用asyncio或者更高级的消息中间件。我见过太多人一上来就上Kafka、RabbitMQ，结果大部分功能根本用不上。

好了，事件循环和队列的基础就聊到这里。这些东西看起来简单，但用好了，你的交易系统会稳定很多。下次遇到系统卡顿或者事件丢失的问题，不妨先从队列大小和线程数量入手排查——八成问题出在这里。

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

← 上一章 📖 返回目录 下一章 →