如何优化带异常捕获的异步批量获取交易所价格代码

以下是针对你需求优化后的异步实现代码，解决了冗余代码、异常捕获以及CCXT异步调用不规范的问题：

import asyncio
import ccxt

async def getprices(exchange, symbol):
    # 初始化CCXT异步交易所实例，启用限速（CCXT官方强制推荐）
    inst = getattr(ccxt, exchange)({'enableRateLimit': True})
    try:
        # 调用CCXT异步版本的行情接口，必须添加await实现真正异步
        prices = await inst.fetch_ticker(symbol)
        return (exchange, prices)
    except Exception as e:
        print(f'price pull fail for {exchange}: {str(e)}')
        return (exchange, None)

async def main(exchanges, symbol):
    # 批量生成异步任务，彻底消除手动创建task的冗余代码
    tasks = [asyncio.create_task(getprices(exchange, symbol)) for exchange in exchanges]
    # 等待所有任务完成，一次性获取所有结果
    results = await asyncio.gather(*tasks)
    
    prices_data = {}
    fails = []
    # 统一处理结果，分离成功/失败数据
    for exchange, price in results:
        if price is not None:
            prices_data[exchange] = price
            print(f'prices successfully pulled for {exchange}')
        else:
            fails.append(exchange)
    
    return prices_data, fails

if __name__ == '__main__':
    # 示例配置：可根据实际需求修改交易所列表和交易对
    exchanges = ['binance', 'coinbasepro', 'kraken', 'kucoin']
    symbol = 'BTC/USDT'
    prices_data, fails = asyncio.run(main(exchanges, symbol))
    
    # 输出最终结果
    print('\n===== Final Results =====')
    print(f'Successfully fetched prices from: {list(prices_data.keys())}')
    print(f'Failed to fetch from: {fails}')

关键优化点 #

• 消除代码冗余：通过列表推导式批量创建异步任务，支持任意长度的交易所列表，无需硬编码task1~taskN，扩展性拉满。
• 规范CCXT异步调用：
  • 开启enableRateLimit=True，遵守交易所API限流规则，避免被封禁。
  • 调用CCXT的异步方法时添加await关键字，真正实现非阻塞的异步IO（原代码未加await，本质还是同步执行）。
• 完善异常处理：在行情获取函数内捕获异常，同时返回失败标识，后续统一归集失败交易所，和原同步代码的业务逻辑完全匹配。
• 增强复用性：将交易所列表、交易对作为参数传入主函数，不用修改核心代码即可适配不同业务场景。

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