上次我们把数据存下来了，SQLite / Parquet / CSV 各种爽。

。第一周一切顺利，5000 只股票每天 18 点拉一次、增量落 SQLite、回测一点不卡。

：没有"上次拉到哪里了"的状态表，每次脚本都假设自己是从零开始。

让脚本只拉"新数据"，不是每次都全量重跑

。完整跑一遍「首次全量 → 增量 → 幂等再跑」的真实流程，看实测能从"每次 30 分钟"压到"每次 60 秒"。


CREATE TABLE sync_state (
    code        TEXT PRIMARY KEY,    -- 股票代码
    last_et     TEXT,                -- 该股上次拉到的最后交易日，YYYY-MM-DD
    rows_total  INTEGER,             -- 累计条数（仅用于健康检查）
    updated_at  TEXT                 -- 上次更新 state 表的时间，YYYY-MM-DD
);
CREATE INDEX idx_sync_state_updated_at ON sync_state(updated_at);


CREATE TABLE klines (
    code TEXT, t TEXT,               -- 代码 + 日期
    o REAL, h REAL, l REAL, c REAL,  -- OHLC
    v INTEGER, a REAL, pc REAL,      -- 成交量/额/昨收
    PRIMARY KEY (code, t)            -- 天然幂等
);
CREATE INDEX idx_klines_t ON klines(t);

 是关键——保证 INSERT OR REPLACE 不会插重复行；state 表 + 幂等 = 断网重连也无害。


import os
import sqlite3
import mairui
​
api = mairui.Client("LICENCE-66D8-9F96-0C7F0FBCD073")  # 演示证书，演示证书（读者复制即用，无需替换）
conn = sqlite3.connect("klines_incremental.db")
​
def seed_state_first_time():
    """首次全量 + 写 state 表"""
    state_rows = []
    for code in SAMPLE_CODES:
        rows = api.stock_history(code=code, period="d", st="",
                                 et="20260814",         # 上周五
                                 dividend="n", lt=200) or []
        if not rows:
            continue
        conn.executemany(
            "INSERT OR REPLACE INTO klines (code,t,o,h,l,c,v,a,pc) "
            "VALUES (?,?,?,?,?,?,?,?,?)",
            [(code, r["t"], r["o"], r["h"], r["l"], r["c"],
              int(r["v"] or 0), r["a"], r["pc"]) for r in rows],
        )
        last_t = max(r["t"] for r in rows)
        state_rows.append((code, last_t, len(rows), "20260814"))
    conn.executemany(
        "INSERT OR REPLACE INTO sync_state VALUES (?,?,?,?)", state_rows
    )
    conn.commit()
​
def incremental_fetch():
    """基于 sync_state 表只拉新数据"""
    for code, last_t in conn.execute(
        "SELECT code, last_et FROM sync_state ORDER BY code"
    ).fetchall():
        if not last_t:
            continue
        # last_t 是 YYYY-MM-DD，加 1 天作为起点
        y, m, d = last_t.split("-")
        from datetime import datetime, timedelta
        st = (datetime(int(y), int(m), int(d)) + timedelta(days=1)).strftime("%Y%m%d")
        # 增量最多 10 条（A 股一天就 1 条 K 线，10 足够节假日补漏）
        rows = api.stock_history(code=code, period="d", st=st,
                                 et="20260817", dividend="n", lt=10) or []
        if rows:
            conn.executemany(
                "INSERT OR REPLACE INTO klines VALUES (?,?,?,?,?,?,?,?,?)",
                [(code, r["t"], r["o"], r["h"], r["l"], r["c"],
                  int(r["v"] or 0), r["a"], r["pc"]) for r in rows],
            )
            new_last = max(r["t"] for r in rows)
            conn.execute(
                "UPDATE sync_state SET last_et=?, rows_total=rows_total+?, "
                "updated_at='20260817' WHERE code=?",
                (new_last, len(rows), code),
            )
    conn.commit()

样本：5 只股（平安银行 / 贵州茅台 / 宁德时代 / 中国平安 / 五粮液）。INITIAL_ET = 

阶段 1：首次全量（脚本"上周五"第一次跑）


✓ 新增 1000 条 K 线，耗时 0.41s（2434 条/s）
✓ SQLite 累计 1000 条
✓ sync_state 表（都停在 2026-08-14）:
    000001.SZ  last_et=2026-08-14  rows_total=200
    600519.SH  last_et=2026-08-14  rows_total=200
    300750.SZ  last_et=2026-08-14  rows_total=200
    601318.SH  last_et=2026-08-14  rows_total=200
    000858.SZ  last_et=2026-08-14  rows_total=200

阶段 2：模拟「过几天后重跑」（今天 2026-08-17 重跑）


✓ 本次新增 5 条 / 调用 5 次 SDK / 耗时 498.9ms
✓ SQLite 累计 1005 条
✓ 当前 sync_state（已全部更新到 2026-08-17）:
    000001.SZ  last_et=2026-08-17  rows_total=201  updated=20260817
    600519.SH  last_et=2026-08-17  rows_total=201  updated=20260817
    ...

（周一 2026-08-17 的 K 线），调用 5 次 SDK。

阶段 3：立即再跑（验证幂等）


✓ 本次新增 0 条 / 调用 5 次 / 耗时 383.1ms
→ 日常只需这种「近乎空跑」开销

增量更新的灵魂是 state 表——记住「上次拉到哪里」，下次只拉「之后」；配合 

：1 分钟 K 线一天 240 条 × 5000 只 = 120 万行/天，分区表 + 索引策略 + 按 (code, date) 切分文件。我们会演示把"全市场 1 分钟 K 线"装进单一 SQLite 文件、按月切分、按需查询。

。历史数据回测不代表未来表现，所有量化决策请结合自身风险承受能力。

然后你加了个定时器——crontab -e 加一行 0 18 * * 1-5 python run.py。第一周一切顺利，5000 只股票每天 18 点拉一次、增量落 SQLite、回测一点不卡。

【Python量化系统工程化 #02】全市场K线一拉就卡死？增量更新让脚本每次只拉新数据

编程语言

数据库

后端

将数据抓取从全量重跑优化为增量更新：通过sync_state状态表记录每只股票的最后交易日，首次全量拉取后，后续仅抓取last_et之后的新数据，配合PRIMARY KEY (code, t)实现幂等写入。实测可将API调用次数从5000次降至5次，耗时从30分钟压缩至60秒，并解决跨节假日、断网重连等边界场景下的数据重复与丢失问题。

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

将数据抓取从全量重跑优化为增量更新：通过sync_state状态表记录每只股票的最后交易日，首次全量拉取后，后续仅抓取last_et之后的新数据，配合PRIMARY KEY (code, t)实现幂等写入。实测可将API调用次数从5000次降至5次，耗时从30分钟压缩至60秒，并解决跨节假日、断网重连等边界场景下的数据...

【Python量化系统工程化 #02】全市场K线一拉就卡死？增量更新让脚本每次只拉新数据

【Python量化系统工程化 #02】全市场K线一拉就卡死？增量更新让脚本每次只拉新数据

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐