Python：如何为股票数据处理函数正确实现Concurrent Futures并解决输出不一致问题

• 最常见诱因是使用concurrent.futures.as_completed时直接按任务完成顺序拼接结果：不同股票接口返回速度不一致，先完成的任务会先被返回，直接追加到结果列表就会和原始股票代码顺序错位。
• 第二种常见错误是多任务共享全局可变对象未加锁：多个线程/进程同时写入同一个列表、字典时，会出现写入竞争，导致数据丢失、覆盖或者乱序。
• 第三种问题是任务返回值未绑定对应股票代码：如果单任务只返回拉取到的金融数据，没有附带对应的股票代码标识，最终汇总时无法匹配对应标的，必然出现错位。

拉取金融数据属于IO密集型任务，优先使用ThreadPoolExecutor即可，不需要用到ProcessPoolExecutor，避免进程通信的额外开销，以下是两种稳定的实现方案：

方案1：使用executor.map（天然保序，适合直接替换串行循环） #

map方法的返回结果顺序和输入的股票代码列表顺序完全一致，不需要额外处理顺序问题，改造量最小。

import concurrent.futures
import requests

# 直接复用你现有串行程序里的单股票拉取逻辑
def fetch_stock_data(stock_code):
    # 此处保留你原本的接口请求、数据解析逻辑
    resp = requests.get(f"你的金融数据接口地址/{stock_code}")
    data = resp.json()
    return (stock_code, data)

if __name__ == "__main__":
    stock_codes = ["SH600000", "SZ000001"] # 替换为你的股票代码列表
    max_workers = 20 # 可根据接口限流规则调整，一般10-50区间即可
    result = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 返回结果顺序和stock_codes的顺序完全匹配，不会错位
        for stock_code, data in executor.map(fetch_stock_data, stock_codes):
            result.append({
                "stock_code": stock_code,
                "data": data
            })
    # 此处result可直接按原有逻辑做后续处理

方案2：使用as_completed（适合需要实时处理单任务结果的场景） #

如果需要每完成一个股票的拉取就做一次增量处理，使用as_completed时必须绑定任务和对应股票代码，避免乱序。

if __name__ == "__main__":
    stock_codes = ["SH600000", "SZ000001"] # 替换为你的股票代码列表
    max_workers = 20
    result_dict = {}
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 提交任务时绑定future和对应股票代码
        future_to_code = {executor.submit(fetch_stock_data, code): code for code in stock_codes}
        for future in concurrent.futures.as_completed(future_to_code):
            code = future_to_code[future]
            try:
                data = future.result()
                result_dict[code] = data
                # 此处可加你需要的实时处理逻辑
            except Exception as e:
                print(f"拉取{code}失败: {str(e)}")
                result_dict[code] = None
    # 如果需要按原始股票列表顺序输出，遍历原始列表取字典值即可
    sorted_result = [{"stock_code": code, "data": result_dict[code]} for code in stock_codes]

• 调整max_workers时请参考你调用的金融数据接口的限流规则，避免请求频率过高被封禁。
• 如果拉取逻辑中有写入本地文件、数据库的操作，写入时要加锁，避免多线程写入冲突。
• 单任务内部要做好异常捕获，避免单个股票拉取失败导致整个进程崩溃。

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