事件驱动框架：事件循环、事件队列、回调函数、异步IO在量化中的应用

做量化交易这些年，我越来越觉得事件驱动是个被低估的东西。很多人一上来就研究策略、回测、机器学习，结果实盘跑起来各种卡顿、漏单、延迟爆炸。说白了，你的系统能不能扛住高频行情，关键就看事件驱动这块做得怎么样。

今天咱们就聊聊事件驱动框架的核心组件：事件循环、事件队列、回调函数、异步IO。这些东西在量化里到底怎么用？我踩过哪些坑？咱们一个一个说。

1. 事件循环：系统的“心脏”

事件循环，你可以把它想象成一个永不休息的保安。它不停地问：“有事件来了吗？有事件来了吗？”一旦有事件，它就立刻处理。

在量化系统里，事件循环就是那个while True。我见过不少新手把事件循环写成死循环里塞一堆逻辑，结果CPU跑满，行情却丢了。为什么？因为事件循环不是让你在里面做计算的，它只负责“分发”。

我个人习惯用Python的asyncio库来实现事件循环。它原生支持异步，而且生态成熟。举个例子：

import asyncio

class EventLoop:
    def __init__(self):
        self.queue = asyncio.Queue()
        self.handlers = {}
    
    async def run(self):
        while True:
            event = await self.queue.get()
            handler = self.handlers.get(event.type)
            if handler:
                await handler(event)

你看，就这么简单。但实际项目中，事件循环的调度策略很重要。比如行情来了，你是先处理tick还是先处理订单回报？我建议按优先级分队列，别混在一起。

3. 事件队列：别让数据堵在路上

事件队列，说白了就是个缓冲区。行情数据、订单回报、定时任务，全都往里面塞。但这里有个坑：队列满了怎么办？

我曾经在实盘里遇到过，某次行情剧烈波动，tick数据像瀑布一样涌进来。队列直接撑爆，内存飙升，最后系统挂了。嗯，从那以后我学乖了——队列必须设置最大长度，并且要有丢弃策略。

队列类型	适用场景	注意事项
FIFO队列	普通行情、订单回报	注意长度限制
优先级队列	风控事件优先处理	优先级别设太多，2-3级足够
环形缓冲区	高频tick数据	覆盖旧数据，适合快照场景

你想想看，如果队列里堆积了10万条tick，等你处理完，行情早变了。所以我的做法是：行情队列用环形缓冲区，只保留最近1000条；订单回报用FIFO，但超时直接报警。

4. 回调函数：别让它变成“回调地狱”

回调函数是事件驱动的灵魂。事件来了，你告诉系统：“嘿，处理这个事件的时候，顺便调用这个函数。”但回调用不好，代码会变得一团糟。

我记得有一次接手一个老项目，里面回调套回调，套了五六层。想改个逻辑，得翻半天代码。这就是典型的“回调地狱”。

举个例子，处理tick数据的回调：

async def on_tick(tick):
    # 只做两件事
    await update_market_data(tick)  # 更新市场数据
    await check_signal(tick)        # 检查交易信号
    # 别在这里写日志、发通知、计算指标

这样做的好处是，回调函数轻量、可测试、容易替换。如果哪天你想换信号逻辑，直接改check_signal就行，不用动回调。

5. 异步IO：别让网络请求卡住整个系统

量化系统里，IO操作是最大的瓶颈。行情订阅、订单发送、数据库写入，这些都是IO。如果用同步方式，一个网络请求卡住，整个事件循环就停了。

异步IO就是解决这个问题的。它让系统在等待IO的时候，先去处理别的事件。说白了，就是“不等了，你先忙别的，等数据回来了再叫我”。

我常用的异步IO框架是asyncio + aiohttp。比如订阅行情：

async def subscribe_market():
    async with aiohttp.ClientSession() as session:
        async with session.ws_connect('ws://market.example.com') as ws:
            async for msg in ws:
                # 收到行情，放入事件队列
                await event_queue.put(Event('tick', msg))

这里要注意：异步IO不是银弹。如果你的回调函数里有CPU密集计算，照样会阻塞事件循环。我建议把计算密集的任务丢给线程池或进程池。

6. 实战：一个简单的事件驱动量化系统

说了这么多，咱们来个完整的例子。这是一个最小化的事件驱动系统，包含行情订阅、信号生成、订单执行：

import asyncio
from collections import deque

class QuantEngine:
    def __init__(self):
        self.event_queue = asyncio.Queue(maxsize=1000)
        self.handlers = {}
        self.market_data = deque(maxlen=100)
    
    def register(self, event_type, handler):
        self.handlers[event_type] = handler
    
    async def run(self):
        while True:
            event = await self.event_queue.get()
            handler = self.handlers.get(event['type'])
            if handler:
                await handler(event)
    
    async def on_tick(self, event):
        self.market_data.append(event['data'])
        # 检查信号
        if self.check_signal():
            await self.send_order()
    
    def check_signal(self):
        # 简单示例：价格突破
        if len(self.market_data) < 10:
            return False
        prices = [d['price'] for d in self.market_data]
        return prices[-1] > max(prices[:-1])
    
    async def send_order(self):
        # 异步发送订单
        print("发送订单...")
        await asyncio.sleep(0.01)  # 模拟网络延迟

# 启动
engine = QuantEngine()
engine.register('tick', engine.on_tick)
asyncio.run(engine.run())

这个例子虽然简单，但包含了事件驱动的核心要素。实际项目中，你还需要加日志、风控、重连机制等。但框架就是这个样子。

7. 知识体系总览

最后，我用一张图来总结事件驱动框架的核心逻辑。这张图是我自己画的，希望能帮你理清思路：

这张图展示了事件驱动的完整链路。数据源产生事件，进入队列，事件循环不断从队列里取事件，然后调用对应的回调函数，回调函数再把事件交给业务逻辑层处理，最终输出订单或日志。

记住这个链路，你的量化系统就不会乱。我在多个实盘项目里都用这个架构，稳定性和可维护性都很好。

好了，今天就聊到这里。如果你在实际项目中遇到事件驱动相关的问题，欢迎交流。记住，别让回调地狱毁了你的系统，也别让同步IO卡住你的行情。