

> 系列前情：#06 我们搭好了「数据 / 配置 / 日志 / 策略」的最小可运行项目骨架。本篇解决「数据有缺口怎么办」——演示证临时切换、断网错过、新上市股票，都要靠断点续传 + 幂等写入补回来。

## 一、数据缺口的常见场景

跑量化系统最怕的不是「策略不赚钱」，而是「数据悄悄少了」：

- **切换证书**：从免费证书换成付费证书时，历史数据补一份
- **断网 / API 限流**：某天 2:00–4:00 拉取失败，第二天补这一段
- **新上市股票**：刚买的标的只有近期数据，要回填到某个起始日期
- **断点续传**：跑了 5 小时的批量拉取中途崩溃，重启时不能再从 0 开始

这些场景都指向同一个工程能力：**断点检测 + 区间补采 + 幂等入库**。

## 二、本文你将得到什么

1. **断点检测**：每只股票一行 SQL，查出已有数据的最大日期
2. **区间补采**：从断点日期 + 1 拉到今天，按区间调用 API（不重不漏）
3. **幂等入库**：`PRIMARY KEY (date, stock)` + `INSERT OR IGNORE`，重复数据自动去重
4. **回填校验**：每只股票的 `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)
   └─ 调 API：st=YYYYMMDD, et=YYYYMMDD, period=d, dividend=f
   ↓
[3] upsert_rows(conn, rows)
   └─ INSERT OR IGNORE ... (date, stock, o, h, l, c, pc, v)
   ↓
[4] 校验：count / first / last / 缺口检测
```

四块职责清晰：检测 → 拉取 → 写入 → 校验。可以独立替换每一块。

## 四、断点检测：MAX(date) 一行 SQL

缺口从哪里开始拉？答案就在已入库数据的最后一天：

```sql
-- 找出指定股票已入库的最大日期
SELECT MAX(date) FROM kline WHERE stock = '600519';
```

代码实现：

```python
from typing import Optional

def get_max_date(conn, stock: str) -> Optional[str]:
    """读取指定股票的最大已入库日期。"""
    cur = conn.execute("SELECT MAX(date) FROM kline WHERE stock=?", (stock,))
    row = cur.fetchone()
    return row[0] if row and row[0] else None


def detect_gap(conn, stock: str, today: str):
    """返回需要补采的 (start_date, end_date)。
    无数据 → 全量；有数据 → 增量。"""
    last = get_max_date(conn, stock)
    if not last:
        # 该股票完全没有数据 → 从默认起点拉（演示证 60 交易日，生产证书 5 年）
        return ("2025-09-01", today)
    # 有数据 → 从最后日期下一天拉到今天
    from datetime import datetime, timedelta
    start = (datetime.strptime(last, "%Y-%m-%d") +
             timedelta(days=1)).strftime("%Y-%m-%d")
    if start > today:
        return (today, today)  # 已是最新
    return (start, today)
```

**注意**：检测必须在 fetch 之前执行，否则会误判为"已有数据 → 增量"，导致首次全量数据被错误识别。

## 五、幂等写入：复合主键 + INSERT OR IGNORE

SQLite 建表时把 `(date, stock)` 设为复合主键，配合 `INSERT OR IGNORE` 实现「存在则跳过」：

```sql
CREATE TABLE IF NOT EXISTS kline (
    date TEXT NOT NULL,
    stock TEXT NOT NULL,
    open REAL, high REAL, low REAL, close REAL,
    pre_close REAL, volume INTEGER,
    PRIMARY KEY (date, stock)
);
```

Python 写入：

```python
import sqlite3

def upsert_with_statcheck(conn, rows):
    """rows = [(date, stock, o, h, l, c, pc, v), ...]
    返回 (inserted, skipped)。"""
    before = conn.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()
    after = conn.execute("SELECT COUNT(*) FROM kline").fetchone()[0]
    inserted = after - before
    skipped = len(rows) - inserted
    return inserted, skipped
```

**两重保险**：

- `PRIMARY KEY (date, stock)`：数据库层保证唯一
- `INSERT OR IGNORE`：SQL 层安全跳过

任意并发场景（脚本重跑、调度器重发、分布式多实例）都不会污染数据。

## 六、调用真实 API：区间补采

```python
from mairui import Client
import time

DEMO_LICENCE = "LICENCE-66D8-9F96-0C7F0FBCD073"

def fetch_history(stock: str, start: str, end: str, licence: str) -> list:
    """拉区间日线（断点续传 + fallback）。"""
    for attempt in range(2):
        try:
            cli = Client(licence)
            return cli.stock_history(
                code=stock, st=start, et=end,
                period="d", dividend="f"
            ) or []
        except Exception as e:
            print(f"[retry {attempt+1}] {stock} {start}~{end}: {str(e)[:80]}")
            time.sleep(0.5)
    return []
```

**关键参数**：

- `code`：股票代码（注意 SDK 是 `code=` 不是 `stock=`）
- `st` / `et`：起止日期 `YYYYMMDD` 或 `YYYY-MM-DD`
- `period="d"`：日线（`m1` 分钟线需单独文档）
- `dividend="f"`：前复权（不复权传 `n`，后复权传 `b`）

## 七、回填校验：count / first / last + 缺口扫描

拉完数据一定要校验，不能「拉完就当搞定」。三个最小校验点：

```python
def validate(conn, stock: str, today: str):
    count = conn.execute("SELECT COUNT(*) FROM kline WHERE stock=?",
                          (stock,)).fetchone()[0]
    first = conn.execute("SELECT MIN(date) FROM kline WHERE stock=?",
                          (stock,)).fetchone()[0]
    last = conn.execute("SELECT MAX(date) FROM kline WHERE stock=?",
                          (stock,)).fetchone()[0]
    return {"stock": stock, "count": count, "first": first, "last": last}
```

更进一步，扫描「预期交易日 vs 实际入库」的差异，发现潜在缺口：

```python
def detect_missing(conn, stock: str, today: str) -> list:
    """粗略按工作日扫缺口（建议生产替换为交易日历）。"""
    have = set(r[0] for r in
               conn.execute("SELECT date FROM kline WHERE stock=?",
                            (stock,)).fetchall())
    if not have:
        return []
    last_date = max(have)
    expected = []
    from datetime import datetime, timedelta
    d = datetime.strptime(last_date, "%Y-%m-%d")
    end = datetime.strptime(today, "%Y-%m-%d")
    while d <= end:
        if d.weekday() < 5:  # 周一到周五
            ds = d.strftime("%Y-%m-%d")
            if ds not in have:
                expected.append(ds)
        d += timedelta(days=1)
    return expected
```

## 八、完整流程与实测

把上面 4 块串起来：

```python
def backfill_one(conn, stock: str, today: str, licence: str):
    was_empty = get_max_date(conn, stock) is None
    start, end = detect_gap(conn, stock, today)
    if start > end:
        return {"stock": stock, "mode": "no_gap", "fetched": 0,
                "inserted": 0, "skipped": 0}

    print(f"[fetch] {stock} {start} ~ {end} ...")
    raw = fetch_history(stock, start, end, licence)
    rows = []
    for r in raw:
        date = r.get("t", "")[:10]
        if not date or date < start or date > end:
            continue
        rows.append((date, stock, r.get("o"), r.get("h"), r.get("l"),
                     r.get("c"), r.get("pc"), r.get("v")))

    if rows:
        inserted, skipped = upsert_with_statcheck(conn, rows)
    else:
        inserted, skipped = 0, 0

    mode = "first_time" if was_empty else "incremental"
    return {"stock": stock, "mode": mode, "gap": f"{start}~{end}",
            "fetched": len(rows), "inserted": inserted, "skipped": skipped}
```

实测运行（演示证，3 只股票）：

```
第 1 轮（首次拉取）：
  600519: mode=first_time, gap=2025-09-01~2025-12-01,
          fetched=50, inserted=50, skipped=0
  000001: mode=first_time, gap=2025-09-01~2025-12-01,
          fetched=50, inserted=50, skipped=0
  300750: mode=first_time, gap=2025-09-01~2025-12-01,
          fetched=50, inserted=50, skipped=0

第 2 轮（再次跑、完全幂等）：
  600519: mode=incremental, gap=2025-12-01~2025-12-01,
          fetched=1, inserted=0, skipped=1
  000001: mode=incremental, gap=2025-12-01~2025-12-01,
          fetched=1, inserted=0, skipped=1
  300750: mode=incremental, gap=2025-12-01~2025-12-01,
          fetched=1, inserted=0, skipped=1

回填校验：
  600519: count=50 first=2025-09-15 last=2025-12-01
  000001: count=50 first=2025-09-15 last=2025-12-01
  300750: count=50 first=2025-09-15 last=2025-12-01

节假日缺口检测：3 只均 0 缺口
```

总结：

- 第 1 轮：每只 50 条入库，共 150 行
- 第 2 轮：每只 fetch=1（区间只有 12-01 一天）→ 全部 PR 命中 → skipped=1
- 最终：每只 50 条 / first=2025-09-15 / last=2025-12-01，三只一致

## 九、常见坑

1. **「今天」硬编码死的**：`today = "2025-12-01"` 只适合演示。生产环境必须 `today = datetime.now().strftime("%Y-%m-%d")` 或从外部配置传入，否则补采永远跑不到真实今天。
2. **节假日判断过粗**：用 `d.weekday() < 5` 判断「周一到周五」会把节假日误判为缺口。建议接入交易接口的交易日历或本地缓存的 `trading_days.csv`。
3. **没记录补采日志**：`fetched / inserted / skipped` 必须落库（`task_log` 表），3 个月后出问题能一键回放。详见 #02 APScheduler 八节。
4. **多实例并发补采**：分布式部署同一份代码会重复拉数据。简单方案是 Redis `SETNX` 加锁，复杂方案上 APScheduler `SQLAlchemyJobStore` 统一调度。
5. **演示证的范围错觉**：演示证只返回最近 60 交易日（截至 2025-12-01），生产证书可拉到 5 年历史。补采脚本逻辑通用，但起始日期按证书档位调整。

## 十、小结

历史数据补采的工程核心是三件事：

- **断点**：`MAX(date)` 查出缺口起点，`start = last + 1`
- **区间**：`st` / `et` 控制 API 拉取范围，不重不漏
- **幂等**：复合主键 + `INSERT OR IGNORE`，并行/重跑都不污染数据

这套方案对回测、监控、生产实盘三个场景全都适用：是策略计算的「上游数据保障」。

下一篇（#08）我们收尾——「量化系统工程上线 checklist」：从脚本到生产的最小闭环，把环境隔离 / 日志轮转 / 备份 / 灰度 / 回滚的兜底动作清点一遍，让你的脚本真正上生产不翻车。

---

**免责声明**：本文仅供技术学习交流，不构成任何投资建议。量化策略回测表现不代表未来收益，投资有风险，决策需谨慎。

**代码与文档**：https://github.com/MaiRuiApi


系列前情：#06 我们搭好了「数据  配置  日志 / 策略」的最小可运行项目骨架。本篇解决「数据有缺口怎么办」——演示证临时切换、断网错过、新上市股票，都要靠断点续传 + 幂等写入补回来。

【Python量化系统工程实战 #07】数据缺口怎么补？断点续传 + 区间补采 + 幂等写入的回填方案

人工智能

代码解释

数据库

Python

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

社区新版编辑器体验调研

【Python量化系统工程实战 #07】数据缺口怎么补？断点续传 + 区间补采 + 幂等写入的回填方案-腾讯云开发者社区-腾讯云

【Python量化系统工程实战 #07】数据缺口怎么补？断点续传 + 区间补采 + 幂等写入的回填方案

【Python量化系统工程实战 #07】数据缺口怎么补？断点续传 + 区间补采 + 幂等写入的回填方案

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐