asyncio 异步编程实战：让你的 Python 量化系统快 10 倍

开篇

凌晨 3:47，你的策略因为 WebSocket 断连没有及时接收到行情，错过了黄金的波段启动点。

这不是运气不好。这是同步编程范式在高频场景下的原罪。

我们用 Python 写量化系统，习惯了 requests.get() 阻塞等响应、time.sleep() 干等行情数据、for 循环一个一个处理订单——这些操作在日常脚本里无可厚非，但一旦进入毫秒级战场，每一行同步代码都在蚕食你的 alpha。

asyncio 不是银弹。但它是目前 Python 生态里，将单线程并发效率压榨到极致的最佳路径。本文不聊语法糖，直接用生产级的代码，拆解一个异步行情处理系统的完整架构。

---

一、为什么量化系统需要 asyncio

1.1 同步 IO 的时间黑洞

看一个典型场景：订阅 10 个交易品种的实时行情，用同步方式处理。

import time
import requests

symbols = ["GCJ27.CME", "CLJ27.NYMEX", "NGJ27.IPE",
           "HGJ27.CMX", "ZSJ27.CBT", "ZCJ27.CBT",
           "KSJ27.CBT", "SBJ27.ICE", "CTZ27.CBT", "LGO27.LME"]

start = time.time()

for symbol in symbols:
    response = requests.get(
        f"https://api.tickdb.ai/v1/market/depth",
        params={"symbol": symbol},
        headers={"X-API-Key": "YOUR_API_KEY"},
        timeout=(3.05, 10)
    )
    # 串行执行，每个请求等待响应才继续下一个
    data = response.json()
    process_depth(symbol, data)

print(f"总耗时: {time.time() - start:.2f}秒")

这段代码的耗时约为：单个请求耗时 × 10。如果每个请求耗时 200ms，光是拉取 10 个品种的深度数据就要 2 秒——这还没算网络波动带来的尾延迟。

对于需要实时监控多个相关市场（原油 → 化工品链、美元指数 → 非美货币 → 贵金属）的量化系统，这是不可接受的。

1.2 异步 IO 的本质：让等待变成计算

同步 IO 的核心问题是：线程在等待 IO 时被阻塞，CPU 空转。asyncio 的解法是：不让线程等待，让它去干别的事。

import asyncio

async def fetch_depth(session, symbol):
    async with session.get(
        f"https://api.tickdb.ai/v1/market/depth",
        params={"symbol": symbol},
        headers={"X-API-Key": "YOUR_API_KEY"},
        timeout=aiohttp.ClientTimeout(total=10)
    ) as response:
        data = await response.json()
        return symbol, data

async def main(symbols):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_depth(session, s) for s in symbols]
        results = await asyncio.gather(*tasks)
        return results

start = time.time()
results = asyncio.run(main(symbols))
print(f"总耗时: {time.time() - start:.2f}秒")

同样的 10 个品种，在理想网络环境下，总耗时 ≈ 单个最慢请求的耗时（200ms 量级），而非 10 × 200ms。这就是并发带来的 10 倍效率提升。

1.3 量化系统的三个核心异步场景

场景	同步方案的问题	异步优势
多品种行情订阅	串行等待，错过后续品种更新	并发接收，同时处理 50+ 品种
WebSocket 长连接	单线程阻塞，断连后无法处理其他逻辑	非阻塞，断连期间仍可处理重连逻辑
订单执行 + 行情监控并行	订单提交后阻塞等待确认，错过行情	协程切换，订单异步等待同时持续接收行情

---

二、asyncio 核心概念快速梳理

在进入实战代码之前，先把几个关键概念对一遍，避免在工程代码里踩坑。

2.1 事件循环（Event Loop）

事件循环是 asyncio 的心脏。它负责：

1. 调度协程的执行顺序
2. 监听 IO 就绪事件（网络响应到达、文件描述符可写）
3. 在协程等待 IO 时切换到其他可执行的任务

import asyncio

# 获取当前事件循环（在 Python 3.10+ 推荐）
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

try:
    loop.run_until_complete(main())
finally:
    loop.close()

工程提醒：asyncio.run() 是 Python 3.7+ 引入的更安全的封装，生产环境中优先使用它，内部已处理好循环的创建与关闭。

2.2 协程（Coroutine）与 Task

协程是 async def 定义的函数，调用它不会立即执行，而是返回一个协程对象。

async def fetch_data():
    return await some_io_operation()

# ❌ 错误：没有 await，协程不会执行
result = fetch_data()

# ✅ 正确：包装为 Task 并发执行
task = asyncio.create_task(fetch_data())
result = await task

asyncio.create_task() 将协程交给事件循环调度，是真正的并发起点。

2.3 Awaitables 与 GIL 的误解

Python 的 GIL（全局解释器锁）限制的是 CPU 密集型操作的并行执行。但 IO 操作（网络请求、磁盘读写）在等待期间会释放 GIL，asyncio 正是利用这个窗口实现高并发。

划重点：asyncio 不替代多进程/多线程。多进程负责 CPU 密集型的因子计算，asyncio 负责 IO 密集型的行情分发，两者配合才是完整的量化系统架构。

┌─────────────────────────────────────────────────┐
│                   主进程                         │
│  ┌───────────────┐    ┌───────────────────────┐ │
│  │  asyncio       │    │  multiprocessing       │ │
│  │  (IO并发)      │    │  (CPU并行)             │ │
│  │                │    │                        │ │
│  │  · WebSocket   │    │  · 因子计算            │ │
│  │  · REST API    │    │  · 历史数据回测         │ │
│  │  · 行情分发     │    │  · 策略信号生成         │ │
│  └───────┬───────┘    └──────────┬────────────┘ │
│          │                       │              │
│          └───────────┬───────────┘              │
│                      ▼                          │
│            ┌─────────────────┐                  │
│            │  共享内存/队列   │                  │
│            │  (信号传递)      │                  │
│            └─────────────────┘                  │
└─────────────────────────────────────────────────┘

---

三、生产级异步 WebSocket 客户端

终于到核心代码了。本节给出一个完整的异步 WebSocket 行情订阅客户端，具备以下生产级特性：

• 心跳保活：检测连接存活状态
• 指数退避重连：断连后自动重试，防止惊群
• 限频处理：识别 API 频率限制并等待
• 优雅关闭：收到退出信号后有序清理资源
• 任务取消：支持协程树的有序取消

import asyncio
import aiohttp
import json
import os
import time
import random
import logging
from dataclasses import dataclass, field
from typing import Callable, Awaitable, Optional

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
)
logger = logging.getLogger("ws_client")


@dataclass
class TickDBConfig:
    """TickDB 连接配置"""
    api_key: str = field(default_factory=lambda: os.environ.get("TICKDB_API_KEY", ""))
    base_url: str = "api.tickdb.ai"
    ping_interval: int = 20          # 心跳间隔（秒）
    ping_timeout: int = 10           # 心跳超时（秒）
    max_retries: int = 10            # 最大重试次数
    base_delay: float = 1.0          # 初始重连延迟（秒）
    max_delay: float = 60.0          # 最大重连延迟（秒）
    rate_limit_code: int = 3001      # TickDB 限频错误码

    def ws_url(self, channels: list[str]) -> str:
        """构建 WebSocket 连接 URL"""
        channel_str = ",".join(channels)
        return (
            f"wss://{self.base_url}/v1/market/stream"
            f"?api_key={self.api_key}&channel={channel_str}"
        )


class AsyncTickDBClient:
    """
    TickDB 异步 WebSocket 客户端（生产级）

    特性：
    - 心跳保活（ping/pong）
    - 指数退避 + 抖动的断线重连
    - API 限频自动处理（code:3001）
    - 优雅关闭与有序任务取消
    - 回调式消息处理器
    """

    def __init__(self, config: Optional[TickDBConfig] = None):
        self.config = config or TickDBConfig()
        self._ws: Optional[aiohttp.ClientWebSocketResponse] = None
        self._session: Optional[aiohttp.ClientSession] = None
        self._tasks: set[asyncio.Task] = set()
        self._running = False
        self._closing = False
        self._last_message_time: float = 0

    # ──────────────────────────────────────────────
    # 连接生命周期
    # ──────────────────────────────────────────────

    async def connect(self, channels: list[str]) -> None:
        """建立 WebSocket 连接"""
        if self._running:
            logger.warning("连接已在运行中，忽略重复 connect() 调用")
            return

        self._running = True
        self._session = aiohttp.ClientSession()

        while self._running:
            try:
                url = self.config.ws_url(channels)
                logger.info(f"正在连接 WebSocket: {url.split('?')[0]}")

                self._ws = await self._session.ws_connect(
                    url,
                    ping_interval=self.config.ping_interval,
                    ping_timeout=self.config.ping_timeout,
                )
                logger.info("WebSocket 连接建立成功")

                # 启动监听和心跳任务
                listener = asyncio.create_task(self._listen_loop())
                heartbeat = asyncio.create_task(self._heartbeat_loop())
                self._tasks.add(listener)
                self._tasks.add(heartbeat)
                listener.add_done_callback(self._tasks.discard)
                heartbeat.add_done_callback(self._tasks.discard)

                # 连接成功，重置重试状态
                await self._wait_until_disconnect()

            except aiohttp.ClientError as e:
                logger.error(f"WebSocket 连接异常: {e}")
                await self._handle_reconnect()

            except Exception as e:
                logger.critical(f"未预期异常: {e}", exc_info=True)
                break

    async def _wait_until_disconnect(self) -> None:
        """阻塞直到连接断开"""
        if self._ws is None:
            return
        async for msg in self._ws:
            self._last_message_time = time.time()
            if msg.type == aiohttp.WSMsgType.CLOSE:
                logger.warning("收到服务端关闭帧，等待重连")
                break
            elif msg.type == aiohttp.WSMsgType.ERROR:
                logger.error(f"WebSocket 错误: {self._ws.exception()}")
                break

    # ──────────────────────────────────────────────
    # 消息处理与心跳
    # ──────────────────────────────────────────────

    async def _listen_loop(self) -> None:
        """消息监听循环"""
        if self._ws is None:
            return

        async for msg in self._ws:
            if self._closing:
                break

            if msg.type == aiohttp.WSMsgType.PING:
                # ⚠️ 浏览器/库自动处理 ping，但这里保留显式处理以兼容某些代理场景
                await self._ws.pong()
                logger.debug("收到 ping，已回复 pong")

            elif msg.type == aiohttp.WSMsgType.TEXT:
                try:
                    data = json.loads(msg.data)
                    await self._dispatch(data)
                except json.JSONDecodeError:
                    logger.warning(f"JSON 解析失败: {msg.data}")

            elif msg.type == aiohttp.WSMsgType.CLOSED:
                break

    async def _dispatch(self, data: dict) -> None:
        """
        消息分发——子类或外部可覆盖此方法实现自定义逻辑
        """
        code = data.get("code", 0)

        if code == self.config.rate_limit_code:
            # API 限频处理
            retry_after = int(data.get("headers", {}).get(
                "Retry-After",
                self._ws.get_extra_info("headers", {}).get("Retry-After", 5)
            ) if self._ws else 5)
            logger.warning(f"触发限频 (code:{code})，等待 {retry_after} 秒")
            await asyncio.sleep(retry_after)
            return

        if code != 0 and code != self.config.rate_limit_code:
            logger.error(f"API 错误 {code}: {data.get('message')}")
            return

        # 分发到业务层
        channel = data.get("channel", "")
        payload = data.get("data", data)

        if channel == "depth":
            await self._on_depth(payload)
        elif channel == "ticker":
            await self._on_ticker(payload)
        elif channel == "trade":
            await self._on_trade(payload)

    async def _heartbeat_loop(self) -> None:
        """心跳保活循环——检测连接是否存活"""
        while self._running and not self._closing:
            await asyncio.sleep(self.config.ping_interval)

            if self._ws is None or self._ws.closed:
                break

            # 检查上次消息时间
            if self._last_message_time > 0:
                idle = time.time() - self._last_message_time
                if idle > self.config.ping_interval + self.config.ping_timeout:
                    logger.warning(
                        f"心跳超时：{idle:.1f}秒未收到消息，强制断开重连"
                    )
                    await self._ws.close()
                    break

            try:
                await self._ws.ping()
                logger.debug("心跳 ping 发送成功")
            except Exception as e:
                logger.warning(f"心跳发送失败: {e}")
                break

    # ──────────────────────────────────────────────
    # 子类可覆盖的回调
    # ──────────────────────────────────────────────

    async def _on_depth(self, data: dict) -> None:
        """订单簿深度数据回调——子类覆盖实现业务逻辑"""
        pass

    async def _on_ticker(self, data: dict) -> None:
        """Ticker 数据回调——子类覆盖实现业务逻辑"""
        pass

    async def _on_trade(self, data: dict) -> None:
        """成交数据回调——子类覆盖实现业务逻辑"""
        pass

    # ──────────────────────────────────────────────
    # 重连与优雅关闭
    # ──────────────────────────────────────────────

    async def _handle_reconnect(self) -> None:
        """指数退避 + 抖动的重连"""
        retry = 0
        while self._running and retry < self.config.max_retries:
            delay = min(self.config.base_delay * (2 ** retry), self.config.max_delay)
            jitter = random.uniform(0, delay * 0.1)
            wait = delay + jitter

            logger.info(f"第 {retry + 1} 次重连尝试，{wait:.1f} 秒后执行")
            await asyncio.sleep(wait)

            if not self._running:
                break

            retry += 1

        if retry >= self.config.max_retries:
            logger.critical("达到最大重试次数，客户端终止")

    async def close(self) -> None:
        """优雅关闭：取消任务 → 关闭连接 → 关闭会话"""
        logger.info("收到关闭信号，开始优雅退出")
        self._closing = True
        self._running = False

        # 取消所有子任务
        for task in self._tasks:
            if not task.done():
                task.cancel()

        if self._tasks:
            await asyncio.gather(*self._tasks, return_exceptions=True)
        self._tasks.clear()

        # 关闭 WebSocket
        if self._ws and not self._ws.closed:
            await self._ws.close()

        # 关闭会话
        if self._session and not self._session.closed:
            await self._session.close()

        logger.info("客户端已完全关闭")

3.1 工程要点逐行拆解

1. 心跳不是写死的，是可配置的

ping_interval: int = 20          # 心跳间隔（秒）
ping_timeout: int = 10           # 心跳超时（秒）

TickDB 的 WebSocket 服务要求客户端定期发送 ping。大多数库会自动处理，但这里显式实现了两层机制：库级别的 ping_interval（让 aiohttp 自动发）+ 自定义心跳任务（主动检查最后消息时间）。两层保障，确保长连接真正存活。

2. 指数退避 + 抖动

delay = min(self.config.base_delay * (2 ** retry), self.config.max_delay)
jitter = random.uniform(0, delay * 0.1)

直接重连会放大服务端压力。指数退避让重试间隔按 1s → 2s → 4s → 8s 增长；抖动（jitter）在间隔上加随机偏移，避免大量客户端同时重连的“惊群效应”。

3. 任务树的有序取消

for task in self._tasks:
    if not task.done():
        task.cancel()
await asyncio.gather(*self._tasks, return_exceptions=True)

直接 close() 不会等待子任务，可能导致资源泄漏。用 gather(..., return_exceptions=True) 收集所有任务的取消结果，即使是取消引发的 CancelledError 也作为正常返回处理。

---

四、实战：基于 TickDB depth 频道的买卖压力监控

光有客户端框架不够，这里给出一个具体用例：监控贵金属板块的买卖压力比，在流动性失衡时触发告警。

import asyncio
import os
from dataclasses import dataclass, field
from collections import deque
from datetime import datetime

os.environ.setdefault("TICKDB_API_KEY", "YOUR_API_KEY_HERE")


@dataclass
class DepthSnapshot:
    """订单簿快照"""
    symbol: str
    timestamp: float
    bids: list[tuple[float, float]]   # [(price, size), ...]
    asks: list[tuple[float, float]]   # [(price, size), ...]

    @property
    def pressure_ratio(self) -> float:
        """买卖压力比：买盘量 / 卖盘量（5档聚合）"""
        bid_vol = sum(size for _, size in self.bids[:5])
        ask_vol = sum(size for _, size in self.asks[:5])
        return bid_vol / ask_vol if ask_vol > 0 else 0

    @property
    def spread_bps(self) -> float:
        """买卖价差（基点）"""
        if not self.bids or not self.asks:
            return 0
        mid = (self.bids[0][0] + self.asks[0][0]) / 2
        spread = self.asks[0][0] - self.bids[0][0]
        return (spread / mid) * 10000 if mid > 0 else 0


class PressureMonitor(AsyncTickDBClient):
    """
    买卖压力监控器

    监控逻辑：
    1. 实时接收 depth 数据，计算压力比
    2. 维护滚动窗口，检测压力比的突变
    3. 当压力比超过阈值时触发飞书告警
    """

    def __init__(
        self,
        symbols: list[str],
        alert_threshold: float = 2.5,
        window_size: int = 20,
    ):
        super().__init__()
        self.symbols = symbols
        self.alert_threshold = alert_threshold
        self.history: dict[str, deque[float]] = {
            s: deque(maxlen=window_size) for s in symbols
        }
        self.channels = ["depth"]  # 订阅 depth 频道
        # ⚠️ 注意：美股 depth 为 1 档，港股/数字货币支持多档

    async def _on_depth(self, data: dict) -> None:
        """处理 TickDB depth 频道数据"""
        symbol = data.get("symbol", "")
        if symbol not in self.symbols:
            return

        bids = data.get("b", [])   # [[price, size], ...]
        asks = data.get("a", [])

        snapshot = DepthSnapshot(
            symbol=symbol,
            timestamp=datetime.now().timestamp(),
            bids=bids,
            asks=asks,
        )

        # 更新历史窗口
        self.history[symbol].append(snapshot.pressure_ratio)

        # 计算滑动均值（剔除首尾极值）
        window = list(self.history[symbol])
        if len(window) < 5:
            return

        # 简单移动平均
        smoothed = sum(window[-5:]) / 5
        current = snapshot.pressure_ratio

        # 突变检测：当前值相比均值偏离超过 50%
        if smoothed > 0:
            deviation = abs(current - smoothed) / smoothed

            if deviation > 0.5 and current > self.alert_threshold:
                msg = (
                    f"🚨 流动性告警\n"
                    f"品种: {symbol}\n"
                    f"压力比: {current:.2f} (均值为 {smoothed:.2f})\n"
                    f"偏离: +{deviation*100:.1f}%\n"
                    f"价差: {snapshot.spread_bps:.1f} bps\n"
                    f"时间: {datetime.now().strftime('%H:%M:%S')}"
                )
                print(msg)
                await self.send_alert(msg)

    async def send_alert(self, message: str) -> None:
        """发送飞书告警——生产环境中替换为实际 webhook 调用"""
        # ⚠️ 示例占位：替换为你的飞书 webhook URL
        # async with aiohttp.ClientSession() as session:
        #     await session.post(
        #         "https://open.feishu.cn/open-apis/bot/v2/hook/YOUR_WEBHOOK",
        #         json={"msg_type": "text", "content": {"text": message}}
        #     )
        pass


async def main():
    symbols = ["GCJ27.CME", "HGJ27.CMX", "PAJ27.NYMEX"]
    monitor = PressureMonitor(
        symbols=symbols,
        alert_threshold=2.5,
        window_size=20,
    )

    # 设置信号处理，实现 Ctrl+C 优雅退出
    loop = asyncio.get_running_loop()
    stop_event = asyncio.Event()

    def shutdown():
        print("\n收到退出信号，正在关闭...")
        stop_event.set()

    loop.add_signal_handler(asyncio.signals.SIGINT, shutdown)
    loop.add_signal_handler(asyncio.signals.SIGTERM, shutdown)

    # 并发运行：监控任务 + 退出信号监听
    monitor_task = asyncio.create_task(
        monitor.connect(channels=["depth"])
    )
    watcher_task = asyncio.create_task(stop_event.wait())

    done, pending = await asyncio.wait(
        [monitor_task, watcher_task],
        return_when=asyncio.FIRST_COMPLETED,
    )

    for task in pending:
        task.cancel()
    await asyncio.gather(*pending, return_exceptions=True)

    await monitor.close()


if __name__ == "__main__":
    print("=" * 60)
    print("买卖压力监控启动 | 品种: GCJ27/HGJ27/PAJ27")
    print("按 Ctrl+C 退出")
    print("=" * 60)
    asyncio.run(main())

4.1 代码核心逻辑图解

TickDB WebSocket
      │
      ▼
┌─────────────────────────────────┐
│  _on_depth() 接收 depth 数据    │
│  data = {"symbol":"GCJ27.CME",  │
│           "b": [[...]],          │
│           "a": [[...]]}          │
└──────────────┬──────────────────┘
               ▼
┌─────────────────────────────────┐
│  构建 DepthSnapshot             │
│  计算 pressure_ratio            │
│  更新 deque 历史窗口            │
└──────────────┬──────────────────┘
               ▼
┌─────────────────────────────────┐
│  滑动均值 vs 当前值对比          │
│  deviation > 50% && ratio > 2.5│
│         → 触发告警              │
└─────────────────────────────────┘

4.2 部署建议

场景	配置	说明
个人研究	1 个监控进程 + 3 个品种	免费层足够，API 限频 10 QPS
团队协作	1 个采集进程 + 多个告警消费者	通过 Redis 队列分发
机构级	采集 + 风控 + 通知三层分离	采集进程独立，不受下游故障影响

---

五、asyncio 在量化系统中的进阶用法

5.1 并发限制：Semaphore 控制并发度

当需要同时发起大量请求（如全市场扫描），但 API 有并发限制时，用 Semaphore 控制。

import asyncio

# 限制最多 5 个并发连接
semaphore = asyncio.Semaphore(5)

async def fetch_single(session, symbol):
    async with semaphore:
        async with session.get(
            f"https://api.tickdb.ai/v1/market/kline/latest",
            params={"symbol": symbol},
            headers={"X-API-Key": os.environ.get("TICKDB_API_KEY")},
            timeout=aiohttp.ClientTimeout(total=10)
        ) as resp:
            return symbol, await resp.json()

async def scan_all(symbols: list[str]):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_single(session, s) for s in symbols]
        results = await asyncio.gather(*tasks)
        return {symbol: data for symbol, data in results}

5.2 带优先级的任务调度

在实盘系统中，行情数据需要优先处理，订单确认次之，日志写入优先级最低。用 asyncio.PriorityQueue 实现优先级队列。

import asyncio
from dataclasses import dataclass, field
from typing import Any
import heapq


@dataclass(order=True)
class PriorityTask:
    priority: int  # 数值越小优先级越高
    content: Any = field(compare=False)


async def priority_worker(queue: asyncio.PriorityQueue):
    """优先级队列消费者：行情(0) > 订单(1) > 日志(2)"""
    while True:
        task = await queue.get()
        priority, content = task.priority, task.content

        if priority == 0:
            await process_market_data(content)
        elif priority == 1:
            await process_order_response(content)
        elif priority == 2:
            await write_log(content)

        queue.task_done()


# 使用示例
queue = asyncio.PriorityQueue()
await queue.put(PriorityTask(2, "INFO: 策略启动"))
await queue.put(PriorityTask(0, {"symbol": "GCJ27", "last": 2034.5}))  # 优先处理
await queue.put(PriorityTask(1, {"order_id": "O-123", "status": "FILLED"}))

5.3 与因子计算的多进程协同

回到开头的架构图：asyncio 负责 IO，multiprocessing 负责 CPU 密集型计算。两者通过 multiprocessing.Queue 通信。

import asyncio
import multiprocessing as mp
from concurrent.futures import ProcessPoolExecutor


def compute_signal(depth_data: list[dict]) -> dict:
    """CPU 密集型：计算买卖信号——在独立进程中运行"""
    # 模拟复杂因子计算
    bid_vol = sum(d["bid_size"] for d in depth_data)
    ask_vol = sum(d["ask_size"] for d in depth_data)
    pressure = bid_vol / ask_vol if ask_vol > 0 else 1.0

    signal = "BUY" if pressure > 1.5 else ("SELL" if pressure < 0.67 else "HOLD")
    return {"signal": signal, "pressure": pressure}


async def mixed_pipeline(market_data_queue: asyncio.Queue):
    """
    混合架构：
    - asyncio 处理 IO（行情接收 + 结果发送）
    - ProcessPoolExecutor 执行 CPU 计算
    """
    loop = asyncio.get_running_loop()
    executor = ProcessPoolExecutor(max_workers=4)

    while True:
        # 异步接收行情数据
        batch = await market_data_queue.get()

        # 在线程池中并发运行多个进程的 CPU 任务
        futures = [
            loop.run_in_executor(executor, compute_signal, batch),
            loop.run_in_executor(executor, compute_signal, batch),
        ]
        results = await asyncio.gather(*futures)

        print(f"计算结果: {results}")
        market_data_queue.task_done()

小 K 提醒：这里容易踩的坑是混用 ProcessPoolExecutor 和 ThreadPoolExecutor。前者用于真正绕过 GIL 做 CPU 并行，后者适合在协程中穿插同步 IO 调用（虽然这种情况更推荐直接用协程）。

---

六、常见陷阱与排查指南

asyncio 的坑大多集中在几个固定模式上，提前知道能省很多调试时间。

陷阱	症状	解决方案
在协程外调用 await	SyntaxError	asyncio.run() 入口，或在另一个协程内调用
忘记 await，协程变成"幽灵任务"	代码执行了但没有任何效果	静态检查：所有协程调用必须被 await
在同步函数中调用异步函数	RuntimeError: no running event loop	用 asyncio.run() 包装，或重构为异步入口
长时同步操作阻塞事件循环	其他协程全部卡死	用 loop.run_in_executor() 扔进线程池
取消协程时资源未释放	连接泄漏、文件句柄耗尽	使用 try/finally 确保清理代码始终执行
GIL 误解导致性能瓶颈	量化计算慢，怀疑 asyncio 有问题	asyncio 不加速 CPU 计算，那是 multiprocessing 的工作

---

七、下一步行动

如果你的系统还在用同步 requests 拉数据：

1. 将 requests.get() 替换为 aiohttp.ClientSession.get()
2. 用 asyncio.gather() 并发请求多个品种
3. 基准测试对比：记录优化前后的端到端延迟

如果你需要处理高频行情流：

1. 直接使用本文的 AsyncTickDBClient 作为起点
2. 按业务需求覆盖 _on_depth()、_on_ticker() 回调
3. 在信号计算层引入 multiprocessing，将 CPU 任务分离

如果你已经是 asyncio 老手：
在 ClawHub 搜索安装 tickdb-market-data SKILL，让 AI 助手帮你生成定制的异步行情处理代码——输入你的策略逻辑，返回可直接运行的协程框架。

---

回测局限性说明：本文示例代码用于演示 asyncio 架构设计，不构成任何策略实盘建议。买卖压力比作为单一信号的预测效力需要经过充分的历史回测验证（建议样本量 ≥ 100 次事件，覆盖至少一个牛熊周期），并充分考虑交易成本和流动性冲击。

市场有风险，投资需谨慎。