异步调用CCXT fetchTickers时如何关联交易所与返回数据？

问题场景 #

异步调用CCXT的fetchTickers获取Binance、Bybit的行情数据，期望返回{'binance': {...}, 'bybit': {...}}的字典结构，但当前用asyncio.gather返回的是无关联的数组，无法区分各交易所对应的数据；尝试用asyncio.create_task指定name属性但不知如何关联数据。

解决方案 #

以下三种方法均可实现结果与交易所的关联，按需选择：

方法1：用字典存储任务，通过键值对关联结果

利用字典保存交易所实例和对应的异步任务，待所有任务完成后，通过zip将交易所ID和结果组装成目标字典：

import asyncio
import os
import sys

root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.append(root + '/python')

import ccxt.async_support as ccxt


async def test():
    # 创建交易所实例字典
    exchanges = {}
    for exchange_id in ['binance', 'bybit']:
        exchange_class = getattr(ccxt, exchange_id)
        exchanges[exchange_id] = exchange_class()

    # 构建任务字典：键为交易所ID，值为fetchTickers异步任务
    tasks = {exchange_id: exchange.fetchTickers() for exchange_id, exchange in exchanges.items()}
    
    # 批量执行所有任务
    results = await asyncio.gather(*tasks.values())
    
    # 组装成目标格式的字典
    ticker_data = dict(zip(tasks.keys(), results))
    print(ticker_data)
    
    # 关闭所有交易所连接，避免资源泄漏
    await asyncio.gather(*[exchange.close() for exchange in exchanges.values()])


asyncio.run(test())

方法2：封装异步函数，返回（交易所ID+数据）的元组

自定义一个异步函数，每次调用fetchTickers时同时返回交易所ID，最后将元组列表转为字典：

import asyncio
import os
import sys

root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.append(root + '/python')

import ccxt.async_support as ccxt


async def fetch_with_exchange_id(exchange):
    # 返回包含交易所ID和行情数据的元组
    return (exchange.id, await exchange.fetchTickers())


async def test():
    instances = []
    for exchange_id in ['binance', 'bybit']:
        exchange_class = getattr(ccxt, exchange_id)
        instances.append(exchange_class())

    # 创建任务列表，每个任务调用封装后的函数
    tasks = [fetch_with_exchange_id(exchange) for exchange in instances]
    results = await asyncio.gather(*tasks)
    
    # 将元组列表直接转为字典
    ticker_data = dict(results)
    print(ticker_data)
    
    # 关闭连接
    await asyncio.gather(*[exchange.close() for exchange in instances])


asyncio.run(test())

方法3：利用asyncio.create_task的name属性（Python 3.8+）

给每个任务指定name为交易所ID，任务完成后通过task.name和task.result()关联数据：

import asyncio
import os
import sys

root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.append(root + '/python')

import ccxt.async_support as ccxt


async def test():
    instances = []
    tasks = []
    for exchange_id in ['binance', 'bybit']:
        exchange_class = getattr(ccxt, exchange_id)
        exchange = exchange_class()
        instances.append(exchange)
        # 创建带name属性的异步任务，name设为交易所ID
        task = asyncio.create_task(exchange.fetchTickers(), name=exchange.id)
        tasks.append(task)
    
    # 等待所有任务执行完成
    await asyncio.gather(*tasks)
    
    # 遍历任务，通过name和result组装字典
    ticker_data = {task.name: task.result() for task in tasks}
    print(ticker_data)
    
    # 关闭连接
    await asyncio.gather(*[exchange.close() for exchange in instances])


asyncio.run(test())

内容的提问来源于stack exchange，提问作者stas stas