如何动态组合150-250个async函数实现Binance币种价格计算需求

核心思路 #

你不需要引入多线程等额外复杂逻辑，直接利用Python可变参数+列表推导式，就可以实现动态生成异步任务的需求，完全适配你150-250个币种的并发计算场景。

基础实现步骤 #

1. 先实现动态拉取币种列表的逻辑，封装为函数（可以是异步也可以是同步，根据你拉取API的方式调整），返回格式为['BTC', 'ETH', ...]的币种列表。
2. 调整main函数，通过列表推导式批量生成所有calculate协程，通过可变参数解包传给asyncio.gather即可。

代码示例 #

import asyncio
from binance import AsyncClient

# 你的原有calculate函数保持不变
async def calculate(client, queue, coin):
    while 1:
        # 原有计算逻辑不变
        ...

# 动态拉取待计算币种列表的函数，自行实现内部逻辑
async def fetch_supported_coins():
    # 示例：调用你的API拉取最新需要计算的币种
    # resp = await your_api_request()
    # return resp['coin_list']
    return ['BTC', 'ETH', 'BNB', 'LINK', 'SHIB', 'ADA', 'XRP', 'VET'] # 示例返回值

async def main(queue):
    # 初始化Binance异步客户端
    client = await AsyncClient.create(api_key="你的API_KEY", api_secret="你的API_SECRET")
    try:
        # 拉取最新币种列表
        coin_list = await fetch_supported_coins()
        # 批量生成所有计算任务
        task_list = [calculate(client, queue, coin) for coin in coin_list]
        # 并发执行所有任务
        await asyncio.gather(*task_list)
    finally:
        # 程序退出时关闭客户端连接
        await client.close_connection()

if __name__ == "__main__":
    queue = asyncio.Queue() # 初始化你的队列
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main(queue))

进阶优化 #

单个任务异常自动重启 #

默认情况下asyncio.gather中任意一个任务抛出异常，会导致所有任务被取消。可以给calculate加一层异常捕获 wrapper，保证单个币种任务异常后自动重启，不影响其他币种的计算：

async def safe_calculate(client, queue, coin):
    while True:
        try:
            await calculate(client, queue, coin)
        except Exception as e:
            print(f"币种{coin}计算任务异常: {str(e)}，3秒后自动重启")
            await asyncio.sleep(3)

生成任务列表时替换为safe_calculate即可：

task_list = [safe_calculate(client, queue, coin) for coin in coin_list]

运行时动态调整币种 #

如果需要在不重启程序的情况下动态增减计算币种，可以额外加一个定时监控任务，每隔一段时间拉取最新的币种列表，和当前运行的任务对比，自动创建新任务、取消不需要的旧任务即可。

注意事项 #

你的场景属于IO密集型异步任务，asyncio单线程即可轻松承载200+的并发计算任务，无需引入多进程、多线程，反而会增加资源消耗和逻辑复杂度。

内容的提问来源于stack exchange，提问作者T. de Jong