如何将Asyncio调用API的结果与对应Ticker配对为嵌套字典？

核心思路是利用**asyncio.gather() 按请求顺序返回结果**的特性，将原始Ticker列表与返回的结果列表一一配对，直接通过zip()函数生成目标字典。

完整实现步骤&代码 #

1. 读取CSV获取Ticker列表
2. 构造与Ticker顺序严格对应的API请求URL列表
3. 异步批量请求API，确保结果顺序与请求顺序一致
4. 将Ticker与对应结果配对成嵌套字典

import asyncio
import aiohttp
import pandas as pd

# 从CSV读取Ticker列表（用pandas简化操作，也可替换为标准库csv模块）
def get_tickers_from_csv(csv_path):
    df = pd.read_csv(csv_path)
    if 'Ticker' not in df.columns:
        raise ValueError("CSV文件中不存在'Ticker'字段")
    return df['Ticker'].tolist()

# 异步请求单个API接口的函数
async def fetch_market_data(session, url):
    try:
        async with session.get(url) as response:
            if response.status == 200:
                return await response.json()
            else:
                return {"error": f"请求失败，HTTP状态码：{response.status}"}
    except Exception as e:
        return {"error": f"请求异常：{str(e)}"}

# 批量异步请求并生成目标字典
async def get_ticker_market_data(csv_path, api_base_url):
    tickers = get_tickers_from_csv(csv_path)
    # 构造与Ticker顺序完全对应的URL列表
    urls = [f"{api_base_url}?ticker={ticker}" for ticker in tickers]
    
    async with aiohttp.ClientSession() as session:
        # 生成所有请求的协程对象
        tasks = [fetch_market_data(session, url) for url in urls]
        # 按请求顺序获取结果集
        results = await asyncio.gather(*tasks)
    
    # 配对Ticker与对应行情数据，生成嵌套字典
    ticker_data_dict = dict(zip(tickers, results))
    return ticker_data_dict

# 调用示例
if __name__ == "__main__":
    # 替换为你的CSV路径和API基础地址
    csv_file = "your_tickers.csv"
    api_url = "https://api.example.com/market/quote"
    result_dict = asyncio.run(get_ticker_market_data(csv_file, api_url))
    
    # 打印结果示例
    for ticker, data in result_dict.items():
        print(f"{ticker}: {data}")

关键细节说明 #

• 顺序一致性保障：asyncio.gather(*tasks)会严格按照传入协程的顺序返回结果，只要URL列表的顺序和Ticker列表完全对应，就能保证结果与Ticker一一匹配。
• 异常容错：在fetch_market_data中捕获请求异常，避免单个请求失败导致整个批量任务中断，同时保留错误信息方便后续排查。
• 无pandas兼容方案：如果不想依赖pandas，可改用标准库csv模块读取Ticker列表：

import csv

def get_tickers_from_csv(csv_path):
    tickers = []
    with open(csv_path, 'r', newline='', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        if 'Ticker' not in reader.fieldnames:
            raise ValueError("CSV文件中不存在'Ticker'字段")
        for row in reader:
            tickers.append(row['Ticker'])
    return tickers

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