第3章做市引擎架构

做市引擎架构：事件驱动、异步IO与任务调度

做市引擎的架构设计，说白了就是解决一个核心问题：如何在极短的时间内，处理来自多个交易所的海量数据，并做出正确的交易决策。

我个人习惯把做市引擎比作一个繁忙的机场塔台。你想想看，塔台要同时监控几十架飞机的状态，接收飞行员的各种请求，还要给出起飞降落的指令。任何一个环节卡顿，都可能出大问题。做市引擎也一样——它要同时监听多个交易所的行情、订单簿、成交回报，还要管理自己的订单状态。

这一章，我们就来拆解这个塔台的核心架构。

事件驱动架构：一切皆事件

做市引擎的本质，是一个事件驱动的状态机。

什么叫事件驱动？说白了就是：没事别找我，有事我再叫你。引擎平时处于空闲状态，只有当某个事件发生时（比如收到一条新的行情数据、某个订单被成交了），它才会被唤醒，执行相应的处理逻辑。

我在项目中遇到过最典型的反例：有人用轮询的方式去检查交易所的订单状态。每10毫秒查一次，结果交易所的API有频率限制，直接把他封了。这就是典型的「没事找事」——明明可以用事件驱动，非要主动去问。

事件驱动架构的核心组件包括：

事件源：产生事件的地方，比如WebSocket连接、REST API回调、定时器
事件总线：负责事件的发布和订阅，是事件的中转站
事件处理器：消费事件并执行具体业务逻辑的模块

嗯，这里要注意：事件处理器之间不能互相阻塞。如果一个处理器在处理事件时卡住了，其他事件就得排队等着。这就是为什么我们需要异步IO。

核心原则：事件处理器必须是「非阻塞」的。任何可能耗时的操作（比如网络请求、数据库写入）都应该异步执行。

异步IO：别让CPU闲着

做市引擎要同时连接多个交易所，每个交易所都有行情推送和交易接口。如果用同步的方式，一个连接卡住了，其他连接都得等。这显然不行。

Python的异步IO（asyncio）就是解决这个问题的利器。它允许我们在单线程内并发处理多个IO操作，当一个操作在等待网络响应时，CPU可以切换到另一个操作继续执行。

我曾经踩过一个坑：刚开始用asyncio时，我在一个协程里调用了time.sleep()，结果整个事件循环都卡住了。后来才明白，在异步代码里要用await asyncio.sleep()，而不是time.sleep()。前者会主动让出控制权，后者会阻塞整个线程。

一个典型的异步IO架构如下：

import asyncio
import aiohttp

class ExchangeConnection:
    """交易所连接管理器"""
    
    def __init__(self, exchange_name, ws_url):
        self.name = exchange_name
        self.ws_url = ws_url
        self.session = None
        self.ws = None
        
    async def connect(self):
        """建立WebSocket连接"""
        self.session = aiohttp.ClientSession()
        self.ws = await self.session.ws_connect(self.ws_url)
        print(f"{self.name} 连接成功")
        
    async def listen(self, callback):
        """监听行情推送"""
        async for msg in self.ws:
            if msg.type == aiohttp.WSMsgType.TEXT:
                # 将原始数据解析为事件
                event = self.parse(msg.data)
                # 回调事件处理器
                await callback(event)
                
    async def close(self):
        """关闭连接"""
        if self.ws:
            await self.ws.close()
        if self.session:
            await self.session.close()

async def main():
    # 同时连接两个交易所
    binance = ExchangeConnection("Binance", "wss://stream.binance.com/ws")
    okx = ExchangeConnection("OKX", "wss://ws.okx.com/v5/public")
    
    await asyncio.gather(
        binance.connect(),
        okx.connect()
    )
    
    # 同时监听两个交易所的行情
    await asyncio.gather(
        binance.listen(handle_binance_event),
        okx.listen(handle_okx_event)
    )

asyncio.run(main())

避坑指南：我曾经在异步代码里混用了同步的requests库，结果导致事件循环频繁阻塞。后来全部换成aiohttp，性能提升了10倍以上。记住：在异步代码里，所有网络操作都要用异步库。

内存队列：解耦生产者和消费者

事件驱动架构里，事件的生产者和消费者往往是不同步的。行情数据可能一秒钟来几百条，但策略处理的速度可能跟不上。这时候就需要一个缓冲区——内存队列。

我习惯用Python的asyncio.Queue来做这件事。它天然支持异步操作，而且线程安全。

内存队列的作用有三个：

解耦：生产者只管往队列里塞数据，消费者只管从队列里取数据，互不干扰
削峰填谷：行情爆发时，队列可以缓冲数据，防止系统过载
优先级控制：可以给不同的事件设置不同的优先级，比如成交回报的优先级应该高于行情数据

import asyncio
from enum import Enum, auto

class EventPriority(Enum):
    HIGH = auto()   # 成交回报、错误事件
    NORMAL = auto() # 订单簿更新
    LOW = auto()    # 行情Ticker

class Event:
    """事件基类"""
    def __init__(self, event_type, data, priority=EventPriority.NORMAL):
        self.type = event_type
        self.data = data
        self.priority = priority

class EventQueue:
    """带优先级的事件队列"""
    
    def __init__(self):
        # 为每个优先级创建一个队列
        self.queues = {
            EventPriority.HIGH: asyncio.Queue(),
            EventPriority.NORMAL: asyncio.Queue(),
            EventPriority.LOW: asyncio.Queue()
        }
        
    async def put(self, event):
        """放入事件"""
        await self.queues[event.priority].put(event)
        
    async def get(self):
        """获取最高优先级的事件"""
        for priority in [EventPriority.HIGH, EventPriority.NORMAL, EventPriority.LOW]:
            if not self.queues[priority].empty():
                return await self.queues[priority].get()
        # 所有队列都为空，等待
        return await self.queues[EventPriority.NORMAL].get()

注意：内存队列不是万能的。如果生产者的速度长期大于消费者的速度，队列会无限增长，最终导致内存溢出。我建议给队列设置最大长度，超过限制时丢弃低优先级的事件，或者触发降级策略。

任务调度：谁先谁后？

有了事件队列，接下来就是任务调度。说白了就是：这么多事件，先处理哪个？

做市引擎的任务调度，我一般分为三个层次：

调度层次	说明	实现方式
事件优先级调度	根据事件类型决定处理顺序	多级队列 + 优先级选择
策略级调度	多个做市策略之间的资源分配	时间片轮转或权重分配
交易所级调度	控制对交易所API的请求频率	令牌桶算法或滑动窗口

我个人最常用的是令牌桶算法来控制对交易所的请求频率。为什么？因为交易所都有API限流，你发太快了会被封。令牌桶可以平滑请求速率，既保证吞吐量，又不会触发限流。

import asyncio
import time

class TokenBucket:
    """令牌桶限流器"""
    
    def __init__(self, rate, capacity):
        """
        rate: 每秒生成的令牌数
        capacity: 桶的最大容量
        """
        self.rate = rate
        self.capacity = capacity
        self.tokens = capacity
        self.last_refill = time.monotonic()
        self.lock = asyncio.Lock()
        
    async def acquire(self):
        """获取一个令牌，如果没有则等待"""
        while True:
            async with self.lock:
                self._refill()
                if self.tokens >= 1:
                    self.tokens -= 1
                    return True
            # 等待一段时间再试
            await asyncio.sleep(0.001)
            
    def _refill(self):
        """补充令牌"""
        now = time.monotonic()
        elapsed = now - self.last_refill
        self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
        self.last_refill = now

# 使用示例
rate_limiter = TokenBucket(rate=10, capacity=20)  # 每秒10个请求，突发20个

async def send_order(exchange, order):
    await rate_limiter.acquire()
    return await exchange.place_order(order)

整体架构图

说了这么多，我们来画一张整体的架构图。这张图展示了事件从产生到被消费的完整流程：

这张图展示了数据流的完整路径：交易所的行情数据通过WebSocket进入事件总线，被封装成事件对象后放入不同优先级的内存队列，最后由任务调度器根据优先级和限流策略分发给各个策略处理器。

我的经验：刚开始做这个架构时，我犯了一个错误——把所有事件都放在一个队列里。结果有一次行情爆发，成交回报被堵在队列后面，导致订单状态更新延迟了整整2秒。后来改成多级优先级队列，这个问题就解决了。

总结

做市引擎的架构，核心就是三件事：

事件驱动：让系统只在需要时工作，避免无谓的轮询
异步IO：充分利用CPU时间，不让网络等待成为瓶颈
内存队列+任务调度：解耦各个模块，让系统有弹性、可扩展

这个架构看起来简单，但实际落地时有很多细节。比如事件丢失怎么办？队列满了怎么处理？多个策略之间如何公平调度？这些我们会在后面的章节里逐一展开。

嗯，这一章就到这里。记住一句话：好的架构不是设计出来的，是踩坑踩出来的。

📚 课程目录

做市引擎架构：事件驱动、异步IO与任务调度

事件驱动架构：一切皆事件

异步IO：别让CPU闲着

内存队列：解耦生产者和消费者

任务调度：谁先谁后？

整体架构图

总结