一、问题场景：我在量化实盘中遇到的真实困境

我曾经维护过一套日频选股策略。

策略本身并不复杂：每天收盘后计算过去一段时间的收益率、波动率和成交量指标，然后筛选股票进入下一交易日的候选池。

有一次，我在检查策略回测结果的时候发现一个很奇怪的现象。

同一份策略代码，我上午跑出来的结果和前一天保存的结果出现了差异。

最开始我以为是代码改动，后来用 Git 对比了一遍，策略代码没有变化。于是我开始怀疑数据。

我随机抽了一只股票检查历史行情，发现某个时间区间少了一根日 K。

问题表面上看只是少了一天，但对于滚动指标来说，影响并不止一天。

假设策略计算过去 20 个交易日收益率：

R20=PtPt−20−1R_{20}=\frac{P_t}{P_{t-20}}-1

如果中间缺了一根 K 线，简单地使用前后两条记录计算，程序实际上会把“19 个真实交易日”误认为“20 个交易日”。

更麻烦的是，这种错误通常不会触发 Python 异常。

这才是我认为量化数据缺口最危险的地方：

很多数据错误不是程序报错，而是程序正常运行并给出一个看起来合理的错误答案。

后来我把数据缺口分成了三类：

从那以后，我不再把“接口成功返回数据”当成“数据正确”。

我后来总结了一下，历史行情缺口通常来自几个不同层面。

1. 请求本身没有覆盖完整时间范围

我以前写数据下载脚本时，经常直接使用 count。

df = qd.klines.get(
    "600519.SH",
    period="1d",
    count=100,
    to_dataframe=True,
)


这种方式适合快速获取最近 N 根数据，但如果我要做历史数据补全，时间范围通常更加重要。

因此我后来更习惯使用 start_time 和 end_time。

我可以明确告诉数据管道“我要检查哪一段时间”，而不是“我要多少条数据”。

比如一个周末没有股票行情，我以前的简单检查逻辑可能会认为：

后来我才意识到，量化数据完整性检查必须基于交易日集合，而不是简单地对日期做加一操作。

预期交易日集合 → 实际数据日期集合 → 集合差集。

MissingDays=ExpectedDays−ActualDaysMissingDays=ExpectedDays-ActualDays

真正需要关注的是 MissingDays，而不是简单检查日期是否连续。

这是我现在做数据管道时非常看重的一条原则。

if not df.empty:
    print("数据获取成功")


这个判断只能说明接口返回了至少一行数据。

因此，我后来把“接口层成功”和“数据层成功”完全分开处理。

data/
├── 600519.SH.csv
├── 000001.SZ.csv
└── 000858.SZ.csv


一旦发现缺口，我还得自己解决数据源、日期范围、重复写入和版本管理。

如果股票池从 100 只扩展到几千只，这套方式很快就会变成维护负担。

但我很快发现，爬虫解决的是“如何拿到网页数据”，并没有真正解决“如何保证量化数据的一致性”。

一旦网页结构变化，整个流程就可能失效。

所以后来我逐渐把数据获取和数据质量控制拆开。

我自己的程序负责判断数据是否可信。

方案三：统一数据接口 + 本地质量校验

在我的研究环境里，我后来使用 QuantDash 获取行情，再在本地 DataFrame 层增加质量检测。

它支持统一的证券代码格式，例如：

对于我这种需要同时研究 A 股、美股、港股的环境，统一代码格式可以减少不少适配代码。

我更看重的是它可以直接返回 Pandas DataFrame。

因为对量化研究来说，真正重要的并不是“接口返回了 JSON”，而是：

拿到数据以后，我能不能立刻进入数据清洗、检查和因子计算流程。

代码本身不依赖特定策略，可以直接作为数据质量检查脚本的基础。

import os
import datetime
from typing import List

import pandas as pd
from quantdash import QuantDash


def to_timestamp_ms(date_text: str) -> int:
    """将 YYYY-MM-DD 转换为毫秒时间戳。"""
    dt = datetime.datetime.strptime(date_text, "%Y-%m-%d")
    return int(dt.timestamp() * 1000)


def check_kline_quality(df: pd.DataFrame) -> List[str]:
    """检查 K 线 DataFrame 的基础数据质量。"""
    problems = []

    if df.empty:
        problems.append("返回结果为空")
        return problems

    required_columns = [
        "symbol",
        "trade_date",
        "open",
        "high",
        "low",
        "close",
        "volume",
    ]

    missing_columns = [
        column for column in required_columns
        if column not in df.columns
    ]

    if missing_columns:
        problems.append(
            f"缺少字段: {', '.join(missing_columns)}"
        )
        return problems

    if df["trade_date"].duplicated().any():
        problems.append("发现重复交易日期")

    price_columns = ["open", "high", "low", "close"]

    if df[price_columns].isna().any().any():
        problems.append("OHLC 字段存在空值")

    if df["volume"].isna().any():
        problems.append("volume 字段存在空值")

    invalid_price = (
        (df["high"] < df["low"])
        | (df["high"] < df["open"])
        | (df["high"] < df["close"])
        | (df["low"] > df["open"])
        | (df["low"] > df["close"])
    )

    if invalid_price.any():
        problems.append("发现 OHLC 价格关系异常")

    return problems


def main() -> None:
    api_key = os.getenv("QUANTDASH_API_KEY")

    if not api_key:
        raise RuntimeError(
            "未检测到 QUANTDASH_API_KEY，请先配置环境变量"
        )

    qd = QuantDash(api_key=api_key)

    start_date = "2026-05-01"
    end_date = "2026-06-01"

    start_time = to_timestamp_ms(start_date)
    end_time = to_timestamp_ms(end_date)

    try:
        df = qd.klines.get(
            "600519.SH",
            period="1d",
            start_time=start_time,
            end_time=end_time,
            to_dataframe=True,
        )

        if df.empty:
            print("未获取到数据")
            return

        df = df.sort_values("trade_date").reset_index(drop=True)

        problems = check_kline_quality(df)

        print("数据行数:", len(df))
        print("日期范围:", df["trade_date"].min(), "->", df["trade_date"].max())

        if problems:
            print("发现数据质量问题:")
            for problem in problems:
                print("-", problem)
        else:
            print("基础数据质量检查通过")

        print("\n最近的数据:")
        print(
            df[
                [
                    "trade_date",
                    "open",
                    "high",
                    "low",
                    "close",
                    "volume",
                ]
            ].tail(5).to_string(index=False)
        )

    except Exception as exc:
        print(f"请求失败: {exc}")


if __name__ == "__main__":
    main()


这里有一个我认为非常重要的工程原则：

数据检查逻辑应该独立于策略逻辑。

不要把数据质量判断写在某个因子函数里面。

数据获取
   ↓
数据质量检查
   ↓
数据清洗
   ↓
数据落盘
   ↓
因子计算
   ↓
回测


数据获取
   ↓
直接计算因子
   ↓
发现结果不对
   ↓
回头排查数据


后者实际上是在拿策略结果充当数据质量监控系统。

量化研究中如何处理数据缺口：从发现缺失到自动补洞的完整实践

编程语言

算法

大数据

量化回测结果异常？很可能不是策略问题，而是历史行情数据存在缺口。本文通过Python实战，演示如何利用交易日连续性、字段完整性和异常值三层校验，自动发现并修复缺失日K，避免“看似正常”的错误结果，保障因子计算与回测的可靠性。

数据完整性

数据清洗

数据管道

数据接口

数据结构

Python

JSON

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

量化研究中如何处理数据缺口：从发现缺失到自动补洞的完整实践

量化研究中如何处理数据缺口：从发现缺失到自动补洞的完整实践

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐