实时股票数据API接口与历史行情查询API接口选型验证与工程化避坑

做量化策略、搭行情看板、写后台盯盘脚本绕来绕去都会撞上同一堵墙实时股票数据API接口从哪来股票历史行情查询API接口怎么批量拉。我最初接触这块时被标题里带大全两个字的文章坑过不止一次——点进去是一长串链接一半已经失效一半要先注册再申请密钥剩下能用的几个还没写清楚返回字段的时区接进去之后每天差八小时。后来自己攒了一份能逐个验证的清单也踩了不少坑这篇就把实时行情与历史行情这两条主线完整摊开讲。这篇东西适合三类人正在做策略回测、需要批量历史K线的要做实时看板或者预警、需要快照或推送的以及刚接手一个数据模块、需要快速判断这个接口到底能不能用的。我会把接口分类逻辑、验证方法、复权口径、断线补数、限频处理、落库结构全部写清楚代码可以直接改改就跑。所有示例里的域名都是占位结构你把域名换成实际接入服务商给的地址即可参数名和返回字段的处理逻辑是通用的。1. 先搞清楚行情接口到底分几类再谈选型1.1 快照轮询、推送流、批量文件三种数据流的本质区别很多人上来就问哪个接口最好这个问题本身就没法回答因为好取决于你要的是哪一种数据流。行情数据在传输形态上只有三类认清这三类是选型的第一步。第一类是快照接口本质是一次HTTP GET你请求一个代码它返回这一刻的最新价、开高低收、成交量、成交额、买卖五档。它天然是拉的模式你想要多新的数据就拉多勤。优点是接入成本极低浏览器地址栏粘上去就能看到JSON缺点是频率受服务端限流约束一般免费档限制在每分钟几十到几百次全市场几千只票想秒级刷新基本不现实。适合的场景是小范围自选股监控、中小规模的看板。第二类是推送流通常是WebSocket长连接服务端主动把逐笔成交或盘口变化推给你。优点是延迟低、不用反复发请求、没有轮询浪费缺点是要处理连接生命周期——心跳、断线、重连、重连之后那段空窗期的数据补全。这块是新手最容易忽略的地方代码写完跑起来看着没问题一觉醒来发现连接早断了三个小时中间的数据全空着。第三类是批量文件或批量查询接口适合一次性把全市场某一天的数据拉下来或者按日期区间拉单只票的历史。它不追求实时性追求的是吞吐量和字段完整度。历史行情回补、全市场因子计算基本都走这条路。我一般建议的做法是实时用推送流兜底用快照历史用批量接口。三条腿配合比死磕一个接口要靠谱得多。1.2 免费与付费的真实分水岭不在价格而在三个参数看了一圈市面上的服务免费档和付费档的差别表面上是价格实际上是三个参数频率上限、历史深度、字段完整度。频率上限决定了你能不能在盘中做有意义的刷新。免费档常见是每分钟 30 到 200 次不等付费档能到每秒几十次甚至不设硬上限。这里有个换算要自己做假设你要盯 200 只票刷新周期 5 秒那一分钟就是 200 × 12 2400 次请求免费档根本撑不住只能改用批量查询一次请求带多个代码或者推送流。历史深度是个更隐蔽的坑。有些服务免费档只给最近一两年的日线你想回测一个横跨五年的策略数据直接就断在半路上。分钟线的深度缩水更严重有的只给最近几个月。选型的时候一定要先确认我要的最早日期它给不给别等写完回测框架才发现数据不够。字段完整度是最容易踩雷的。同一个日线有的只给开高低收和成交量有的给成交额、换手率、复权因子、涨跌幅、前后收盘。差一个复权因子你就得自己去推算除权日这个工作量比接十个接口还大。提示评估一个数据源别只看它首页写的字段列表一定要拿一只发生过送转的票把除权日前后的数据拉出来对比看它给的是原始价还是调整后价。这一条能筛掉一大半看起来很美的源。1.3 动手前先问自己四个问题能省掉一半返工我在帮人排查数据问题时发现返工大多源于动手前没想清楚。接行情接口之前这四个问题建议先写下来频率盘中我需要多快的数据秒级、分钟级、还是收盘后一次性拿秒级就必须考虑推送流分钟级快照轮询足够。范围只盯几十只自选还是要覆盖全市场几千只范围决定了是单代码查询还是批量查询。口径我要的是原始成交价还是复权后价格做技术指标计算和做收益率统计对复权的要求完全不同。留存数据是拉完即用还是要落库长期积累要留存的话表结构和主键设计得提前定后期改表很痛苦。把这四个问题的答案写在需求文档第一页后面所有选型争议都能拿它来裁决。我见过太多项目前期随手接了个免费快照接口做到一半发现需要五年分钟线推倒重来。2. 接口清单的整理逻辑按什么维度分组才不乱2.1 按接入方式分组HTTP、长连接、文件下载网上那些接口大全之所以让人越看越乱是因为它们把所有东西平铺在一张列表里没有维度。我自己的清单是按接入方式做一级分类的因为接入方式直接决定了你的代码结构。HTTP类是最多的包括单代码快照、多代码批量快照、按日期区间查历史、按代码查基本资料。这一类共用一套请求封装写好一个带重试和超时的通用函数所有接口都能套用。长连接类是推送流通常是WebSocket。这一类要单独写一套连接管理逻辑包括建连、鉴权、订阅、心跳、消息解析、断线重连、重连后的补数。它和HTTP类的代码几乎不共用所以我在清单里单独列一栏。文件类是批量下载一般是按交易日打包的压缩包或者分页的批量查询接口。这一类适合离线跑批代码形态是下载—解压—解析—入库和在线请求完全是两套流程。按这个维度分完之后你会发现自己的代码库自然就是三个模块边界清晰维护起来也轻松。2.2 按数据粒度分组Tick、分钟线、日线、复权因子各管一段第二个维度是数据粒度这个维度决定了存储成本。Tick级数据一天全市场就是几个GB分钟线小两个数量级日线几乎可以忽略。同样是历史行情粒度不同架构完全不同。粒度单只票单日数据量估算典型用途存储建议Tick逐笔数千到数万条微观结构研究、滑点分析列式存储按日分区分钟线240条左右日内策略、盘中监控时序库或分区表日线1条中长期策略、因子计算普通关系表即可复权因子每次除权1条价格序列还原单独小表和日线关联这张表的意义在于不要把不同粒度的数据塞进同一张表。我一开始图省事把日线和分钟线放在一个表里用一个freq字段区分结果查询时每次都要带过滤条件索引效率很差后来拆成两张表查询速度直接上来了。复权因子单独建表这个习惯是我踩坑之后养成的。很多数据源把复权后的价格直接给你看似省事但你没法还原原始价也没法做跨源的交叉验证。自己存一份因子前复权后复权随时能算灵活得多。2.3 一张选型对照表把场景和接口对上号下面这张表是我实际在用的对照关系左边是你的需求场景右边是对应的接口类型和注意事项。可以直接拿去当选型清单。使用场景推荐接口形态关键注意点自选股看板十几只票单代码快照5-10秒轮询注意限频做好本地缓存全市场盘中扫描批量快照或多代码合并查询单次代码数量上限、返回是否分页策略回测五年日线按区间批量历史查询复权口径、停牌日是否包含日内策略回测分钟线历史查询数据深度、是否含集合竞价实时预警推送流心跳间隔、断线补数机制全市场因子计算按日批量文件或分页查询文件完整性校验、日期覆盖这张表我贴在项目文档里每次有人问这个需求该用哪个接口直接指过去比口头解释快得多。3. 可直接点击验证的测试方法把能不能用量化3.1 浏览器地址栏就能验证的快照接口先把URL结构拆开看标题里说可直接点击验证测试我理解这句话的真正价值在于一个接口能不能用五秒钟就能判断不用写代码。方法就是把完整的GET请求粘到浏览器地址栏回车看返回。一个典型的行情快照请求结构大致是这样的https://api.example-quote.com/v1/quote?symbol600000.SHfieldslast,open,high,low,vol,amount,ts拆开看四个部分域名加版本路径、资源名quote、代码参数、字段参数。你把域名换成实际服务商给的那个把代码换成一个你确定存在且当天有成交的回车。判断标准很直接返回200并且是一个结构化JSON里面有数字说明这条路通了。返回401或403说明要鉴权需要带密钥返回404路径写错了返回一个HTML页面说明域名或路径不对被重定向到首页了返回空数组大概率是代码格式不对比如缺了交易所后缀或者用了全角字符。这里有个细节值得单独说代码格式。同一个票有的服务要600000.SH有的要sh600000有的要纯数字600000。这三种写法在浏览器里测试时一眼就能看出来因为返回结果会明确告诉你查无此代码。我建议在清单里给每个接口都标注它要求的代码格式写进代码的适配层别让上层业务去关心这个。另外要观察返回的时间戳字段。拿到JSON之后第一件事看ts或者time字段是秒级还是毫秒级是UTC还是本地时间。我遇到过返回东八区时间戳但字段名写成timestamp的也遇到过返回UTC但从注释到文档都没提的直接导致入库时间整体偏移。3.2 用curl和几行Python把返回结构摸透浏览器能验证通不通但要验证结构对不对还得靠命令行和脚本。第一步用 curl 看响应头curl -i https://api.example-quote.com/v1/quote?symbol600000.SHfieldslast,open,high,low,vol,amount,ts重点看三个响应头Content-Type是不是application/json如果是text/html基本就是错页Content-Encoding是不是gzip影响你的客户端要不要解压以及有没有X-RateLimit-Remaining这类限频提示头。有些服务把剩余额度放在响应头里这个信息非常有用可以在代码里读取并动态调整请求节奏。看完头之后再看体把JSON格式化逐个字段对照它的文档。这一步我习惯做一个字段映射表因为不同源的字段名几乎不可能一致import requests URL https://api.example-quote.com/v1/quote PARAMS { symbol: 600000.SH, fields: last,open,high,low,vol,amount,ts } resp requests.get(URL, paramsPARAMS, timeout5) resp.raise_for_status() data resp.json() # 打印完整结构先看清再解析 import json print(json.dumps(data, ensure_asciiFalse, indent2))跑出来之后把你关心的字段逐个对上。我这里列一个通用的映射思路你要的语义常见的字段名候选需要确认的点最新价last / price / close / now是成交价还是买卖中间价开盘价open是否包含集合竞价最高/最低high / low盘中快照是否为当日累计成交量vol / volume单位是股还是手差100倍成交额amount / turnover单位是元还是万元时间戳ts / time / timestamp秒/毫秒、UTC/本地注意成交量单位这个坑非常普遍。同样是12345有的源是12345股有的是12345手即1234500股。判断方法很简单——用成交额除以成交量看看算出来的均价是不是落在当天的最低价和最高价之间。如果差100倍均价就会离谱到明显不合理。3.3 验证清单字段、时区、复权、限频四项必须过我把验证过程固化成一张清单每个新接入的源都要过一遍不过关就不进生产。这张清单四项缺一不可。第一项字段完整性。拿一只正常交易的票拉到快照检查你要用的字段是不是都有值是不是都是数字类型而不是字符串。有的源把价格返回成字符串10.50你在Python里做算术会直接报错或者悄悄转换成浮点后者更危险因为不会报错。第二项时区正确性。拉一次快照把返回的ts转成可读时间和当前时间对比。差8小时就是UTC没转差几秒是正常的网络延迟。这一步不做后面所有基于时间的计算都会歪。第三项复权口径。找一只近两年有过送转的票拉除权日前后的日线看价格序列有没有跳空。如果除权日当天出现一个明显的不连续缺口比如前一天收盘20当天开盘10说明给的是不复权原始价如果序列平滑说明给的是复权价。两种都可用但你必须知道自己拿到的是哪种否则收益率算出来完全是错的。第四项限频行为。这个只能靠压测。写个循环快速发50次请求观察第几次开始返回429或者空结果。同时记录响应头里的限频信息。知道阈值之后才能在代码里设合理的请求间隔和退避策略。这四项过完一个接口基本就可以放心用了。我的经验是四项里最容易翻车的是第三项复权因为它不会报错只会让你在几个月后发现回测结果完全不对。4. 历史行情接口的坑复权、停牌、数据对齐4.1 前复权后复权的计算逻辑以及什么时候该用哪个复权这件事说到底就是一句话公司发生送转或分红之后股价会出现一个非交易性的跳空你要不要把这个跳空抹平。举个具体例子。某票除权前收盘 20 元实施每10股送10股除权后理论开盘价变成 10 元。如果你拿原始数据画K线会看到一根巨大的阴线跌幅50%但你的持仓市值其实一分没少。做技术指标的人会被这根假阴线彻底带偏。复权的做法是乘一个因子。因子怎么算因子 除权后理论价 / 除权前收盘价上面这个例子就是 10/20 0.5。前复权是以上市以来的最新价为基准往前把所有历史价格乘以累计因子。算完之后最新价和真实盘面一致历史价格被调整。优点是看图直观最新价是真的缺点是每次发生新的除权整条历史序列都会变如果你把前复权数据落库了除权之后旧数据就失效了必须重算。后复权是以上市首日为基准往后累乘。算完之后历史价保持不变最新价被放大。优点是历史数据一旦入库就永远不变适合做长期存储和回测缺点是最新价和盘面对不上看着别扭。我的做法是库里存不复权的原始价 独立的复权因子表用的时候现算。这样两种口径随时能出也不会因为一次除权把整个库刷一遍。def adjust_price(raw_price, factor, modeqfq, base_factor1.0): raw_price: 原始价格序列 factor: 对应日期的复权因子序列累计因子 mode: qfq 前复权 / hfq 后复权 base_factor: 前复权时用最新因子做基准后复权时用首日因子做基准 if mode qfq: return [p * f / base_factor for p, f in zip(raw_price, factor)] elif mode hfq: return [p * f / base_factor for p, f in zip(raw_price, factor)] else: return list(raw_price)这段代码里 qfq 和 hfq 的公式形态一样差别只在base_factor取哪个值——前复权取最新那一天的累计因子后复权取第一天。理解这一点复权就不会再绕晕。4.2 停牌日、零成交、一字板三种异常数据的处理方式历史数据里最烦人的不是缺数据而是看起来有数据但其实是假的。三种情况要单独处理。停牌日。有的源在停牌期间直接不返回记录你的日期序列就出现空洞有的源会返回一条记录开高低收全部等于前收盘成交量是0。两种处理方式完全不同。如果是空洞你在算N日收益率的时候必须按交易日对齐不能按自然日如果是填充记录你必须用成交量0 这个条件把它们识别出来否则会把停牌日算成零波动日污染波动率指标。零成交。极少数情况下比如某些流动性极差的标的盘中可能出现某一分钟完全没成交。分钟线接口这时候可能返回空也可能返回一条成交量为0的记录。处理方式和停牌日类似识别出来该跳过就跳过。一字板。涨停或跌停封死的时候全天最高价等于最低价等于收盘价成交量很小。这不是异常数据是真实行情但它会让很多指标失效——比如用最高最低价计算的真实波幅ATR会算出一个接近0的值。这个不需要清洗但做策略的时候要意识到它的存在。我一般在数据入库之后跑一个简单的体检脚本统计每个交易日的记录数、成交量为0的比例、最高价等于最低价的比例。这三个数字异常就说明数据源那边有问题或者当天的日期对齐出了问题。4.3 多源交叉校验抽10只票10个交易日就够单源数据你没法判断对错因为你没有参照。做法是同时接两个源做抽样交叉验证。不需要全量比对成本太高抽样就够。具体做法随机抽 10 只票每只随机抽 10 个交易日把两个源的收盘价拉出来对比。判断标准是相对偏差abs(a-b)/a。日线数据正常情况下应该在万分之一以内如果超过千分之一就要查原因了。常见原因是复权口径不同、除权日处理不同、或者其中一个源的数据本身就是错的。这套抽样脚本我写成了一个固定的小工具每次接入新源就跑一遍二十分钟能出结论。它帮我发现过好几次问题其中一次是某个源在除权日的收盘价用的是除权前价格导致那一天的收益率算出来是负50%非常隐蔽。成交量也可以用同样的方法比对但要注意先统一单位。价格比对的是绝对值成交量比对的是数量级很多偏差其实是手和股的区别不是数据错误。5. 实时接口的工程化从能跑到长期不出事5.1 心跳、重连与断线补数三件套缺一不可推送流的代码新手版和能上生产的版本差别就在这三件事上。心跳。长连接如果长时间没有数据流动中间的网络设备可能会把它判定为空闲连接然后掐掉。所以必须定期发心跳间隔一般 15 到 30 秒。有的服务是客户端主动发ping服务端回pong有的是服务端定时推一条心跳消息。两种都要处理收到pong或者心跳消息就更新一下最后活跃时间。重连。断线是必然事件不是异常事件。重连策略我固定用指数退避第一次断1秒后重连第二次2秒第三次4秒一直到60秒封顶重连成功后重置计数。为什么要退避而不是立刻重连因为如果是服务端在重启或者网络整体抖动你疯狂重连只会加重负担还容易被限流拉黑。断线补数。这是最容易被漏掉的一环。重连成功不等于数据完整断线的那几分钟数据是空的。做法是在内存里维护一个last_ts记录最后一条成功处理的数据时间重连之后先用快照接口或者历史接口把last_ts到当前时间的区间数据拉回来补齐再恢复正常接收。没有这一步你的时序数据会静悄悄地出现空洞而且很难发现。import time import random def connect_with_backoff(connect_fn, on_message): delay 1 while True: try: conn connect_fn() delay 1 # 连接成功重置退避 last_ts None while True: msg conn.recv(timeout30) if msg is None: # 超时发心跳 conn.send_ping() continue if msg.type pong: continue last_ts on_message(msg, last_ts) except Exception as e: print(f连接异常: {e}, {delay}秒后重连) time.sleep(delay random.uniform(0, 0.5)) # 加抖动避免同时重连 delay min(delay * 2, 60)这段代码里加了random.uniform(0, 0.5)的抖动作用是避免多个客户端在同一时刻一起重连把服务端打崩。如果是单机单连接这个抖动可有可无如果是多进程部署它很重要。5.2 限频与缓存令牌桶加TTL把请求量压下来限频处理的核心思路是主动控制而不是撞到429再退避。我一般用令牌桶桶容量设成服务端每分钟额度每秒钟往桶里补额度/60个令牌每次请求取一个令牌取不到就等。比限频更重要的是缓存。很多请求其实是不必要的。日线数据在收盘之后就不会变了缓存到第二天开盘完全没问题快照数据在几秒内变化不大重复请求同一只票在极短时间内没有意义。我的缓存策略分三层日线历史按(代码, 日期)做键永久缓存。除非发现数据错误否则不重拉。当日快照TTL 设 3 到 5 秒同一个代码在窗口内重复请求直接返回缓存。基本资料股票名称、上市日期这类缓存一天。这三层加起来实际请求量能压到理论值的十分之一以下。我做过一个粗略统计加了缓存之后同样的看板刷新逻辑每分钟请求数从一千多降到不足一百限频的烦恼基本消失。提示缓存一定要设上限并做过期清理。我见过一个服务用字典做缓存没设上限跑了半个月内存吃满被系统杀掉排查了半天才发现是缓存字典无限增长。5.3 落库结构设计主键选错后面全在改表时序数据的表结构核心就一件事主键怎么定。我踩过的坑是用自增ID做主键结果每次补数据都产生重复记录查的时候要写一堆去重逻辑。后来统一改成业务主键用INSERT OR REPLACE或者等价的 upsert重复拉取自动覆盖逻辑一下子简单了。CREATE TABLE kline_daily ( symbol VARCHAR(16) NOT NULL, trade_date DATE NOT NULL, open DECIMAL(12,4), high DECIMAL(12,4), low DECIMAL(12,4), close DECIMAL(12,4), volume BIGINT, amount DECIMAL(20,4), adjust_flag CHAR(1) DEFAULT N, -- N不复权 Q前复权 H后复权 source VARCHAR(32), updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (symbol, trade_date, adjust_flag) );几个设计上的考虑。adjust_flag放进主键是因为同一只票同一天可能有三种口径的数据混在一起必然出问题。source字段记录数据来源做多源交叉验证的时候非常有用出问题能立刻知道是哪家给的。updated_at用于排查这条数据是什么时候写进来的补数据之后能看出来。分钟线的表结构类似主键换成(symbol, trade_time, adjust_flag)但要注意分区。分钟线数据量大我一般按月或者按年做范围分区查询的时候带上时间范围能直接命中分区性能差别很明显。还有一点时间字段统一存UTC展示的时候再转本地时区。这个规则一旦定下来就不要破例否则你会陷入这张表是UTC那张表是本地时间的混乱每次写查询都要翻文档确认。6. 常见问题与排查实录6.1 一张速查表覆盖八成的报错场景下面这张表是我这几年攒下来的遇到问题先扫一遍大部分情况能直接定位。现象大概率原因排查动作返回空数组代码格式不符或缺交易所后缀打印完整请求URL逐个替换代码写法试返回HTML而不是JSON域名或路径错误被重定向看响应头 Content-Type401 / 403缺密钥或密钥失效检查请求头里的鉴权字段429触发限频看响应头限频信息加令牌桶和退避时间戳差整8小时UTC与本地时区未转换统一转UTC存储成交量差100倍手与股的单位混淆用成交额除以成交量反推均价验证复权后价格对不上复权基准日不同统一基准日重新计算历史数据缺最近一天源端收盘后延迟入库次日再拉或换源补当天中文显示乱码响应头未声明字符集显式指定 encodingutf-8连接几分钟后自动断开缺少心跳加定时ping间隔30秒以内这张表建议直接放进项目的故障排查文档新人接手时能省下大量沟通成本。6.2 我实际踩过的几个坑都是不报错的那种比报错更麻烦的是不报错但结果是错的。下面这几个坑我都真实遇到过共同点是程序照常运行日志干干净净只有结果不对。第一个坑快照的最新价其实有延迟。有的免费源给的快照数据本身就有几分钟延迟但字段名和实时源一模一样你看不出来。表现是策略在盘中做出的判断总是慢半拍。发现方法很简单在明显有成交的时段把接口返回的价格和你看到的行情界面对比如果持续差一截就是延迟档。解决办法只有一个——换源靠代码弥补不了数据本身的延迟。第二个坑分页参数不生效。有个历史接口文档写着支持page和page_size实际调用的时候不管传多少永远返回前100条。我是拉了五年数据发现总条数对不上才察觉的。从此养成习惯拉完历史数据之后算一下记录数跟预期的交易日数量对比差太多就说明分页有问题。校验逻辑很简单预期条数 ≈ 交易日数量偏差超过5%就要查。第三个坑除权日的复权因子断层。有一次因子表漏了一个除权事件导致那一段的前复权序列整体偏移回测出来的收益率比实际高了十几个点。这个错误藏得很深因为它不影响程序运行。后来我加了一个自动校验每次算完前复权序列检查相邻两日的涨跌幅如果出现超过涨跌停限制的跳变就报警。这个检查能拦住绝大多数复权错误。第四个坑多个数据源的日期口径不一致。有的源把成交量为0的停牌日记进日线有的不记。做多源合并的时候如果不统一这个口径会出现某一天只在一个源里存在的情况合并逻辑就会出问题。解决办法是合并前先按是否有成交量过滤一遍停牌记录统一成只保留有效交易日。第五个坑并发请求把本地网络打满。拉全市场历史数据的时候我一开始开了50个并发结果本地连接数耗尽大量请求超时重试逻辑又把负载翻倍最后雪崩。后来把并发降到8加了请求间的随机间隔反而总耗时更短。这个经验是并发不是越高越快找到那个拐点比堆数值重要一般家用或办公网络并发8到16是比较舒服的区间。6.3 批量拉全市场历史数据的节奏控制单独说这一块因为它是很多人第一次真正被数据量教育的地方。全市场几千只票每只拉五年日线就是几万次请求。如果每次请求耗时200毫秒串行就是两个多小时。这里有两个优化方向。一是减少请求次数。优先找支持一次请求返回多只代码或者按日期拉全市场的接口形态。按日期拉是最划算的一天一次请求拿全市场五年也就一千多次请求几十分钟就完了。如果接口只支持单代码那就没法省只能优化节奏。二是控制节奏避免雪崩。我用的参数是并发8到12请求之间加50到150毫秒的随机间隔失败重试最多3次且用指数退避。这套参数在普通宽带上实测很稳五年的全市场日线大概一个多小时能拉完中途出错率很低。拉完之后一定要做完整性校验这一步不能省。校验项总记录数是否接近预期、每只票的最早和最晚日期是否覆盖目标区间、有没有整段缺失的日期、有没有重复记录。我一般写一个校验脚本输出一份报告看一眼就知道这次拉取质量如何。from collections import defaultdict def check_completeness(rows, expected_days): by_symbol defaultdict(list) for r in rows: by_symbol[r[symbol]].append(r[trade_date]) report [] for sym, dates in by_symbol.items(): dates.sort() got len(set(dates)) dup len(dates) - got coverage got / expected_days if expected_days else 0 if coverage 0.95 or dup 0: report.append({ symbol: sym, records: got, duplicates: dup, coverage: round(coverage, 4) }) return report这段脚本输出的报告能一次性找出覆盖不足和重复入库的票比人肉抽查靠谱得多。最后分享一个我自己的习惯每接一个新数据源先写一个十分钟就能跑完的冒烟测试把快照、历史、复权、限频四项各测一次输出一份简短的通过报告。这个脚本我已经迭代了十几版每次遇到新问题就往里加一条检查。它不解决任何业务问题但每次换源或者升级接口版本的时候能让我少熬好几个晚上。数据这东西出问题时从来不会主动告诉你只能靠这些前置检查把它揪出来。