如何用并行请求优化基于urllib的Bittrex API数据爬取效率？

嘿，我太懂你串行请求295个币种要耗5分钟的崩溃感了——这种网络请求属于IO密集型任务，大部分时间都在等服务器响应，用并行处理绝对能把耗时大幅压缩！我给你一套用concurrent.futures实现的方案，比直接用multiprocessing更简洁，完美适配你的需求，还能保留DataFrame写入CSV的功能。

核心思路 #

• 用线程池并行发起请求：网络请求是IO密集型，线程切换开销远小于进程，效率更高
• 单线程独立处理单个币种：每个线程负责请求、解析数据并构造DataFrame
• 统一写入CSV：先收集所有有效结果再一次性写入，避免多线程写文件的竞争冲突

完整可运行代码 #

import pandas as pd
import json
import urllib.request
import concurrent.futures
import datetime

# 替换成你的295个币种列表，这里先放示例数据
tickers = ["BTC-1ST", "BTC-2GIVE", "BTC-ABY", "BTC-ARDR", "BTC-WAVE"]

def fetch_single_ticker_data(ticker):
    """单个币种的数据请求与处理逻辑，每个线程执行这个函数"""
    request_url = f"https://bittrex.com/Api/v2.0/pub/market/GetTicks?marketName={ticker}&tickInterval=thirtyMin"
    try:
        with urllib.request.urlopen(request_url) as response:
            raw_data = json.loads(response.read().decode())
            ticker_df = pd.DataFrame(data=raw_data['result'])
            ticker_df['Market'] = ticker  # 标记当前币种
            print(f"✅ 完成 {ticker} 的数据获取")
            return ticker_df
    except Exception as e:
        print(f"❌ 请求 {ticker} 失败: {str(e)}")
        return None

if __name__ == "__main__":
    start_time = datetime.datetime.now()
    
    # 初始化线程池，max_workers可根据网络情况调整（建议20-50，避免触发API限流）
    with concurrent.futures.ThreadPoolExecutor(max_workers=30) as executor:
        # 批量提交所有币种的请求任务，map按顺序返回结果
        all_results = executor.map(fetch_single_ticker_data, tickers)
    
    # 过滤请求失败的结果，合并所有有效DataFrame
    valid_dataframes = [df for df in all_results if df is not None]
    if valid_dataframes:
        combined_df = pd.concat(valid_dataframes, ignore_index=True)
        # 一次性写入CSV，避免多线程写文件的内容错乱
        combined_df.to_csv('all.csv', encoding="utf-8-sig", index=False)
        print(f"\n🎉 所有数据已写入all.csv，总耗时: {datetime.datetime.now() - start_time}")
    else:
        print("\n😮‍💨 没有获取到任何有效数据，请检查网络或API地址")

关键细节说明 #

1. 线程池选择原因：网络请求大部分时间在等待服务器响应，线程切换的开销远小于进程，IO密集型任务用线程池效率更高。如果后续有大量CPU计算，再考虑ProcessPoolExecutor，用法几乎一致。
2. API限流规避：Bittrex的API有请求频率限制，别把max_workers设得过高（比如超过50）。如果遇到限流，可以在fetch_single_ticker_data里加短暂延迟（如time.sleep(0.1)），或者降低线程数。
3. 错误容错机制：每个请求都加了异常捕获，单个币种请求失败不会导致整个程序崩溃，还能打印错误信息方便排查问题。
4. 文件写入安全：采用先收集所有结果再统一写入的方式，避免多线程同时操作文件导致的内容错乱。如果数据量超大担心内存不足，也可以用threading.Lock实现线程安全的分批写入，但统一写入更简单高效。

可选优化：实时监控进度 #

如果需要实时看到每个任务的完成顺序，可以用as_completed替代map：

with concurrent.futures.ThreadPoolExecutor(max_workers=30) as executor:
    # 提交任务并保存future对象与币种的映射
    future_to_ticker = {executor.submit(fetch_single_ticker_data, ticker): ticker for ticker in tickers}
    for future in concurrent.futures.as_completed(future_to_ticker):
        ticker = future_to_ticker[future]
        try:
            _ = future.result()
        except Exception as e:
            print(f"{ticker} 处理失败: {e}")

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