4、Python实时数据处理：asyncio异步编程、队列管理、数据缓冲与批处理、性能优化技巧

做市交易里，数据是命根子。行情一来，每秒几千笔 tick 是常事。如果你用同步代码一条条处理，CPU 大部分时间都在等 I/O——等网络、等数据库、等磁盘。说白了，就是浪费。

我刚开始做量化系统时，就吃过这个亏。行情接口明明能扛 1000 笔/秒，我的程序却只能处理 200 笔，剩下的全堵在缓冲区里。后来一查，问题出在同步阻塞上。嗯，从那以后，我再也不敢轻视异步编程了。

4.1 asyncio 异步编程：让 CPU 不再空等

Python 的 asyncio 是协程的基础库。它不像多线程那样有 GIL 锁的问题，也不像多进程那样有高额上下文切换成本。它本质上是单线程的，但通过事件循环，让 CPU 在等待 I/O 时去干别的活。

举个例子，你同时订阅了三个交易所的深度数据。同步写法是：先等 A 返回，再等 B 返回，再等 C 返回。异步写法是：同时发出三个请求，谁先回来先处理谁。

import asyncio
import aiohttp

async def fetch_orderbook(exchange, symbol):
    url = f"https://api.{exchange}.com/orderbook/{symbol}"
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            data = await resp.json()
            print(f"{exchange} 深度数据已到达")
            return data

async def main():
    tasks = [
        fetch_orderbook("binance", "BTCUSDT"),
        fetch_orderbook("okx", "BTCUSDT"),
        fetch_orderbook("bybit", "BTCUSDT"),
    ]
    results = await asyncio.gather(*tasks)
    # 三个结果同时到手，开始做市策略计算
    process_market_making(results)

asyncio.run(main())
我的习惯： 我会把所有的网络请求、数据库写入、文件读写都包装成 async 函数。这样主循环里永远不阻塞。你想想看，做市策略里每一毫秒都可能是利润，也可能是亏损。

4.2 队列管理：数据流的交通警察

数据来了，不能直接往策略里塞。为什么？因为行情速度不稳定，有时一秒 10 笔，有时一秒 1000 笔。策略处理速度是固定的，直接对接会导致策略被撑爆。

队列就是缓冲区。我一般用 asyncio.Queue，它天然支持异步，不会阻塞事件循环。

import asyncio

class MarketDataQueue:
    def __init__(self, maxsize=10000):
        self.queue = asyncio.Queue(maxsize=maxsize)
        self.dropped_count = 0

    async def put(self, tick):
        try:
            self.queue.put_nowait(tick)  # 非阻塞写入
        except asyncio.QueueFull:
            self.dropped_count += 1
            # 记录丢包，但不要阻塞主流程
            if self.dropped_count % 100 == 0:
                print(f"警告：已丢弃 {self.dropped_count} 笔数据")

    async def get(self):
        return await self.queue.get()  # 阻塞读取，但不会卡死事件循环
我曾经踩过的坑： 队列大小设得太小，行情爆发时疯狂丢数据。后来我改成动态队列——根据过去 10 秒的平均到达速率，自动调整 maxsize。嗯，这个技巧救了我好几次。

4.3 数据缓冲与批处理：攒够了再干

做市交易里，很多操作是有固定成本的。比如写数据库，单条写入和批量写入的耗时差 10 倍。再比如计算风险敞口，每笔 tick 都算一遍，CPU 直接拉满。

我的做法是：缓冲 + 定时批处理。

import asyncio
from collections import deque

class BatchProcessor:
    def __init__(self, batch_size=100, flush_interval=0.1):
        self.buffer = deque()
        self.batch_size = batch_size
        self.flush_interval = flush_interval
        self._flush_task = None

    async def add(self, item):
        self.buffer.append(item)
        if len(self.buffer) >= self.batch_size:
            await self.flush()

    async def flush(self):
        if not self.buffer:
            return
        batch = list(self.buffer)
        self.buffer.clear()
        # 批量写入数据库或计算风险敞口
        await self._process_batch(batch)

    async def _process_batch(self, batch):
        # 模拟批量处理：计算平均价、总持仓、风险敞口
        total_qty = sum(item['qty'] for item in batch)
        avg_price = sum(item['price'] * item['qty'] for item in batch) / total_qty
        print(f"批处理：{len(batch)} 笔，总数量 {total_qty}，均价 {avg_price:.2f}")
        # 实际项目中这里会写数据库或更新风险矩阵
        await asyncio.sleep(0.01)  # 模拟 I/O

    async def start_periodic_flush(self):
        while True:
            await asyncio.sleep(self.flush_interval)
            await self.flush()

你看，这里有两个触发条件：要么攒够 100 笔，要么每 0.1 秒强制刷一次。这样既保证了实时性，又控制了 I/O 次数。

核心思路： 用空间换时间。缓冲区就是你的临时仓库，批处理就是你的卡车。仓库太小容易爆仓，卡车太少运输效率低。找到平衡点，就是性能优化的关键。

4.4 性能优化技巧：从毫秒到微秒的追求

做市交易里，延迟就是成本。我总结了几条实战经验，分享给你。

优化方向 具体做法 效果
数据结构 用 deque 代替 list 做队列 头部插入/删除从 O(n) 降到 O(1)
内存分配 预分配对象池，避免频繁创建销毁 GC 压力降低 60%
I/O 合并 多个小写入合并为一个大写入 磁盘吞吐提升 5-10 倍
锁优化 用 asyncio.Lock 代替 threading.Lock 避免 GIL 竞争，延迟降低 30%
计算优化 用 NumPy 向量化代替 Python 循环 计算速度提升 100 倍以上

举个例子，计算风险敞口时，如果你用纯 Python 循环遍历所有持仓，1000 个合约可能要 5 毫秒。但用 NumPy 向量化，0.05 毫秒就搞定了。你想想看，这 4.95 毫秒的差距，在行情剧烈波动时意味着什么。

import numpy as np

# 坏写法：Python 循环
def calc_exposure_slow(positions):
    total = 0.0
    for pos in positions:
        total += pos['qty'] * pos['price']
    return total

# 好写法：NumPy 向量化
def calc_exposure_fast(positions):
    qty = np.array([p['qty'] for p in positions])
    price = np.array([p['price'] for p in positions])
    return np.dot(qty, price)
我的个人习惯： 我会在代码里埋性能监控点。每个异步任务开始和结束时记录时间戳，然后汇总到 Prometheus。这样哪个环节慢了，一眼就能看出来。别靠感觉优化，要靠数据。

4.5 整体架构：一张图看懂

下面这张图，是我做市系统里实时数据处理的核心流程。你看一眼就能明白各个组件怎么配合。

做市实时数据处理架构 行情数据源 交易所 WebSocket 异步队列 asyncio.Queue 缓冲批处理 BatchProcessor async put async get 风险敞口计算 NumPy 向量化 数据库写入 批量 INSERT 策略信号 做市决策 flush 性能监控层 延迟追踪 | 队列深度 | 丢包率 | 批处理吞吐量 → 反馈给 Prometheus + Grafana 可视化

你看，数据从行情源进来，先经过异步队列缓冲，再进入批处理器攒批，最后分发给风险计算、数据库写入和策略信号。每一层都有性能监控盯着。这个架构我用了三年，在每秒 5000 笔的行情下依然稳定运行。

注意： 异步编程不是银弹。如果你的任务是 CPU 密集型的（比如大量数学计算），异步反而会因为单线程而变慢。这时候应该用多进程 + 异步的组合。我一般把计算密集的部分扔给 concurrent.futures.ProcessPoolExecutor，I/O 密集的部分用 asyncio。

好了，这一章的内容就这些。核心就三句话：用 asyncio 处理 I/O，用队列解耦生产者和消费者，用批处理降低固定成本。剩下的，就是在实战中不断调优了。

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

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