量化数据的 batch 接口有多好用从 1 只到 500 只批量拉数据的正确姿势做量化最烦的事之一一只一只地拉数据。# 你可能写过这种代码importtime results{}forsyminmy_500_stocks:try:dfsome_api.get_klines(sym)results[sym]dfexceptException:passtime.sleep(0.5)# 怕被封# 500 只票 × 0.5 秒 250 秒 4 分钟起步循环、sleep、try/except、进度不可见、失败了不知道哪些没拉到。AlphaFeed 的batch接口把这些全解决了传一个列表进去内部自动分块、并发、重试几秒钟全部拉完。这篇文章详细讲 batch 接口的用法、原理和实战技巧。最基本的用法fromalphafeedimportAlphaFeed afAlphaFeed()symbols[600519.SH,000001.SZ,300750.SZ,002594.SZ,601318.SH]# 一行搞定dfsaf.klines.batch(symbols,period1d,count250,adjustforward,to_dataframeTrue,)# 返回字典{symbol: DataFrame}forsym,dfindfs.items():print(f{sym}:{len(df)}条数据)就这么简单。没有循环、没有 sleep、没有 try/except。batch 内部做了什么当你调用af.klines.batch(symbols, ...)时SDK 内部做了这些事你传入 500 个标的 ↓ 自动分块每 100 个一组分成 5 组 ↓ 5 个线程并发请求不是串行 ↓ 每个请求自动重试最多 3 次遇到超时或 5xx 自动重试 ↓ 合并所有结果返回一个字典关键参数默认分块大小100 个标的/请求默认并发数5 个线程默认重试3 次这意味着 500 只票分成 5 组5 个线程同时请求——时间约等于请求 1 组的时间而不是请求 5 组的时间。显示进度条加上show_progressTrue拉数据的过程可视化dfsaf.klines.batch(symbols,period1d,count500,adjustforward,to_dataframeTrue,show_progressTrue,# 显示 tqdm 进度条)Fetching data: 100%|████████████████████| 5/5 [00:0300:00, 1.67it/s]进度条显示的是分块的进度。500 只票分 5 块进度条就是 5 步。实战 1批量拉 K 线做对比分析同时拉 20 只票的 K 线算过去一年的涨幅和波动率importpandasaspdfromalphafeedimportAlphaFeed afAlphaFeed()symbols[600519.SH,000001.SZ,300750.SZ,002594.SZ,601318.SH,000858.SZ,600036.SH,000333.SZ,601012.SH,600276.SH,600900.SH,601398.SH,600030.SH,000651.SZ,002415.SZ,600887.SH,601166.SH,000568.SZ,600809.SH,002304.SZ,]dfsaf.klines.batch(symbols,period1d,count250,adjustforward,to_dataframeTrue,show_progressTrue,)results[]forsym,dfindfs.items():dfdf.sort_values(trade_date).reset_index(dropTrue)retdf[close].iloc[-1]/df[close].iloc[0]-1voldf[close].pct_change().std()*(252**0.5)avg_amountdf[amount].mean()results.append({代码:sym,年涨幅:f{ret:.1%},年化波动:f{vol:.1%},日均成交额(亿):f{avg_amount/1e8:.1f},})result_dfpd.DataFrame(results).sort_values(年涨幅,ascendingFalse)print(result_df.to_string(indexFalse))20 只票的 250 天数据几秒钟拉完。实战 2批量拉分时数据不只日线分时数据也支持批量fromalphafeedimportAlphaFeed afAlphaFeed()symbols[600519.SH,000001.SZ,300750.SZ]# 批量拉日内 1 分钟线dfsaf.klines.intraday_batch(symbols,to_dataframeTrue)forsym,dfindfs.items():morning_voldf[df[trade_time]df[trade_time].iloc[0][:11]11:30:00][volume].sum()total_voldf[volume].sum()pctmorning_vol/total_voliftotal_vol0else0print(f{sym}: 上午成交占比{pct:.1%}总{len(df)}根分钟线)实战 3批量查五档盘口fromalphafeedimportAlphaFeed afAlphaFeed()symbols[600519.SH,000001.SZ,300750.SZ,002594.SZ]# 批量查盘口depthsaf.depth.batch(symbols)print(f{标的:12}{买一:10}{卖一:10}{价差(bps):10})print(-*45)forsym,dindepths.items():bid1d[bid_prices][0]ask1d[ask_prices][0]mid(bid1ask1)/2spread_bps(ask1-bid1)/mid*10000ifmid0else0print(f{sym:12}{bid1:10.2f}{ask1:10.2f}{spread_bps:10.1f})一次请求拿到多只票的盘口适合做流动性分析或估算交易成本。实战 4批量查标的信息fromalphafeedimportAlphaFeed afAlphaFeed()# 查一批标的的基本信息symbols[600519.SH,000001.SZ,AAPL.US,00700.HK]instsaf.instruments.batch(symbols)forinstininsts:extinst.get(ext,{})print(f{inst[symbol]:12s}{inst[name]:8s}f地区{inst[region]}类型{inst[type]}f上市{ext.get(listing_date,N/A)})实战 5批量查复权因子fromalphafeedimportAlphaFeed afAlphaFeed()symbols[600519.SH,000001.SZ,000858.SZ,002594.SZ]dfaf.klines.ex_factors(symbols,to_dataframeTrue)# 每只票有多少次除权除息forsyminsymbols:subdf[df[symbol]sym]print(f{sym}:{len(sub)}次除权除息)实战 6200 只票的策略批量回测batch 的真正威力在大规模分析。比如对 200 只票跑同一个策略importpandasaspdimportnumpyasnpfromalphafeedimportAlphaFeed afAlphaFeed()# 假设你有一个 200 只票的股票池stock_pool[600519.SH,000001.SZ,300750.SZ,002594.SZ,601318.SH,000858.SZ,600036.SH,000333.SZ,601012.SH,600276.SH,# ... 省略实际上你可以放 200 只]# 一次性拉取所有数据print(正在拉取数据...)dfsaf.klines.batch(stock_pool,period1d,count500,adjustforward,to_dataframeTrue,show_progressTrue,)print(f拉取完成:{len(dfs)}只票)# 对每只票跑双均线策略defbacktest_ma(df:pd.DataFrame)-dict:dfdf.sort_values(trade_date).reset_index(dropTrue)df[ma20]df[close].rolling(20).mean()df[ma60]df[close].rolling(60).mean()df[signal](df[ma20]df[ma60]).astype(int)df[position]df[signal].shift(1).fillna(0)df[ret]df[close].pct_change().fillna(0)df[strat_ret]df[position]*df[ret]equity(1df[strat_ret]).cumprod()return{total_return:equity.iloc[-1]-1,max_drawdown:(equity/equity.cummax()-1).min(),}results[]forsym,dfindfs.items():iflen(df)100:continuerbacktest_ma(df)r[symbol]sym results.append(r)rdfpd.DataFrame(results).sort_values(total_return,ascendingFalse)print(f\n{len(rdf)}只票回测完成 )print(f盈利的:{(rdf[total_return]0).sum()}只)print(f亏损的:{(rdf[total_return]0).sum()}只)print(f平均收益:{rdf[total_return].mean():.2%})print(f\n表现最好的 5 只:)print(rdf.head()[[symbol,total_return,max_drawdown]].to_string(indexFalse))print(f\n表现最差的 5 只:)print(rdf.tail()[[symbol,total_return,max_drawdown]].to_string(indexFalse))重点200 只票的 500 天 K 线用 batch 拉取只需要几秒钟。拉数据不再是瓶颈你可以把时间花在策略分析上。batch vs 手动循环代码对比手动循环传统方式importtime results{}failed[]fori,syminenumerate(symbols):try:dfsome_api.get_klines(sym,period1d,count500)results[sym]dfprint(f[{i1}/{len(symbols)}]{sym}OK)exceptExceptionase:print(f[{i1}/{len(symbols)}]{sym}失败:{e})failed.append(sym)time.sleep(0.5)# 怕被封# 重试失败的forsyminfailed:try:dfsome_api.get_klines(sym,period1d,count500)results[sym]dfexceptException:print(f{sym}重试仍失败)time.sleep(1)print(f成功:{len(results)}/{len(symbols)})AlphaFeed batchdfsaf.klines.batch(symbols,period1d,count500,adjustforward,to_dataframeTrue,show_progressTrue,)一行代码替代了上面的 20 行。分块、并发、重试、进度显示全内置。batch 的技术细节参数默认值说明分块大小100 只/请求SDK 自动把标的列表切成 100 个一组并发数5 个线程5 组同时请求不是排队重试次数3 次超时、网络错误、5xx 自动重试进度条show_progressTrue显示基于 tqdm显示分块级进度为什么分块而不是一个请求传 500 个标的因为 URL 有长度限制500 个标的的代码拼成逗号分隔的字符串会超过 HTTP URL 的安全长度。SDK 帮你处理了这个问题。哪些接口支持 batch接口单个查询批量查询K 线af.klines.get(symbol)af.klines.batch(symbols)日内分时af.klines.intraday(symbol)af.klines.intraday_batch(symbols)实时行情af.quotes.get(symbols[...])本身就支持多标的五档盘口af.depth.get(symbol)af.depth.batch(symbols)标的信息af.instruments.get(symbol)af.instruments.batch(symbols)复权因子—af.klines.ex_factors(symbols)和竞品的对比操作AlphaFeedakshareTushare拉 100 只票日 Kaf.klines.batch(symbols)一行搞定手动 for 循环 sleep手动 for 循环 token 配额自动重试SDK 内置需要自己写 try/except需要自己写并发请求SDK 内置 5 线程不支持爬虫会被封有频率限制进度显示show_progressTrue自己用 tqdm 包一层自己实现500 只票耗时约 5-10 秒5 分钟含 sleep1-3 分钟含频率等待结语batch 接口是 AlphaFeed 最实用的功能之一。它解决的问题很朴素当你需要分析很多只票的时候不应该等很久、写很多错误处理代码。一个列表传进去进度条跑一下数据就到手了。剩下的时间你可以专注在分析和策略上而不是和网络请求搏斗。相关链接AlphaFeed 官网https://alphafeed.org/Python SDK 快速开始https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart使用示例https://docs.alphafeed.org/zh-Hans/sdk/python-examples