系列前情#06 我们搭好了「数据 / 配置 / 日志 / 策略」的最小可运行项目骨架。本篇解决「数据有缺口怎么办」——演示证临时切换、断网错过、新上市股票都要靠断点续传 幂等写入补回来。一、数据缺口的常见场景跑量化系统最怕的不是「策略不赚钱」而是「数据悄悄少了」切换证书从免费证书换成付费证书时历史数据补一份断网 / API 限流某天 2:00–4:00 拉取失败第二天补这一段新上市股票刚买的标的只有近期数据要回填到某个起始日期断点续传跑了 5 小时的批量拉取中途崩溃重启时不能再从 0 开始这些场景都指向同一个工程能力断点检测 区间补采 幂等入库。二、本文你将得到什么断点检测每只股票一行 SQL查出已有数据的最大日期区间补采从断点日期 1 拉到今天按区间调用 API不重不漏幂等入库PRIMARY KEY (date, stock)INSERT OR IGNORE重复数据自动去重回填校验每只股票的count / first / last报告 节假日缺口扫描三、整体架构启动回填任务 ↓ [1] detect_gap(conn, stock, today) ├─ last MAX(date) WHERE stock? ├─ 无数据 → start 默认起点, end today └─ 有数据 → start last 1, end today ↓ [2] fetch_history(stock, start, end) └─ 调 APIstYYYYMMDD, etYYYYMMDD, periodd, dividendf ↓ [3] upsert_rows(conn, rows) └─ INSERT OR IGNORE ... (date, stock, o, h, l, c, pc, v) ↓ [4] 校验count / first / last / 缺口检测四块职责清晰检测 → 拉取 → 写入 → 校验。可以独立替换每一块。四、断点检测MAX(date) 一行 SQL缺口从哪里开始拉答案就在已入库数据的最后一天-- 找出指定股票已入库的最大日期SELECTMAX(date)FROMklineWHEREstock600519;代码实现fromtypingimportOptionaldefget_max_date(conn,stock:str)-Optional[str]:读取指定股票的最大已入库日期。curconn.execute(SELECT MAX(date) FROM kline WHERE stock?,(stock,))rowcur.fetchone()returnrow[0]ifrowandrow[0]elseNonedefdetect_gap(conn,stock:str,today:str):返回需要补采的 (start_date, end_date)。 无数据 → 全量有数据 → 增量。lastget_max_date(conn,stock)ifnotlast:# 该股票完全没有数据 → 从默认起点拉演示证 60 交易日生产证书 5 年return(2025-09-01,today)# 有数据 → 从最后日期下一天拉到今天fromdatetimeimportdatetime,timedelta start(datetime.strptime(last,%Y-%m-%d)timedelta(days1)).strftime(%Y-%m-%d)ifstarttoday:return(today,today)# 已是最新return(start,today)注意检测必须在 fetch 之前执行否则会误判为已有数据 → 增量导致首次全量数据被错误识别。五、幂等写入复合主键 INSERT OR IGNORESQLite 建表时把(date, stock)设为复合主键配合INSERT OR IGNORE实现「存在则跳过」CREATETABLEIFNOTEXISTSkline(dateTEXTNOTNULL,stockTEXTNOTNULL,openREAL,highREAL,lowREAL,closeREAL,pre_closeREAL,volumeINTEGER,PRIMARYKEY(date,stock));Python 写入importsqlite3defupsert_with_statcheck(conn,rows):rows [(date, stock, o, h, l, c, pc, v), ...] 返回 (inserted, skipped)。beforeconn.execute(SELECT COUNT(*) FROM kline).fetchone()[0]conn.executemany( INSERT OR IGNORE INTO kline (date, stock, open, high, low, close, pre_close, volume) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ,rows)conn.commit()afterconn.execute(SELECT COUNT(*) FROM kline).fetchone()[0]insertedafter-before skippedlen(rows)-insertedreturninserted,skipped两重保险PRIMARY KEY (date, stock)数据库层保证唯一INSERT OR IGNORESQL 层安全跳过任意并发场景脚本重跑、调度器重发、分布式多实例都不会污染数据。六、调用真实 API区间补采frommairuiimportClientimporttime DEMO_LICENCELICENCE-66D8-9F96-0C7F0FBCD073deffetch_history(stock:str,start:str,end:str,licence:str)-list:拉区间日线断点续传 fallback。forattemptinrange(2):try:cliClient(licence)returncli.stock_history(codestock,ststart,etend,periodd,dividendf)or[]exceptExceptionase:print(f[retry{attempt1}]{stock}{start}~{end}:{str(e)[:80]})time.sleep(0.5)return[]关键参数code股票代码注意 SDK 是code不是stockst/et起止日期YYYYMMDD或YYYY-MM-DDperiodd日线m1分钟线需单独文档dividendf前复权不复权传n后复权传b七、回填校验count / first / last 缺口扫描拉完数据一定要校验不能「拉完就当搞定」。三个最小校验点defvalidate(conn,stock:str,today:str):countconn.execute(SELECT COUNT(*) FROM kline WHERE stock?,(stock,)).fetchone()[0]firstconn.execute(SELECT MIN(date) FROM kline WHERE stock?,(stock,)).fetchone()[0]lastconn.execute(SELECT MAX(date) FROM kline WHERE stock?,(stock,)).fetchone()[0]return{stock:stock,count:count,first:first,last:last}更进一步扫描「预期交易日 vs 实际入库」的差异发现潜在缺口defdetect_missing(conn,stock:str,today:str)-list:粗略按工作日扫缺口建议生产替换为交易日历。haveset(r[0]forrinconn.execute(SELECT date FROM kline WHERE stock?,(stock,)).fetchall())ifnothave:return[]last_datemax(have)expected[]fromdatetimeimportdatetime,timedelta ddatetime.strptime(last_date,%Y-%m-%d)enddatetime.strptime(today,%Y-%m-%d)whiledend:ifd.weekday()5:# 周一到周五dsd.strftime(%Y-%m-%d)ifdsnotinhave:expected.append(ds)dtimedelta(days1)returnexpected八、完整流程与实测把上面 4 块串起来defbackfill_one(conn,stock:str,today:str,licence:str):was_emptyget_max_date(conn,stock)isNonestart,enddetect_gap(conn,stock,today)ifstartend:return{stock:stock,mode:no_gap,fetched:0,inserted:0,skipped:0}print(f[fetch]{stock}{start}~{end}...)rawfetch_history(stock,start,end,licence)rows[]forrinraw:dater.get(t,)[:10]ifnotdateordatestartordateend:continuerows.append((date,stock,r.get(o),r.get(h),r.get(l),r.get(c),r.get(pc),r.get(v)))ifrows:inserted,skippedupsert_with_statcheck(conn,rows)else:inserted,skipped0,0modefirst_timeifwas_emptyelseincrementalreturn{stock:stock,mode:mode,gap:f{start}~{end},fetched:len(rows),inserted:inserted,skipped:skipped}实测运行演示证3 只股票第 1 轮首次拉取 600519: modefirst_time, gap2025-09-01~2025-12-01, fetched50, inserted50, skipped0 000001: modefirst_time, gap2025-09-01~2025-12-01, fetched50, inserted50, skipped0 300750: modefirst_time, gap2025-09-01~2025-12-01, fetched50, inserted50, skipped0 第 2 轮再次跑、完全幂等 600519: modeincremental, gap2025-12-01~2025-12-01, fetched1, inserted0, skipped1 000001: modeincremental, gap2025-12-01~2025-12-01, fetched1, inserted0, skipped1 300750: modeincremental, gap2025-12-01~2025-12-01, fetched1, inserted0, skipped1 回填校验 600519: count50 first2025-09-15 last2025-12-01 000001: count50 first2025-09-15 last2025-12-01 300750: count50 first2025-09-15 last2025-12-01 节假日缺口检测3 只均 0 缺口总结第 1 轮每只 50 条入库共 150 行第 2 轮每只 fetch1区间只有 12-01 一天→ 全部 PR 命中 → skipped1最终每只 50 条 / first2025-09-15 / last2025-12-01三只一致九、常见坑「今天」硬编码死的today 2025-12-01只适合演示。生产环境必须today datetime.now().strftime(%Y-%m-%d)或从外部配置传入否则补采永远跑不到真实今天。节假日判断过粗用d.weekday() 5判断「周一到周五」会把节假日误判为缺口。建议接入交易接口的交易日历或本地缓存的trading_days.csv。没记录补采日志fetched / inserted / skipped必须落库task_log表3 个月后出问题能一键回放。详见 #02 APScheduler 八节。多实例并发补采分布式部署同一份代码会重复拉数据。简单方案是 RedisSETNX加锁复杂方案上 APSchedulerSQLAlchemyJobStore统一调度。演示证的范围错觉演示证只返回最近 60 交易日截至 2025-12-01生产证书可拉到 5 年历史。补采脚本逻辑通用但起始日期按证书档位调整。十、小结历史数据补采的工程核心是三件事断点MAX(date)查出缺口起点start last 1区间st/et控制 API 拉取范围不重不漏幂等复合主键 INSERT OR IGNORE并行/重跑都不污染数据这套方案对回测、监控、生产实盘三个场景全都适用是策略计算的「上游数据保障」。下一篇#08我们收尾——「量化系统工程上线 checklist」从脚本到生产的最小闭环把环境隔离 / 日志轮转 / 备份 / 灰度 / 回滚的兜底动作清点一遍让你的脚本真正上生产不翻车。免责声明本文仅供技术学习交流不构成任何投资建议。量化策略回测表现不代表未来收益投资有风险决策需谨慎。代码与文档https://github.com/MaiRuiApi