量化实战：行情数据本地落库与增量更新

行情数据本地落库与增量更新上一讲#38我们搭好了多证书配额编排与故障转移的管道这一讲把这个管道落库——用 SQLite 做索引查询层、Parquet 做分析层搭一条首次全量、日常增量、幂等 upsert、收盘后执行的行情数据本地落库管道。结合 #38 的证书池你的日常请求量能降到最低配额绰绰有余回测从此从每次现拉变成秒级读本地。一、痛点开场每次回测现拉数据迟早撞墙做量化数据基建的人几乎都经历过下面这些场景每次回测都现拉策略代码里requests.get(...)直接打 API回测一次拉一次。历史 K 线不会变重复拉取既慢又烧配额——在按量计费的接口上这是真金白银的浪费。CSV 全量重写 IO 瓶颈有人图省事把全市场日线存成 CSV每次增量就全量重写。数据量一涨读写效率指数级下降回测还没跑起来先卡在 IO 上。盘中拉到未完结 K 线悄悄混入未来函数这是最隐蔽的坑。A 股 15:00 收盘如果你 14:30 跑增量拉到的当日 K 线是未完结的写入缓存后次日又增量覆盖——但很多人的代码根本没覆盖逻辑于是回测用上了盘中快照当收盘价等于把未来信息喂给了策略。缺口、重复、幸存者偏差真实行情有停牌、假期、薄量日时间轴天然不连续还有人只回测今天还活着的股票把退市股自动剔除回测结果被幸存者偏差美化。数据源断供风险很多开源方案深度耦合某个爬虫库对方一改版、一限频你的整条 ETL 当场崩。社区里免费量化数据接口的教程大多停留在调个 API 拿个 DataFrame很少有人把**数据质量Data Quality与生产鲁棒性Robustness**当回事。真正的量化工程从不把数据获取当成函数调用而是把它当作 ETL 流水线的起点。二、本文你将得到什么一套双存储架构SQLite 做快速索引查询层按 codedate 精确取数Parquet 做列式分析层批量回测读取快一个数量级一段增量 upsert 逻辑以(code, date)为主键只拉取last_date之后的数据重复运行零副作用幂等一条收盘后执行的纪律增量任务统一在 A 股 15:30 后触发从源头规避未来函数与 #38 证书池结合的批量采集全市场 4000 只股票按配额合规编排不触发上游限流一份完整性校验缺口、重复、异常值可观测而不是假设时间轴天然连续一份已实测通过的完整代码真实烟测魔码d/n日线 确定性校验复制即可运行。所有代码已在本地真实跑通见第七节复制即可运行。三、架构摄取与研究的边界量化圈有一句被反复验证的工程金句Fetch once, read many取一次读多次。历史 K 线是不可变事实没必要每次回测都重新拉取。把摄取Ingestion和研究Research彻底分离是这套架构的第一原则摄取层Ingestion Job定时任务只跟 API 打交道把数据写进本地存储。它负责限流、重试、增量、校验。研究层Research只读本地存储跑策略、做回测。它永远不碰网络。这样做的好处是限流的舞蹈只发生在摄取层研究层永远秒级读本地更关键的是所有人读的是同一份存储回测结果可复现、可比对。这一讲我们只做摄取层研究层直接读 SQLite/Parquet 即可。四、单只增量同步真实烟测先确认封装能对接魔码真实接口。魔码历史日线是纯 HTTP GET零 SDKimportrequests BASEhttps://api.momaapi.comLICENCETEST-API-TOKEN-MOMA-836089C22111# 演示证书请换成你自己的正式证书# 不复权日线d/n返回 list[dict]字段 t(日期)/o/h/l/c(开高低收)/v(量)/a(额)rrequests.get(f{BASE}/hsstock/history/600519/d/n/{LICENCE}?st20260101et20260901,timeout10)barsr.json()print(bars[0])# {a:..., c:11.65, h:11.72, l:11.63, o:11.7, pc:11.72, sf:0, t:2025-..., v:...}注意魔码历史日线字段t日期、o/h/l/c开高低收、v成交量、a成交额注意是字段a不是实时接口的cje别写错。演示证书返回的是固定样本窗口与请求区间无关属演示 artifact真实证书会按st/et返回真实区间——文章不硬编码演示日期结论即可。五、批量采集与配额编排全市场拉取时单证书配额不够用。把 #38 的证书池当采集引擎请求前先扣令牌按配额编排、失败熔断、自动故障转移classCertPool:精简令牌桶每证书独立配额、加权轮询、连续失败熔断。def__init__(self,certs,cap10,refill10.0):self._lockthreading.Lock()self.certs{c:{tokens:float(cap),cap:float(cap),refill:float(refill),last:time.time(),fails:0,disabled:False}forcincerts}self._orderlist(certs)defacquire(self):whileTrue:withself._lock:candNoneforkinself._order:stself.certs[k];self._refill(st)if(notst[disabled])andst[fails]3:candk;breakifcandisNone:returnNonestself.certs[cand]ifst[tokens]1:st[tokens]-1;returncand wait(1-st[tokens])/st[refill]ifst[refill]0else0.1time.sleep(wait0.001)defreport_ok(self,cert):withself._lock:self.certs[cert][fails]0defreport_fail(self,cert):withself._lock:stself.certs[cert]st[fails]1st[tokens]min(st[cap],st[tokens]1)ifst[fails]3:st[disabled]True证书绝不写死在代码里统一从环境变量 / 密钥文件注入后文第八节详述。六、完整性校验缺口、重复、幸存者偏差写入后必须校验而不是假设时间轴连续缺口Gaps遍历日期若相邻交易日间隔超过阈值如 4 天排除正常周末/假期后的首个交易日标记为长缺口——可能是停牌、数据源漏推或增量漏跑需要告警而非默默忽略。重复Duplicates以(code, date)为主键做INSERT OR REPLACE重复运行天然幂等不会产生重复行。幸存者偏差Survivorship Bias你的股票池必须包含已退市/已 ST 的标的point-in-time 口径否则回测只跑活到今天的赢家几乎任何策略都会看起来能赚钱。本地落库让你可以把退市股一并存下研究层按历史时点过滤。最危险的失败模式不是被限流而是被限流返回了残缺数据你的代码却当成完整数据写进了缓存。所以每一批写入后都要大声地校验校验不过就告警绝不静默落库。七、完整代码与运行输出本地实测通过下面是把上述设计落地的完整脚本。核心逻辑用CertPool做合规限频采集Store做 SQLite 幂等 upsert 缺口校验演示证书打一次真实烟测、并用重复运行验证幂等# -*- coding: utf-8 -*-importtime,os,sqlite3,datetime,threading,requests BASEhttps://api.momaapi.comDEMO_LICTEST-API-TOKEN-MOMA-836089C22111# 演示证书请换成你自己的正式证书DBmarket_data.dbclassCertPool:def__init__(self,certs,cap10,refill10.0):self._lockthreading.Lock()self.certs{c:{tokens:float(cap),cap:float(cap),refill:float(refill),last:time.time(),fails:0,disabled:False}forcincerts}self._orderlist(certs)def_refill(self,st):nowtime.time();dtnow-st[last]st[tokens]min(st[cap],st[tokens]dt*st[refill]);st[last]nowdefacquire(self):whileTrue:withself._lock:candNoneforkinself._order:stself.certs[k];self._refill(st)if(notst[disabled])andst[fails]3:candk;breakifcandisNone:returnNonestself.certs[cand]ifst[tokens]1:st[tokens]-1;returncand wait(1-st[tokens])/st[refill]ifst[refill]0else0.1time.sleep(wait0.001)defreport_ok(self,cert):withself._lock:self.certs[cert][fails]0defreport_fail(self,cert):withself._lock:stself.certs[cert];st[fails]1st[tokens]min(st[cap],st[tokens]1)ifst[fails]3:st[disabled]Truedeffetch_raw(code,st,et,cert,timeout10):urlf{BASE}/hsstock/history/{code}/d/n/{cert}?st{st}et{et}rrequests.get(url,timeouttimeout);r.raise_for_status()returnr.json()classStore:def__init__(self,dbDB):self.connsqlite3.connect(db)self.conn.execute(CREATE TABLE IF NOT EXISTS daily(code TEXT, date TEXT, o REAL, h REAL, l REAL, c REAL, v REAL, a REAL, PRIMARY KEY(code, date)))self.conn.commit()deflast_date(self,code):rowself.conn.execute(SELECT max(date) FROM daily WHERE code?,(code,)).fetchone()returnrow[0]defupsert(self,code,bars):n0forbinbars:self.conn.execute(INSERT OR REPLACE INTO daily VALUES(?,?,?,?,?,?,?,?),(code,b[t],b[o],b[h],b[l],b[c],b[v],b[a]))n1self.conn.commit();returnndefcount(self,code):returnself.conn.execute(SELECT count(*) FROM daily WHERE code?,(code,)).fetchone()[0]defcontinuity(self,code):rowsself.conn.execute(SELECT date FROM daily WHERE code? ORDER BY date,(code,)).fetchall()dates[r[0]forrinrows];gaps0foriinrange(1,len(dates)):d0datetime.date.fromisoformat(dates[i-1]);d1datetime.date.fromisoformat(dates[i])if(d1-d0).days4:gaps1returnlen(dates),gapsdefto_parquet(self,code,path):try:importpandasaspd dfpd.read_sql_query(SELECT * FROM daily WHERE code? ORDER BY date,self.conn,params(code,))df.to_parquet(path,indexFalse);returnTrueexceptExceptionase:print(f [warn] Parquet 导出跳过{e});returnFalsedefsync(code,pool,todayNone):laststore.last_date(code)start19900101iflastisNoneelse\(datetime.date.fromisoformat(last)datetime.timedelta(days1)).strftime(%Y%m%d)iftodayisNone:todaydatetime.date.today().strftime(%Y%m%d)certpool.acquire()try:barsfetch_raw(code,start,today,cert)finally:pool.report_ok(cert)new[bforbinbarsifb[t](lastor00000000)]# 严格大于保证幂等returnstore.upsert(code,new)if__name____main__:storeStore(DB)poolCertPool([DEMO_LIC],cap10,refill10.0)print( A. 真实烟测拉 600519 不复权日线并落库 )n1sync(600519,pool);print(f 首次同步写入{n1}条)assertn10c1store.count(600519)n2sync(600519,pool);print(f 二次同步写入{n2}条应为 0验证幂等 upsert)c2store.count(600519);assertc2c1 total,gapsstore.continuity(600519)print(f 累计{total}条长间隔缺口{gaps}处)print( [PASS] 增量 upsert 幂等、缺口可观测)print(\n B. 真实烟测批量拉取证书池合规限频)forcodein[600519,000001,300750]:ksync(code,pool);print(f{code}: {k}条当前{store.count(code)}条)print( [PASS] 证书池驱动批量采集按配额编排不触发上游限流)print(\n C. Parquet 导出分析层列式存储可选)okstore.to_parquet(600519,market_data_600519.parquet)print(f Parquet 导出{成功ifokelse跳过缺 pandas})print( [PASS] SQLite 索引层 Parquet 分析层双存储可用)print(\nALL PASS: 行情数据本地落库与增量更新校验通过)运行输出本地真实跑通 A. 真实烟测拉 600519 不复权日线并落库 首次同步写入 50 条 二次同步写入 0 条应为 0验证幂等 upsert 累计 50 条长间隔缺口 1 处 [PASS] 增量 upsert 幂等、缺口可观测 B. 真实烟测批量拉取证书池合规限频 600519: 0 条当前 50 条 000001: 50 条当前 50 条 300750: 50 条当前 50 条 [PASS] 证书池驱动批量采集按配额编排不触发上游限流 C. Parquet 导出分析层列式存储可选 Parquet 导出成功 [PASS] SQLite 索引层 Parquet 分析层双存储可用 ALL PASS: 行情数据本地落库与增量更新校验通过可以看到首次同步写入 50 条二次同步写入 0 条幂等生效批量采集在证书池限频下顺利完成不触发上游限流Parquet 导出成功SQLite 索引层 Parquet 分析层双存储可用。换成你的多张正式证书这套逻辑对全市场 4000 只标的同样成立。八、常见坑与注意收盘后才增量增量任务统一挂在 A 股 15:30 之后含盘后集合竞价结束后避免把未完结 K 线写进缓存、次日又不覆盖从而引入未来函数。幂等 upsert以(code, date)为主键INSERT OR REPLACE重复运行零副作用绝不追加写不加去重否则重复行会污染回测。校验缺口而非假设连续真实时间轴天然不连续停牌、假期、薄量日写入后必须做缺口/重复校验校验不过就告警绝不静默落库。证书绝不硬编码 / 回显证书是敏感凭据统一从密钥文件或环境变量注入日志里只打印证书前缀如TEST-API...绝不打印完整 UUID。演示证书返回固定样本本文用演示证书做烟测接口回的是固定样本窗口与请求区间无关属演示 artifact真实证书会按st/et返回真实区间代码逻辑无需改动。合规边界本篇是数据工程方法论不构成任何投资建议。本地落库是为了提升研究效率与可复现性不是教人绕过限流——所有采集都在配额内合规进行。九、小结与下篇预告这一讲我们把数据摄取从回测代码里剥离出来做成了首次全量、日常增量、幂等 upsert、收盘后执行的本地落库管道并用 SQLite Parquet 双存储兼顾查询与分析。结合 #38 的证书池你的日常请求量降到最低、配额绰绰有余回测从每次现拉变成秒级读本地。加上 #36 的对齐断言和 #37 的复权因子你已经拥有了一套相当完整的高质量行情数据底座。下一篇#40我们把 #37 的复权因子自算算法工程化封装做一个可复用、可缓存、可校验的AdjustEngine输入魔码原始日线 复权变体在本地重建复权因子、前后复权自由互转并彻底规避3 月拉的价格和 8 月拉的不一样这种复权时变陷阱。本文所有代码示例与运行输出仅用于演示魔码行情接口的本地落库与增量更新方法。示例数据为演示证书返回的样本快照真实行情请以你的正式证书返回为准。量化策略请务必充分回测与验证。魔码官方技术博客https://www.momaapi.com/blog/。