本文是「Python量化实战」系列第 16 篇从这篇起进入工程化模块前 15 篇解决数据怎么取、策略怎么写接下来 5 篇解决怎么把它做成一个稳定运转的系统。写量化脚本的人几乎都经历过这个阶段单只股票的数据拉取调通了兴冲冲地套个 for 循环拉全市场——然后发现 5000 多只股票要跑将近一个小时中途网络抖一下还前功尽弃只能从头再来。批量拉取从来不是加个循环那么简单它是一个标准的工程问题并发提速、失败重试、进度可见、结果合并四件事缺一不可。本文用不到 100 行代码把这四件事一次做对实测并发方案比串行提速 6.9 倍且 40 只样本全部成功、0 失败。本文你将得到什么一套线程池并发拉取模板ThreadPoolExecutoras_completed可直接套用到任何接口带指数退避的失败重试装饰逻辑网络抖动不再毁掉整批任务实时进度输出 失败清单跑到哪、挂了谁一目了然串行 vs 并发的真实耗时对比数据0.46 秒/只 → 0.07 秒/只线程数怎么定、限频怎么躲的工程经验一、在线体验想先在线试试接口效果打开 API Playground 即可直接调用测试https://mairuiapi.com/playground本文用到两个接口全市场股票列表stock_list和历史 K 线stock_history可以先在 Playground 里看看返回结构。二、环境准备本文代码使用 mairui SDK 获取股票数据安装方法如下pipinstallmairuiSDK 的完整接口文档与使用说明请查阅 GitHub 仓库https://github.com/MaiRuiApi/mairui接口的详细参数说明请查阅官网 API 文档https://mairuiapi.com/hsdata运行环境Python 3.9concurrent.futures为标准库无需额外安装mairui SDK 1.0.0pandas 2.2importosimportmairui# 证书从环境变量读取官网注册后获取不要写死在代码里apimairui.Client(LICENCE-66D8-9F96-0C7F0FBCD073)# 证书从环境变量读取本文数据截至 2026-07-28拉取区间 2026-06-01 至 2026-07-28读者复现时数据可能略有差异。三、先拿到任务清单全市场股票列表批量任务的第一步是确定任务边界。stock_list一次返回全市场 A 股列表stock_listapi.stock_list()print(f全市场 A 股数量:{len(stock_list)})# 每个元素形如 {dm: 000001.SZ, mc: 平安银行, jys: SZ}真实运行输出全市场 A 股数量: 52055205 只股票就是我们的任务全集。本文演示取前 40 只把样本换成全量列表代码完全不用改只是跑得久一点。四、单只拉取函数先把重试做进去并发的前提是单任务函数足够健壮。网络请求天然会失败——超时、抖动、瞬时限频——正确姿势不是祈祷不失败而是失败了自动重试并且一次比一次等得久指数退避importtimeimportpandasaspd MAX_RETRIES3# 单只失败最大重试次数RETRY_BACKOFF1.5# 退避基数第 k 次重试等待 1.5^k 秒ST,ET20260601,20260728# 拉取区间st/et 格式为 YYYYMMDD不带横杠deffetch_one(api,code,name):拉单只股票日K带重试与指数退避。失败到底则抛出最后一次异常。last_excNoneforattemptinrange(MAX_RETRIES):try:klineapi.stock_history(code.split(.)[0],d,n,stST,etET)kline_dfpd.DataFrame(kline)kline_df.insert(0,code,code)# 打上股票标识合并后可区分kline_df.insert(1,name,name)returnkline_dfexceptExceptionasexc:last_excexc time.sleep(RETRY_BACKOFF**(attempt1))# 1.5s → 2.25s → 3.4sraiselast_exc关键解释每个 DataFrame 拉下来立刻insert股票代码列——合并之后还能知道每行属于谁这是新手最常漏的一步指数退避的意义如果失败是限频导致的立刻重试只会继续被拒等待时间递增才能自愈。五、并发拉取线程池 as_completed数据拉取是典型的IO 密集型任务时间都花在等网络响应上线程池就是标准答案fromconcurrent.futuresimportThreadPoolExecutor,as_completed MAX_WORKERS5# 线程数控制在 5 以内避免触发接口频率限制defrun_concurrent(api,stocks):并发拉取进度实时输出 失败清单登记。frames,failed[],[]withThreadPoolExecutor(max_workersMAX_WORKERS)aspool:futures{pool.submit(fetch_one,api,s[dm],s[mc]):sforsinstocks}done0forfutinas_completed(futures):# 谁先完成先处理谁stockfutures[fut]done1try:frames.append(fut.result())exceptExceptionasexc:# 重试后仍失败登记不中断整批failed.append((stock[dm],type(exc).__name__))ifdone%100ordonelen(stocks):print(f 进度{done}/{len(stocks)}失败{len(failed)})mergedpd.concat(frames,ignore_indexTrue)ifframeselsepd.DataFrame()returnmerged,failed关键解释as_completed按完成顺序返回结果天然适合做进度条单只失败绝不抛出中断整批而是记入failed清单批量任务跑完后单独补拉失败部分——这是批量工程的基本素养。六、真实耗时对比串行 0.46 秒/只 vs 并发 0.07 秒/只同一台机器、同一网络环境的真实测试2026-07-29 验证Step2 串行基准前 10 只 串行拉取 10 只耗时: 4.6 秒约 0.46 秒/只 Step3 并发拉取40 只5 线程重试上限 3 进度 10/40失败 0 进度 20/40失败 0 进度 30/40失败 0 进度 40/40失败 0 并发拉取 40 只耗时: 2.7 秒约 0.07 秒/只 提速比按单只均摊: 6.9x 失败清单: 无按此速度外推全市场 5205 只串行约 40 分钟5 线程并发约 6 分钟。而且这 6 分钟是带重试保险的 6 分钟——中途抖动自动兜住不再需要从头重跑。合并结果同样一步到位合并后 DataFrame: 1640 行 x 11 列覆盖 40 只股票 code name a c h l o pc sf t v 000001.SZ 平安银行 1042306455.0 10.99 10.99 10.81 10.90 10.93 0 2026-06-01 954596 000001.SZ 平安银行 978159336.0 11.08 11.10 10.94 10.98 10.99 0 2026-06-02 885428七、完整可运行示例# -*- coding: utf-8 -*-importosimporttimefromconcurrent.futuresimportThreadPoolExecutor,as_completedimportpandasaspdimportmairui SAMPLE_SIZE40MAX_WORKERS5MAX_RETRIES3RETRY_BACKOFF1.5ST,ET20260601,20260728# fetch_one / run_concurrent 定义见上文第四、五节defmain():apimairui.Client(LICENCE-66D8-9F96-0C7F0FBCD073)# 证书从环境变量读取stock_listapi.stock_list()# ① 任务清单samplestock_list[:SAMPLE_SIZE]# 全市场就把切片去掉merged_df,failedrun_concurrent(api,sample)# ② 并发拉取print(f成功{merged_df[code].nunique()}只失败清单:{failedor无})merged_df.to_csv(batch_kline.csv,indexFalse,encodingutf-8-sig)# ③ 落盘if__name____main__:main()八、避坑与进阶坑 1线程开得越多越快不是。接口侧有频率限制线程数超过阈值后失败率飙升重试反而拖慢整体。实测 3~5 线程是稳定与速度的平衡点。坑 2用多进程做 IO 任务。multiprocessing适合 CPU 密集型计算拉数据是 IO 等待线程池更轻、共享内存更方便。坑 3合并时索引错乱。pd.concat记得ignore_indexTrue否则各 DataFrame 的行索引会重复。坑 4失败任务无记录。批量任务必须产出失败清单跑完针对性补拉而不是整批重来。进阶方向拉下来的数据往哪存CSV 会越来越慢——下一篇讲 CSV/SQLite/MySQL 三种存储方案的对比与选型让 5000 只股票的历史数据存得下、查得快、能增量。九、总结与延伸批量拉取的工程要点浓缩成一句话并发提速、重试兜底、进度可见、失败可补。这套模板不只适用于 K 线——财务数据、实时快照、公告列表任何按代码逐只拉取的场景都能直接套用。配合稳定的数据接口全市场级的数据任务从跑一次要祈祷变成每天定时无人值守。延伸阅读在线体验更多接口https://mairuiapi.com/playground查看完整 API 文档https://mairuiapi.com/hsdataSDK 文档与源码https://github.com/MaiRuiApi/mairui关注公众号获取本系列更新本文为技术演示不构成投资建议。