如何使用multiprocessing进程池改写for循环加速yfinance数据下载

改写思路 #

你原有代码的性能瓶颈主要在网络IO等待，串行逐个拉取股票数据浪费了大量等待时间。我们可以把单只股票的处理逻辑抽成独立函数，用进程池并行执行，最后在主进程统一拼接结果，完全避免多进程操作全局变量的竞态问题。

完整可运行代码 #

import yfinance as yf
import pandas as pd
import concurrent.futures

def process_single_ticker(ticker, start_date, end_date):
    print(f"正在处理{ticker}")
    # 拉取行情数据
    prc = yf.download(ticker, interval="1d", start=start_date, end=end_date, progress=False)
    if prc.empty:
        return None
    # 处理价格序列
    prc_series = prc['Adj Close'].rename(ticker)
    # 拉取标的基础信息
    stk = yf.Ticker(ticker)
    # 兼容字段缺失情况
    float_shares = stk.info.get('floatShares')
    shares_outstanding = stk.info.get('sharesOutstanding')
    market_cap = stk.info.get('marketCap')
    prev_close = stk.info.get('previousClose')
    # 计算市值序列
    if float_shares:
        mcap_series = prc_series * float_shares
    elif shares_outstanding:
        mcap_series = prc_series * shares_outstanding
    elif market_cap and prev_close:
        mcap_series = prc_series * (market_cap / prev_close)
    else:
        mcap_series = pd.Series(index=prc_series.index, name=ticker, dtype='float64')
    return (prc_series, mcap_series)

if __name__ == '__main__':
    # 原有前置逻辑
    start_date = "2021-01-04"
    end_date = "2021-11-29"
    idx = "^STOXX50E"

    Index = yf.download(idx, interval="1d", start=start_date, end=end_date)
    Index = pd.DataFrame(Index['Adj Close'].rename(idx))

    page = pd.read_html('https://en.wikipedia.org/wiki/EURO_STOXX_50')
    constituents = page[2]['Ticker'].tolist()
    constituents.pop(46) # 移除报错的UMG.AS

    # 进程池并行处理
    prc_list = []
    mcap_list = []
    # max_workers可根据你的网络情况调整，IO密集型建议设为8-16
    with concurrent.futures.ProcessPoolExecutor(max_workers=8) as executor:
        tasks = [executor.submit(process_single_ticker, ticker, start_date, end_date) for ticker in constituents]
        for future in concurrent.futures.as_completed(tasks):
            res = future.result()
            if res is not None:
                prc_list.append(res[0])
                mcap_list.append(res[1])

    # 统一拼接结果，输出和原有代码完全一致
    df_prc = pd.concat(prc_list, axis=1)
    df_mcap = pd.concat(mcap_list, axis=1)

注意事项 #

• Windows系统必须把所有执行逻辑放到if __name__ == '__main__'块中，否则会出现多进程启动异常
• 你的场景以网络IO等待为主，也可以直接把ProcessPoolExecutor替换为ThreadPoolExecutor，无需修改其他代码，运行效率差别不大，还能避免部分序列化兼容问题
• 代码中yf.download加了progress=False避免多进程打印进度条混乱，需要查看下载进度可以删除该参数

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