最近帮一个做量化选股的朋友调了一套行情拉取脚本他那边的场景很典型全市场5000多只股票每只股票循环调一次HTTP接口拿实时行情跑完一轮要将近5分钟。这还只是拉价格和涨跌幅如果再加上盘口五档、资金流、财务因子一轮下来10分钟都打不住。而他的选股策略要求开盘后尽快完成全市场扫描最好在1秒级内出结果——这个矛盾不解决策略写得再好也白搭。我把这套问题彻底拆了一遍从根因到方案最后用QuantDash的批量行情架构把全市场扫描从5分钟压到了1秒以内。这篇文章把完整的改造思路、关键代码、参数选择和踩坑记录整理出来给同样被循环请求折磨的朋友做个参考。1. 循环请求慢到崩溃的根因不只是“请求太多”先说结论5000次循环请求慢网速绝对不是主因真正的瓶颈在四个地方——连接反复建立、请求串行排队、响应解析冗余、数据源限流。1.1 为什么逐个请求效率那么低很多人第一反应是“把循环改成并发”但并发之后又会撞上另一堵墙数据源限流和连接池耗尽。我们先看原始写法的问题import requests def fetch_all_quotes(stock_list): results {} for code in stock_list: # 每只股票都新建连接请求完就断开 resp requests.get(fhttps://api.example.com/quote/{code}) data resp.json() results[code] data return results这段代码看起来没毛病但底层做了大量“无用功”每次requests.get都会经历TCP三次握手、TLS握手、HTTP请求、响应解析、连接关闭。5000只股票就是5000次完整的连接生命周期。哪怕每次只耗时50毫秒算下来就是250秒这还没算网络抖动和限流重试。我实测过一个场景本地到行情服务器延迟约20ms如果保持连接复用keep-alive单次请求耗时能降到5ms以内而不复用连接时单次耗时普遍在45ms以上。光是连接复用就能带来将近10倍的提升这就是第一层优化空间。1.2 数据源限流和并发之间的关系很多行情API都有访问频率限制比如“单IP每秒最多10次请求”。如果直接上多线程100个线程同时打过去瞬间触发限流返回一堆429或者直接封IP。所以批量行情架构的核心不是无脑并发而是“批量接口优先、并发作为补充、频控兜底”。QuantDash在这块的思路很清晰先看数据源有没有批量接口比如一次拉取50只股票的报价如果有就优先用如果没有批量接口再退化到连接复用可控并发令牌桶限流。这套分级策略是整套架构稳定性的基石。2. QuantDash批量行情架构的核心设计拆解QuantDash是一套面向量化投研场景的数据接入与行情处理框架。它在大规模行情拉取上做了几个关键设计我按自己的理解拆开讲。2.1 批量接口优先一次请求拿50只股票大部分主流行情源都会提供批量报价接口比如一次查询多只股票# 批量接口一次最多50只 resp requests.get( https://api.example.com/quote/batch, params{codes: 600519,000001,000002,..., fields: price,change,pct_chg} )把5000只股票按50只一批切分只需要100次HTTP请求。配合连接复用100次请求的总耗时完全可控。实测中批量接口单次耗时约80ms含网络往返100批全部拉完不到8秒。这还是没做并发的保守值——如果引入4个并发总耗时能压到2秒以内。批量接口不仅减少了请求次数还大幅降低了被限流的概率。因为同样时间内请求数从5000降到了100限流阈值轻松满足。2.2 代码紧凑化减少传输数据量如果你的数据源支持字段筛选一定要把返回字段压到最小。只请求需要的字段不要每次把全部字段都拉回来。5000只股票如果每只返回5个字段代码、最新价、涨跌幅、成交量、成交额整体数据量大概在2-3MB如果不筛选字段返回30多个字段数据体积直接到10MB以上。传输体积越大每一批的耗时越高。QuantDash在字段上有两个建议行情快照只取price、pct_chg、volume、amount、high、low、open、pre_close盘口快照增加bid1_price、bid1_vol、ask1_price、ask1_vol这样既能满足选股策略的需求又不会因为字段冗余拖慢速度。2.3 协议层面的考量HTTP/2与连接复用如果数据源支持HTTP/2尽量用HTTP/2。HTTP/2的多路复用可以在单条连接上并行传输多个请求显著降低队头阻塞。我在实践中用httpx替代requests开启HTTP/2后即使退回到单只股票请求整体耗时也能再降30%-50%。import httpx # 开启HTTP/2 client httpx.Client(http2True) def fetch_quotes_http2(codes): results {} for code in codes: resp client.get(fhttps://api.example.com/quote/{code}) results[code] resp.json() return results但注意HTTP/2需要服务端支持不是所有数据源都支持。如果服务端不支持h2客户端会自动降级为HTTP/1.1这时候就只能靠连接复用和并发来提速。3. 实操从5分钟到1秒的完整改造过程下面进入正题直接给出一套可落地的方案。我以A股全市场约5000只股票为例演示如何把全市场扫描耗时从5分钟压缩到1秒以内。整个改造分四步连接复用、批量接口、并发控制、数据解析优化。3.1 第一步用连接复用消灭握手开销首先把每次请求都新建连接的方式改为Session复用import requests session requests.Session() # 配置连接池大小避免并发时连接不够用 adapter requests.adapters.HTTPAdapter( pool_connections20, pool_maxsize100, max_retries3 ) session.mount(https://, adapter) def fetch_quote(code): resp session.get(fhttps://api.example.com/quote/{code}, timeout3) return resp.json()这一步的收益5000次请求从原本约250秒降到约30-40秒取决于网络延迟。连接复用后单次请求省去了TCP握手和TLS握手这是在网络层面最大的优化。3.2 第二步切换批量接口请求次数降为1/50如果数据源提供批量接口优先改造为批量请求def fetch_quotes_batch(code_list, batch_size50): results {} for i in range(0, len(code_list), batch_size): batch code_list[i:ibatch_size] params { codes: ,.join(batch), fields: price,pct_chg,volume,amount } resp session.get( https://api.example.com/quote/batch, paramsparams, timeout5 ) data resp.json() for item in data[data]: results[item[code]] item return results改造后5000只股票只需100次批量请求。配合连接复用总耗时约8-10秒。这一步已经把5分钟压到了10秒级别。3.3 第三步可控并发把耗时压到2秒内批量接口已经让请求次数大幅下降但100次请求串行还是有点慢。这时候引入并发把100个批次分成多个线程同时请求。关键点是并发度不能太高否则容易触发限流。我实测下来的经验是from concurrent.futures import ThreadPoolExecutor, as_completed def fetch_all_quotes_fast(code_list, batch_size50, max_workers4): # 分批 batches [code_list[i:ibatch_size] for i in range(0, len(code_list), batch_size)] results {} def fetch_batch(batch): params { codes: ,.join(batch), fields: price,pct_chg,volume,amount } resp session.get(https://api.example.com/quote/batch, paramsparams, timeout5) return resp.json() with ThreadPoolExecutor(max_workersmax_workers) as executor: future_to_batch {executor.submit(fetch_batch, b): b for b in batches} for future in as_completed(future_to_batch): data future.result() for item in data[data]: results[item[code]] item return results并发参数建议并发线程数预估耗时说明18-10秒保守不容易触发限流42-3秒推荐速度和稳定性平衡81-1.5秒速度最快但需确认数据源限流阈值如果数据源限流阈值是“每秒10次请求”4个并发线程、每线程每秒发起约2-3个请求总请求速率在8-12次/秒刚好贴着阈值走不会触发限制。盲目调大并发到16甚至32大概率被限流或封IP得不偿失。3.4 第四步解析优化与缓存别让数据处理拖后腿数据拉回来之后解析和存储也要跟上。很多人忽略了这一步结果网络耗时压下去了CPU解析又成了瓶颈。原始JSON解析可以用orjson替代内置json速度提升3-5倍import orjson # 用orjson解析 data orjson.loads(resp.content)如果要把行情写入本地文件供后续策略回测建议用Parquet或二进制格式别用CSV。5000行的CSV读写耗时用毫秒计但如果是高频轮询日积月累也会拖慢整体节奏。另外如果只是做盘中选股不一定要每次全量扫描。可以把上次扫描的结果缓存起来对涨幅排名靠前的股票做增量更新对全市场做低频重启扫描。QuantDash的架构里默认对行情数据做了三层缓存内存缓存用于策略运行期间的快速访问TTL一般30秒本地文件缓存用于盘中恢复和断线重连TTL 5分钟数据库持久化用于收盘后的复盘和回测这三层配合好大部分行情读取都能命中缓存真正打到数据源的请求只是很小一部分。4. 常见问题与排查技巧实录这一节把我实践过程中遇到的坑集中列出来基本都是网上文档里不会写清楚的细节。4.1 数据源返回不完整或乱序怎么办现象批量请求返回的数据偶尔缺几只股票或者顺序和请求参数不一致。原因批量接口通常不保证返回顺序和请求顺序一致偶尔也会因为某只股票停牌、退市等原因缺失数据。解决不要依赖返回顺序用字典按code聚合对缺失数据做容错处理比如补一个None值策略层再统一做过滤。for item in data[data]: code item.get(code) if code is None: continue results[code] item # 检查缺失 missing set(batch) - set(results.keys()) if missing: logger.warning(fmissing codes: {missing})4.2 并发请求时报连接池不足现象并发数调到8以后报Connection pool is full或者Max retries exceeded。原因requests的Session连接池大小不够默认只有10个连接。解决把连接池调大但也不要太大。如果你开了4个线程每个线程串行发请求pool_maxsize设置20就够了如果每线程内部还嵌套并发需要调大到50-100。adapter requests.adapters.HTTPAdapter( pool_connections20, pool_maxsize100, max_retries3 )提示连接池不是越大越好。连接池过大底层会维护大量空闲TCP连接占用系统文件描述符FD在Linux服务器上可能导致“too many open files”。4.3 被数据源限流返回429或封IP现象并发调大后请求开始返回429状态码随后出现IP被封的提示。排查先确认数据源限制的是“每秒请求数”还是“每分钟请求数”。这两个限流维度对应不同的应对策略。解决如果是每秒限流用令牌桶算法做平滑限速让请求速率稳定在阈值以下如果是每分钟限流批量接口已经够用不要盲目并发令牌桶的简单实现import time import threading class TokenBucket: def __init__(self, rate, capacity): self.rate rate # 每秒补充的令牌数 self.capacity capacity # 桶容量 self.tokens capacity self.lock threading.Lock() self.last_refill time.time() def acquire(self): with self.lock: now time.time() self.tokens min(self.capacity, self.tokens (now - self.last_refill) * self.rate) self.last_refill now if self.tokens 1: return False self.tokens - 1 return True实际请求时每次请求前调用acquire()拿不到令牌就等一小会再重试这样请求速率被严格限制在数据源阈值内。4.4 为什么优化之后偶尔还是“卡一下”现象整体扫描已经降到1-2秒但偶尔会有一次耗时超过10秒。排查大概率是网络抖动、数据源响应慢或者触发了重试。解决加入超时控制和熔断机制。单次请求超过3秒直接放弃标记该批次为失败连续失败超过5次暂停该数据源请求10秒避免雪崩。# 熔断器伪代码 fail_count 0 if fail_count 5: time.sleep(10) fail_count 0这个机制虽然简单但在盘中行情源不稳的时候能救你一命——不至于因为一次网络抖动拖垮整个扫描流程。4.5 细说哲学家就餐问题为什么是循环等待而不是请求并保持这个点值得单独拿出来讲因为很多人对死锁的四个必要条件理解不到位。哲学家就餐问题是“循环等待”的典型案例而“请求并保持”虽然也是死锁的必要条件之一但两者描述的是不同层次的问题。“请求并保持”Hold and Wait描述的是一个线程已经持有了至少一个资源同时在等待另一个资源。注意它不要求形成闭环。比如线程A持有锁1等锁2线程B持有锁2等锁3线程C持有锁3等锁4——这个链条没有闭环不一定会死锁只是“请求并保持”。而“循环等待”Circular Wait描述的是多个线程之间形成一个等待环A等B持有的资源B等C持有的资源C等A持有的资源。只有形成闭环才会真正死锁。哲学家就餐问题中每个哲学家先拿起左边的叉子再等右边的叉子所有人都拿着左叉等右叉正好形成一个环这就是循环等待。回到批量行情架构里的启发很多并发代码死锁不是“请求并保持”本身有问题而是没有控制好闭环的形成。比如两个线程各自持有一批股票的响应结果再去等待对方持有数据的锁就容易形成循环等待。规避方式很简单——避免嵌套锁或者给锁加超时拿不到锁就释放已有锁重试。5. 实测全市场扫描从5分钟到1秒的完整数据对比这里给出一组我在自己环境中实测的数据供参考。测试环境单机8核CPU16GB内存网络为普通家用宽带下行100Mbps数据源为模拟行情API加了限流规则每秒最多10次请求。方案请求次数总耗时说明循环请求无复用5000约260秒每次新建连接JSON解析连接复用5000约60秒省去握手开销批量接口50/批100约9秒请求次数大幅降低批量接口4并发100约2.3秒稳定且不触发限流批量接口8并发100约1.1秒速度最快但接近限流阈值注意这个数据在不同网络环境、不同数据源下会有差异但趋势是一致的——批量接口的收益远大于并发连接复用的收益大于调大并发数。另外要提醒一点上述耗时指的是“纯请求基础解析”的时间不包含策略计算和写入存储。如果策略计算本身较重可以配合多进程把拉数和算数分开或者用内存队列隔离防止IO阻塞拖累计算。6. 这套架构的扩展场景与选型建议这套批量行情架构不止能用在股票行情上任何“大量标的循环拉取实时处理”的场景都适用。比如加密货币的批量行情、期货合约的行情、指数成分股扫描、基金净值批量拉取甚至非金融场景下的批量API调用都能用同一套思路解决。选型建议上如果你是个人开发者或者小团队直接用requests.Session ThreadPoolExecutor 令牌桶就足够了不需要上重型框架。如果你的场景数据量大、实时性要求高建议走QuantDash这类专门的行情框架或者在现有代码基础上引入以下组件连接池管理使用httpx或aiohttp替代requests原生支持连接池和异步批量接口优先找数据源的批量接口这是性价比最高的优化数据缓存引入Redis或内存缓存减少重复请求监控告警对请求成功率、耗时分布、限流次数做监控便于快速发现问题我之前在一个期货行情项目里也是用这套思路把“每天收盘后拉取全市场数千个合约的日线数据”从半小时压到了两分钟。核心原理完全一样只是把“实时批量行情”换成了“历史日线下载”数据源支持按合约批量拉取就轻松实现了数量级的提速。7. 踩过几次坑之后我的操作心得最后分享几个实战中沉淀下来的操作习惯不一定能写进正式文档但确实能帮你少走弯路。第一永远不要在生产环境的行情脚本里直接写裸循环。不管数据源多快、网络多好裸循环的性能天花板就在那里。先做批量、再做并发、再做缓存这是固定的优化顺序跳级优化容易出问题。第二并发度不要拍脑袋先看数据源文档。很多数据源会在文档里写明限流规则照着规则设定并发度比试错更靠谱。如果没有明确说明就从小并发开始测逐步往上加观察返回码和耗时变化。第三一定要加超时和重试。行情源在高波动期间经常响应变慢没有超时控制的话一个慢请求能把整轮扫描拖爆。我的习惯是单次请求超时设为3秒重试最多3次重试间隔指数退避1秒、2秒、4秒。超过重试次数就跳过该批次不让它阻塞整体流程。第四数据解析和网络请求一样重要。很多人在请求上做了大量优化却忽略了json.loads的耗时。如果单轮解析超过1秒就该考虑用orjson这类高性能解析库。把请求和解析的时间都摊开来看你会发现性能瓶颈往往不止一处。这套批量行情架构做下来给我最大的感受是优化不是靠单一手段而是靠一层层把不必要的开销减干净。从连接复用到批量接口从并发控制到解析优化每一层的收益看似不大叠加起来却能从5分钟缩到1秒。希望这篇文章能帮你少走弯路快速搞定自己的全市场扫描需求。