1. 为什么需要批量获取Tushare数据Tushare作为国内知名的金融数据接口为量化投资、金融分析等领域提供了丰富的数据支持。但在实际使用中我们经常会遇到这样的场景需要获取某只股票过去5年的日线数据或者需要同时下载多只股票的财务指标。这时候单次请求的方式就显得效率低下了。我曾在开发一个多因子选股系统时需要获取沪深300成分股过去3年的日线行情数据。如果按单只股票逐个请求的方式完成全部数据获取需要近2个小时。而通过合理的批量获取和并发控制这个时间可以缩短到10分钟以内。2. Tushare批量数据获取的核心方法2.1 使用pro_bar接口批量获取行情数据Tushare的pro_bar接口支持批量获取多只股票的行情数据。这是最直接的批量获取方式import tushare as ts # 初始化pro接口 pro ts.pro_api(你的token) # 批量获取多只股票的日线数据 df pro.daily(ts_code600519.SH,000858.SZ, start_date20200101, end_date20201231)这个方法的优点是简单直接但有两个限制一次最多只能获取3000条记录不同股票的数据会混合在一个DataFrame中需要后续处理2.2 使用股票列表循环获取更灵活的方式是准备一个股票列表然后循环获取stock_list [600519.SH, 000858.SZ, 601318.SH] all_data [] for stock in stock_list: df pro.daily(ts_codestock, start_date20200101, end_date20201231) all_data.append(df) result pd.concat(all_data)这种方法虽然代码量稍多但更加灵活可控可以针对每只股票进行个性化处理。3. 并发控制的实现方案3.1 为什么需要并发控制Tushare接口有调用频率限制基础版每分钟最多500次每天最多10000次高级版每分钟最多1000次每天最多100000次如果不加控制地并发请求很容易触发限流导致IP被封禁。我曾经因为没做好并发控制导致API权限被临时冻结过2小时。3.2 使用线程池控制并发Python的concurrent.futures模块提供了方便的线程池实现from concurrent.futures import ThreadPoolExecutor, as_completed import time def get_stock_data(stock): try: df pro.daily(ts_codestock, start_date20200101, end_date20201231) return df except Exception as e: print(f获取{stock}数据失败: {e}) return None # 控制并发数为5 with ThreadPoolExecutor(max_workers5) as executor: futures {executor.submit(get_stock_data, stock): stock for stock in stock_list} for future in as_completed(futures): stock futures[future] try: data future.result() if data is not None: all_data.append(data) except Exception as e: print(f{stock}生成异常: {e}) # 控制请求频率 time.sleep(0.1)3.3 使用信号量控制并发更精细的控制可以使用threading.Semaphoreimport threading semaphore threading.Semaphore(5) # 同时最多5个线程 def get_stock_data_with_semaphore(stock): with semaphore: try: df pro.daily(ts_codestock, start_date20200101, end_date20201231) return df except Exception as e: print(f获取{stock}数据失败: {e}) return None finally: time.sleep(0.1) # 每个请求间隔0.1秒4. 实战中的优化技巧4.1 数据缓存机制频繁请求相同数据既浪费资源又容易触发限流。可以建立简单的缓存机制from functools import lru_cache lru_cache(maxsize100) def cached_get_stock_data(stock, start_date, end_date): return pro.daily(ts_codestock, start_datestart_date, end_dateend_date)4.2 异常处理与重试机制网络请求难免会遇到异常合理的重试机制能提高成功率import random from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def get_stock_data_with_retry(stock): try: df pro.daily(ts_codestock, start_date20200101, end_date20201231) return df except Exception as e: print(f尝试获取{stock}数据失败: {e}) raise4.3 进度显示对于大批量数据获取显示进度很有必要from tqdm import tqdm with ThreadPoolExecutor(max_workers5) as executor: futures {executor.submit(get_stock_data, stock): stock for stock in stock_list} for future in tqdm(as_completed(futures), totallen(stock_list)): stock futures[future] try: data future.result() if data is not None: all_data.append(data) except Exception as e: print(f{stock}生成异常: {e})5. 完整实战案例下面是一个完整的批量获取股票数据并保存到本地的示例import tushare as ts import pandas as pd from concurrent.futures import ThreadPoolExecutor, as_completed import time from tqdm import tqdm import os # 初始化 pro ts.pro_api(你的token) data_dir stock_data os.makedirs(data_dir, exist_okTrue) # 获取沪深300成分股 hs300 pro.hs_const() stock_list hs300[ts_code].tolist() def get_stock_data(stock): try: df pro.daily(ts_codestock, start_date20200101, end_date20201231) if not df.empty: df.to_csv(f{data_dir}/{stock}.csv, indexFalse) return stock, True except Exception as e: print(f获取{stock}数据失败: {e}) return stock, False # 控制并发数为5每秒最多10个请求 with ThreadPoolExecutor(max_workers5) as executor: futures {executor.submit(get_stock_data, stock): stock for stock in stock_list} success_count 0 for future in tqdm(as_completed(futures), totallen(stock_list)): stock, status future.result() if status: success_count 1 time.sleep(0.1) # 控制请求频率 print(f数据获取完成成功获取{success_count}只股票数据)这个案例中我们获取沪深300成分股列表为每只股票创建独立的下载任务使用线程池控制并发数通过sleep控制请求频率将数据保存为单独的CSV文件使用tqdm显示进度6. 性能对比与调优建议在实际测试中我对比了不同并发数下的性能表现并发数获取300只股票数据耗时成功率1约15分钟100%5约5分钟99.7%10约3分钟98.5%20约2分钟92.3%从测试结果可以看出并发数5-10是比较理想的区间超过10并发后成功率明显下降单线程虽然稳定但效率太低我的建议调优策略先从并发数5开始测试根据网络环境和API响应情况逐步提高监控失败率如果超过5%就降低并发数对于重要数据可以适当降低并发保证稳定性7. 常见问题与解决方案7.1 遇到操作太频繁错误怎么办这是最常遇到的问题解决方法立即停止所有请求检查代码中的并发控制和频率限制添加更长的间隔时间如从0.1秒增加到0.5秒分批获取数据不要一次性请求太多7.2 数据获取不完整怎么处理可以采取以下措施记录失败的任务实现自动重试机制对于仍然失败的可以手动补充获取考虑使用Tushare的增量更新接口7.3 如何验证数据的完整性我通常使用以下检查方法检查每只股票的数据条数是否合理检查日期范围是否完整检查关键字段如收盘价是否有缺失值随机抽样几只股票进行人工验证8. 高级技巧分布式数据获取对于超大规模数据获取需求如获取全市场股票10年历史数据可以考虑分布式方案import redis from rq import Queue # 使用Redis队列分发任务 r redis.Redis() q Queue(connectionr) # 将任务放入队列 for stock in stock_list: q.enqueue(get_stock_data, stock)然后可以启动多个工作进程并行处理# 启动5个工作进程 rq worker -n 5这种方案的优点可以水平扩展工作节点任务失败会自动重试可以监控任务进度适合长时间运行的大批量任务我在获取全市场股票20年财务数据时使用这个方案将总时间从预计的3天缩短到了6小时。