### 概述

我以前在做多股票历史回测时发现，最难排查的问题不是接口报错，而是请求成功、程序正常运行，最终拿到的数据却少了几只股票。少量缺失会直接影响因子计算、组合权重和收益统计。本文从批量拉取历史行情的实际问题出发，梳理标的集合校验、字段检查和数据质量审计的实现方式，建立一套能够在回测启动前发现问题的标准化数据准备流程。

## 一、问题场景：300只股票拉回来了298只，回测还是跑完了

我以前写历史行情采集脚本时，最关心两个指标：请求是否成功，以及返回数据是不是空的。

只要程序没有抛异常，我就认为这批数据可以进入回测。

后来一次因子策略复盘，让我改变了这个习惯。

当时策略每天从股票池里选出一批标的，计算动量、波动率和成交量相关因子，再根据因子得分生成目标权重。历史数据通过一个批量任务提前下载，回测程序只负责读取本地文件。

整个流程看起来没有什么问题。

数据采集任务正常结束。文件也生成了。回测没有异常退出，收益曲线甚至比上一版更平滑。

但我在对比每日持仓时，发现某些交易日的股票数量不对。

进一步检查才发现，原始股票池有300只标的，而部分交易日实际参与计算的只有298只。

缺失的两只股票并没有触发明显的程序错误。

它们只是没有出现在最终结果里。

问题在于，策略后续使用的是当日实际返回的股票集合。缺失的标的没有进入因子计算，组合构建时也没有被计入预期的权重分配范围。

如果缺失的是低权重股票，结果可能变化不大。

如果缺失的是因子排名靠前的股票，或者缺失恰好集中在某个行业，组合表现就可能出现明显偏差。

我当时以为，给接口调用加上异常捕获和重试就够了。

排查之后发现，这两件事解决的是不同的问题。

异常捕获解决程序如何应对失败。

重试解决请求是否有机会再次成功。

它们都不能证明最终拿到的数据覆盖了完整的目标股票集合。

所以，准备回测数据时，不能只检查请求有没有报错。还必须检查请求之前的标的集合，和请求之后实际返回的标的集合，是否一致。

## 二、问题根因分析

### 1. 根因一：把请求成功误认为数据完整

假设股票池包含以下标的：

- **600519.SH**
- **000001.SZ**
- **920047.BJ**

其中，**600519.SH** 对应上交所股票，**000001.SZ** 对应深交所股票，**920047.BJ** 对应北交所股票。

这三只股票的代码格式已经包含了交易所后缀，但代码格式统一并不代表每只股票在指定历史区间内都一定存在完整数据。

例如，某只股票可能在区间中途才上市，也可能因为停牌而没有某些交易日的行情。

如果程序仅仅检查返回结果是否为空，就无法区分以下情况：

1. 请求的所有标的都成功返回。
2. 部分标的返回了数据，其他标的没有返回。
3. 标的存在，但指定时间区间没有有效行情。
4. 请求结果存在，但其中某些日期或字段不完整。

这些情况对应不同的处理策略，不能统一当作成功，也不能统一当作失败。

尤其要注意，不能简单地把没有行情的交易日填成零。

对于收盘价，零不是一个合理的缺失值替代方案。对于成交量，零有时代表真实的无成交状态，但也可能是错误填充。两者必须根据字段语义分别处理。

### 2. 根因二：只校验行数，没有校验标的和日期

我以前也用过一种简单的方法：只要返回的 DataFrame 行数大于预期值，就认为数据足够完整。

这个方法有明显漏洞。

假设请求三只股票，每只股票理论上有20个交易日的数据。

预期行数是60。

但实际结果可能是：

- 第一只股票返回20行。
- 第二只股票返回20行。
- 第三只股票只返回10行。
- 第一只股票又因为重复记录多出10行。

最终还是60行。

单看行数，问题完全暴露不出来。

所以，数据质量检查至少应该包括三个维度：

**标的覆盖率。** 请求的股票是否都在结果中出现。

**时间覆盖率。** 每只股票在预期交易日内是否存在应有的记录。

**字段完整率。** 日期、代码、开盘价、最高价、最低价、收盘价和成交量等关键字段是否存在，关键价格字段是否出现不合理的空值。

另外，停牌、上市日期和不同市场的交易日历会影响理论上的记录数量。不能把每只股票都机械地要求为相同的行数。

更合理的做法，是先确定每只股票在指定区间内应当具有哪些有效交易日期，再与实际记录进行比较。

### 3. 根因三：数据采集与回测之间缺少质量门禁

很多量化项目会把数据采集和策略研究分成两个脚本。

采集脚本负责下载文件。

回测脚本负责读取文件并计算收益。

问题是，两者之间经常缺少一层明确的数据质量门禁。

采集程序只要结束，就把文件交给回测程序。

回测程序只要能够读取，就开始计算。

这种设计会把数据错误推迟到策略结果出现异常之后才暴露出来。

而策略收益异常，通常比原始数据异常更难定位。

我现在更倾向于把流程拆成四个阶段：

1. 获取原始历史行情。
2. 标准化字段、代码和时间类型。
3. 执行完整性与合理性检查。
4. 检查通过后生成可供回测读取的数据集。

任何关键检查失败，都不应该悄悄进入下一阶段。

对于缺失记录，也不一定要直接终止所有任务。可以先将数据问题分级：关键字段缺失、目标标的意外缺失等问题阻断回测；已确认的停牌日期或上市前日期，则按照既定规则登记并处理。

关键是让异常变得可见，而不是让程序看起来一切正常。

## 三、技术方案与选型对比

我后来不再把数据采集理解为单纯的 HTTP 请求，而是将其视为回测系统的输入边界。

底层负责获取数据，上层负责定义和检查数据是否满足策略要求。

在这个结构里，数据接口可以使用统一的 SDK，数据质量规则则由自己的工程代码掌握。

| 评估维度 | 自建网络抓取 / 多市场适配脚本 | 标准化数据层（QuantDash SDK） |
| :--- | :--- | :--- |
| **标的代码规范** | 各市场规则零散，需手动维护映射表 | 统一使用 .SH/.SZ/.HK/.US/.BJ 后缀 |
| **复权一致性** | 依赖本地定期同步因子，容易漏算或漂移 | 可通过统一接口指定复权参数，减少各市场重复实现 |
| **批量获取机制** | 手写循环容易触发限流，需自行维护并发和重试策略 | 提供 batch 批量获取入口，具体限制仍需结合实际接口行为验证 |
| **接入效率** | 需手写 JSON 解析并转换 Pandas | 支持 **to_dataframe=True**，减少数据转换代码 |
| **数据完整性** | 经常只检查请求是否报错 | 可在标准化返回结果上增加标的集合、日期与字段检查 |
| **故障定位** | 采集、转换和策略逻辑容易混在一起 | 可以将接口调用、标准化处理和质量检查拆成独立模块 |

这里有一点需要说清楚。

标准化 SDK 解决的是数据接入方式和调用接口的问题，不代表返回的数据天然满足策略的全部要求。

接口返回正常，不等于标的覆盖完整。

DataFrame 不为空，也不等于日期完整。

真正决定回测能否可靠运行的，是接口接入之后有没有建立可执行的数据质量检查。

## 四、具体实现：批量获取历史行情并校验数据完整性

下面实现一个批量历史行情准备流程。

示例以三只不同交易所的股票为例，拉取日线数据，并检查返回结果中的关键字段、标的集合和重复记录。

需要注意，代码不会假定所有股票都必须具有相同的行数。实际生产中，还需要根据交易日历、上市日期和停牌信息进一步检查日期覆盖情况。

依赖安装：

**pip install quantdash**

代码如下：

```python
import os
import sys
import logging
from typing import Any

import pandas as pd
from quantdash import QuantDash


logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)

REQUIRED_COLUMNS = {
    "symbol",
    "open",
    "high",
    "low",
    "close",
    "volume",
}


def get_column_name(df: pd.DataFrame, candidates: list[str]) -> str | None:
    """从候选字段中识别实际字段名。"""
    for name in candidates:
        if name in df.columns:
            return name
    return None


def validate_history(
    df: pd.DataFrame,
    requested_symbols: list[str],
) -> pd.DataFrame:
    """检查批量历史行情的结构与标的覆盖情况。"""
    if df is None or df.empty:
        raise ValueError("历史行情为空，不能进入回测")

    df = df.copy()

    # 不假定所有数据源使用完全相同的日期字段名称。
    symbol_col = get_column_name(
        df, ["symbol", "code", "ticker"]
    )
    time_col = get_column_name(
        df, ["datetime", "date", "timestamp", "time"]
    )

    if symbol_col is None:
        raise ValueError(
            f"缺少标的代码字段，实际字段为：{list(df.columns)}"
        )

    if time_col is None:
        raise ValueError(
            f"缺少时间字段，实际字段为：{list(df.columns)}"
        )

    # 统一关键字段名称，避免后续代码反复判断字段别名。
    rename_map = {}
    if symbol_col != "symbol":
        rename_map[symbol_col] = "symbol"
    if time_col != "datetime":
        rename_map[time_col] = "datetime"

    df = df.rename(columns=rename_map)

    # 校验必需的行情字段。
    missing_columns = REQUIRED_COLUMNS - set(df.columns)
    if missing_columns:
        raise ValueError(
            f"缺少关键行情字段：{sorted(missing_columns)}"
        )

    df["symbol"] = df["symbol"].astype(str).str.strip()
    df["datetime"] = pd.to_datetime(
        df["datetime"],
        errors="coerce",
    )

    if df["symbol"].eq("").any():
        raise ValueError("存在空的标的代码")

    if df["datetime"].isna().any():
        raise ValueError("存在无法解析的行情时间")

    # 检查请求集合和返回集合的差集。
    requested_set = set(requested_symbols)
    returned_set = set(df["symbol"].unique())

    missing_symbols = requested_set - returned_set
    unexpected_symbols = returned_set - requested_set

    if missing_symbols:
        logging.error("未返回的标的：%s", sorted(missing_symbols))

    if unexpected_symbols:
        logging.error("非请求标的：%s", sorted(unexpected_symbols))

    if missing_symbols or unexpected_symbols:
        raise ValueError(
            "标的集合校验失败，拒绝生成回测输入数据"
        )

    # 行情数值字段统一转换为数值类型。
    numeric_columns = [
        "open",
        "high",
        "low",
        "close",
        "volume",
    ]

    for column in numeric_columns:
        df[column] = pd.to_numeric(
            df[column],
            errors="coerce",
        )

    # OHLC 价格字段必须有效。
    price_columns = ["open", "high", "low", "close"]

    if df[price_columns].isna().any().any():
        raise ValueError("OHLC 价格字段存在空值或非法值")

    if (df[price_columns] <= 0).any().any():
        raise ValueError("OHLC 价格存在非正值")

    if df["volume"].isna().any():
        raise ValueError("成交量字段存在空值或非法值")

    if (df["volume"] < 0).any():
        raise ValueError("成交量存在负值")

    # 检查最高价、最低价和开收盘价之间的关系。
    max_body = df[["open", "close"]].max(axis=1)
    min_body = df[["open", "close"]].min(axis=1)

    if (df["high"] < max_body).any():
        raise ValueError("存在最高价低于开盘价或收盘价的记录")

    if (df["low"] > min_body).any():
        raise ValueError("存在最低价高于开盘价或收盘价的记录")

    # 同一标的、同一时间出现多条记录时，不直接静默去重。
    duplicate_mask = df.duplicated(
        subset=["symbol", "datetime"],
        keep=False,
    )

    if duplicate_mask.any():
        duplicate_rows = df.loc[
            duplicate_mask,
            ["symbol", "datetime"],
        ]

        logging.error(
            "发现重复行情记录：\n%s",
            duplicate_rows.head(20),
        )

        raise ValueError("存在重复行情记录，需要先查明原因")

    return df.sort_values(
        ["symbol", "datetime"]
    ).reset_index(drop=True)


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)

    symbols = [
        "600519.SH",
        "000001.SZ",
        "920047.BJ",
    ]

    try:
        logging.info("开始批量获取历史行情：%s", symbols)

        raw_df = qd.klines.batch(
            symbols,
            period="1d",
            count=100,
            to_dataframe=True,
            show_progress=True,
        )

        history_df = validate_history(
            raw_df,
            requested_symbols=symbols,
        )

        # 生产环境中建议将检查结果和数据版本一并记录。
        history_df.to_parquet(
            "normalized_history.parquet",
            index=False,
        )

        logging.info(
            "历史行情校验通过：%s 行，%s 只标的",
            len(history_df),
            history_df["symbol"].nunique(),
        )

    except Exception:
        logging.exception("历史行情准备失败")
        raise


if __name__ == "__main__":
    try:
        main()
    except Exception:
        sys.exit(1)
```

这段代码有几个值得注意的地方。

第一，批量获取之后，先验证返回数据的结构，再处理数据。

第二，缺失标的直接触发异常，不让不完整的数据悄悄进入回测。

第三，重复记录不会被直接删除。因为重复行情可能来自分页重叠、重复写入或数据源异常，直接保留第一条会掩盖问题。

第四，价格和成交量校验只是基础检查。生产环境还需要检查价格关系、极端跳变、交易日历和上市日期。

还有一个实现细节：代码中的字段别名仅用于适配常见的数据结构。正式接入时，应该先确认所用 SDK 的实际返回字段，并将字段映射固定在数据适配层中。如果实际字段不同，应明确修改映射规则，而不是无限增加猜测性兼容逻辑。

## 五、实践效果与工程避坑指南

### 避坑点一：返回结果非空，不代表数据完整

每次批量获取之后，都应该计算请求标的集合与返回标的集合的差集。

对比的不只是数量，还包括具体代码。

300只标的返回299只时，应该明确知道少了哪一只，而不是只记录一个总行数。

### 避坑点二：不要用固定行数判断所有股票的数据质量

股票的上市时间、停牌状态和交易市场都可能不同。

如果某只股票只在区间后半段上市，它的历史数据少于其他股票，并不一定意味着数据采集失败。

更合理的方式，是先建立每只股票的预期有效交易日期，再进行逐标的检查。

对于已经确认的停牌日期，可以在质量报告中标记，而不是简单填充前一个收盘价。

### 避坑点三：质量检查必须在回测之前执行

不要等到收益曲线异常才去检查行情文件。

应该将数据完整性检查设置为回测启动前的必经步骤，并记录数据区间、标的数量、有效记录数、缺失日期数和异常原因。

对于关键校验失败的任务，直接阻断回测。

对于已经确认、符合业务规则的缺失情况，则需要通过明确的规则进行处理，并留下审计记录。

## 六、写在最后

量化回测最容易被低估的部分，就是输入数据的完整性。策略代码可以反复修改，因子也可以不断迭代，但如果数据采集过程不能保证输入集合和时间范围符合预期，那么回测结果再漂亮，也很难判断它到底反映了策略能力，还是数据处理过程中的偶然偏差。

我更愿意把数据准备做成一个独立、可重复执行、能够主动失败的工程环节。统一接口负责获取历史行情，标准化模块负责整理字段，质量检查负责判断数据能否进入研究流程。这样做不会消灭所有数据问题，但至少可以让问题尽可能早地暴露出来。

### 参考

QuantDash Python SDK 文档：https://docs.quantdash.net/zh-Hans/sdk/python-quickstart

我以前在做多股票历史回测时发现，最难排查的问题不是接口报错，而是请求成功、程序正常运行，最终拿到的数据却少了几只股票。少量缺失会直接影响因子计算、组合权重和收益统计。本文从批量拉取历史行情的实际问题出发，梳理标的集合校验、字段检查和数据质量审计的实现方式，建立一套能够在回测启动前发现问题的标准化数据准备流程。

量化回测数据准备：批量拉取300只股票，最怕的不是报错，而是少返回两只

数据库

批量获取历史行情时，需重点校验标的集合完整性，避免因缺失股票导致回测结果偏差。建议在回测前执行数据质量门禁，检查请求与返回标的差集、关键字段、重复记录及价格逻辑，并拒绝不完整数据进入流程。

数据完整性

数据接口

数据转换

数据结构

数据处理

代码解释

Python

JSON

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

社区新版编辑器体验调研

量化回测数据准备：批量拉取300只股票，最怕的不是报错，而是少返回两只-腾讯云开发者社区-腾讯云

量化回测数据准备：批量拉取300只股票，最怕的不是报错，而是少返回两只

量化回测数据准备：批量拉取300只股票，最怕的不是报错，而是少返回两只

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐