### 📌 摘要

> 我以前把数据获取理解成“调用 API 拿到 DataFrame”，后来真正维护多市场量化系统以后才发现，数据获取只是第一步，真正困难的是验证数据能不能进入回测。本文从数据工程角度，介绍我如何检查标的代码、字段完整性、时间连续性、OHLC 关系、复权口径和批量数据异常，并通过 Python 建立一套可以复用的数据质量检查流程。

**标签：量化交易、Python、数据分析、API、回测**

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

我以前维护量化研究环境的时候，遇到过一个特别典型的问题。

刚开始只有 A 股策略，所以我把很多东西都写死了。

例如：

- 股票代码默认按照 A 股格式处理；
- 时间字段默认是交易日；
- 行情字段默认就是 open、high、low、close、volume；
- 回测数据默认采用某一种复权方式。

当策略开始扩展到美股和港股之后，这些假设逐渐暴露出了问题。

我第一次真正意识到这个问题，是在做跨市场股票池的时候。

研究程序需要同时读取 A 股、美股和港股数据。

如果还是按照原来单市场代码处理，就需要在不同地方增加各种判断。

例如：

“如果是 A 股，就走这里。”

“如果是美股，就换一种代码格式。”

“如果是港股，就再转换一次。”

这种代码短期还能运行，时间一长就很难维护。

更麻烦的是：

**代码能运行，并不代表数据正确。**

后来我开始重新设计数据层，把问题拆成：

**标的识别 → 数据获取 → 数据质量检查 → 标准化 → 回测。**

这时候我才真正理解为什么量化回测首先应该检查数据质量。

---

## 二、问题根因分析

### 1. 标的代码是数据质量的一部分

以前我总觉得股票代码只是一个字符串。

后来发现不是。

在跨市场系统里：

**600519.SH**

和：

**AAPL.US**

以及：

**00700.HK**

本质上都是标的标识，但市场上下文不同。

如果代码映射错误，后面所有数据即使格式完全正确，也没有意义。

因此我现在的数据检查第一步，就是确认：

**symbol 是否符合策略预期。**

---

### 2. 数据结构一致，不代表业务含义一致

Pandas DataFrame 对量化研究非常方便。

但 DataFrame 最大的风险之一，就是太容易让人产生“结构统一”的错觉。

例如两个数据集都有：

- close
- volume

它们就真的可以直接拼起来吗？

我不会直接这么做。

我至少会检查：

- symbol；
- market；
- trade_date 或 trade_time；
- price；
- volume；
- adjust；
- frequency。

尤其是频率。

日线数据的时间字段和分钟线的时间字段不是一回事。

QuantDash Python SDK 同时支持日线和分钟 K 线。

例如日线：

```python
import os
from quantdash import QuantDash

api_key = os.getenv("QUANTDASH_API_KEY")

if not api_key:
    raise RuntimeError("请设置 QUANTDASH_API_KEY")

qd = QuantDash(api_key=api_key)

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

    if df.empty:
        print("没有获取到日线数据")
    else:
        print(
            df[
                [
                    "symbol",
                    "name",
                    "trade_date",
                    "open",
                    "high",
                    "low",
                    "close",
                    "volume",
                ]
            ].to_string(index=False)
        )

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

分钟线则使用不同的时间字段：

```python
import os
from quantdash import QuantDash

api_key = os.getenv("QUANTDASH_API_KEY")

if not api_key:
    raise RuntimeError("请设置 QUANTDASH_API_KEY")

qd = QuantDash(api_key=api_key)

try:
    df = qd.klines.get(
        "600519.SH",
        period="5m",
        count=10,
        to_dataframe=True,
    )

    if df.empty:
        print("没有获取到分钟数据")
    else:
        print(
            df[
                [
                    "symbol",
                    "name",
                    "trade_time",
                    "open",
                    "high",
                    "low",
                    "close",
                    "volume",
                ]
            ].to_string(index=False)
        )

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

我在工程上会刻意把这两类数据分开。

因为：

**trade_date 和 trade_time 的语义不同。**

---

### 3. “数据有多少条”不是质量指标

以前我会检查：

“这个月是不是拿到了 20 条左右的数据？”

现在我不会只这么检查。

我更关心：

- 日期是否重复；
- 日期是否排序；
- 数据是否存在明显缺口；
- OHLC 是否合理；
- 成交量是否异常；
- 不同股票是否存在结构差异。

因为数据量正常，只能证明“有数据”。

不能证明“数据可以用于回测”。

---

### 4. 复权因子应该成为可追溯信息

在股票量化中，公司行为是数据工程里非常重要的一部分。

我现在如果发现某个历史价格发生明显变化，会进一步查看除权因子，而不是直接修改价格。

QuantDash 可以获取除权因子。

```python
import os
import sys
from quantdash import QuantDash


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

    if not api_key:
        print("请设置 QUANTDASH_API_KEY")
        sys.exit(1)

    qd = QuantDash(api_key=api_key)

    try:
        df = qd.klines.ex_factors(
            ["600519.SH"],
            to_dataframe=True,
        )
    except Exception as exc:
        print(f"获取除权因子失败: {exc}")
        sys.exit(1)

    if df.empty:
        print("没有获取到除权因子")
        return

    print(
        df[
            [
                "symbol",
                "trade_date",
                "ex_factor",
            ]
        ].tail(10).to_string(index=False)
    )


if __name__ == "__main__":
    main()
```

这一点对排查历史数据特别有用。

我的经验是：

**遇到异常价格，先找数据原因，再决定是否清洗。**

不要因为一个数字看起来“不符合预期”，就直接改掉。

---

## 三、解决方案探索与选型

### 1. 自己维护多个市场数据源

这是我早期走过的一条路。

A 股一个来源。

美股一个来源。

港股再一个来源。

研究层自己做统一。

刚开始看起来没有什么问题。

但是维护一段时间以后，数据转换代码越来越多。

最典型的问题就是代码映射。

同一个研究任务里，我需要不断判断：

“这个 symbol 是哪个市场？”

“这个数据源返回的代码格式是什么？”

“这个字段是不是同一个含义？”

这些代码和策略本身没有关系，却消耗了大量维护成本。

---

### 2. 自己维护统一数据表

后来我尝试把所有数据先转换成统一表结构。

这一步确实解决了一些问题，但新的问题也出现了：

**统一字段不等于统一语义。**

例如我可以让所有数据都有 close 字段，但如果一个是前复权，一个是不复权，那么统一字段也没有意义。

所以我后来把“数据定义”也纳入数据层。

---

### 3. 使用统一 API 减少接口差异

我后来采用的思路是：

让数据接口尽量统一，自己的数据质量模块负责验证。

在实践中，我使用 QuantDash 的 Python SDK 来获取不同市场的行情数据。

它采用：

**代码.市场后缀**

的形式标识标的。

例如：

**600519.SH**

**000001.SZ**

**AAPL.US**

**00700.HK**

这对我的研究代码比较方便。

因为策略层不用关心每个市场对应哪个独立接口，只需要接收统一格式的数据。

但我还是会强调一点：

**统一接口不等于可以跳过数据检查。**

API 只是解决“怎么拿数据”。

数据质量检查解决的是：

**“拿到的数据能不能用于这个研究任务。”**

---

## 四、具体实现：代码实战

### 1. 建立统一的数据质量检查器

下面是我比较常用的一种简化结构。

```python
import os
import sys
from typing import List
import pandas as pd
from quantdash import QuantDash


def validate_kline_data(
    symbol: str,
    df: pd.DataFrame,
) -> List[str]:
    problems = []

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

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

    if missing:
        problems.append(
            f"{symbol}: 缺少字段 {missing}"
        )
        return problems

    if df.empty:
        problems.append(
            f"{symbol}: 数据为空"
        )
        return problems

    if df["symbol"].nunique() != 1:
        problems.append(
            f"{symbol}: 返回数据包含多个标的"
        )

    if df["trade_date"].duplicated().any():
        problems.append(
            f"{symbol}: 存在重复交易日"
        )

    if not df["trade_date"].is_monotonic_increasing:
        problems.append(
            f"{symbol}: 日期没有按升序排列"
        )

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

    if df[price_columns].isnull().any().any():
        problems.append(
            f"{symbol}: 价格字段存在空值"
        )

    if df["volume"].isnull().any():
        problems.append(
            f"{symbol}: volume 存在空值"
        )

    if (df[price_columns] <= 0).any().any():
        problems.append(
            f"{symbol}: 价格存在非正数"
        )

    high_invalid = (
        df["high"]
        < df[["open", "close"]].max(axis=1)
    )

    low_invalid = (
        df["low"]
        > df[["open", "close"]].min(axis=1)
    )

    if high_invalid.any():
        problems.append(
            f"{symbol}: high 不满足 OHLC 约束"
        )

    if low_invalid.any():
        problems.append(
            f"{symbol}: low 不满足 OHLC 约束"
        )

    return problems


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

    if not api_key:
        print("请设置 QUANTDASH_API_KEY")
        sys.exit(1)

    qd = QuantDash(api_key=api_key)

    symbols = [
        "600519.SH",
        "000001.SZ",
        "000858.SZ",
    ]

    try:
        dfs = qd.klines.batch(
            symbols,
            period="1d",
            count=200,
            adjust="forward",
            to_dataframe=True,
            show_progress=True,
        )
    except Exception as exc:
        print(f"批量获取数据失败: {exc}")
        sys.exit(1)

    total_errors = 0

    for symbol in symbols:
        df = dfs.get(symbol)

        if df is None:
            print(f"{symbol}: API 没有返回数据")
            total_errors += 1
            continue

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

        problems = validate_kline_data(
            symbol,
            df,
        )

        if problems:
            total_errors += len(problems)

            for problem in problems:
                print(f"[ERROR] {problem}")
        else:
            print(
                f"[OK] {symbol}: "
                f"{len(df)} 条数据通过检查"
            )

    if total_errors:
        print(
            f"数据质量检查失败，"
            f"共发现 {total_errors} 个问题"
        )
        sys.exit(1)

    print("全部数据质量检查通过")


if __name__ == "__main__":
    main()
```

我喜欢这种代码结构的原因，是因为它把：

**数据获取**

和：

**数据判断**

分开了。

以后即使数据接口发生变化，我也不需要重写所有质量规则。

---

### 2. 检查跨市场标的

如果策略同时研究 A 股、美股和港股，我还会额外做标的级检查。

```python
import os
import sys
from quantdash import QuantDash


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

    if not api_key:
        print("请设置 QUANTDASH_API_KEY")
        sys.exit(1)

    qd = QuantDash(api_key=api_key)

    symbols = [
        "600519.SH",
        "AAPL.US",
        "00700.HK",
    ]

    try:
        instruments = qd.instruments.get(symbols)
    except Exception as exc:
        print(f"获取标的信息失败: {exc}")
        sys.exit(1)

    if not instruments:
        print("没有获取到标的信息")
        sys.exit(1)

    for item in instruments:
        symbol = item.get("symbol", "")
        name = item.get("name", "")
        exchange = item.get("exchange", "")
        instrument_type = item.get("type", "")

        print(
            f"{symbol:>12s} "
            f"{name:<10s} "
            f"交易所={exchange:<4s} "
            f"类型={instrument_type}"
        )


if __name__ == "__main__":
    main()
```

这个步骤看起来很简单，但我认为它非常重要。

尤其是在股票池动态变化的时候。

如果标的映射错了，后面再精细的数据质量检查也没有意义。

---

### 3. 把数据质量检查放到回测入口

我最终希望整个系统形成这样的调用关系：

**获取数据 → 数据检查 → 数据标准化 → 策略计算 → 回测统计**

而不是：

**获取数据 → 策略计算 → 收益异常 → 回头查数据**

前一种方式最大的好处就是：

**错误会更早暴露。**

---

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

经过实际项目调整以后，我现在会特别注意以下三个问题。

### 避坑一：不要默认所有数据都是“已经清洗好的”

即使 API 返回的是结构化数据，我仍然会做自己的校验。

原因是：

**结构化 ≠ 正确。**

API 可以保证返回格式，但策略需要什么样的数据质量，仍然要由策略工程自己定义。

---

### 避坑二：批量数据一定要逐标的检查

批量请求非常方便。

但批量请求成功，不代表每个标的都成功。

所以我会检查：

- 返回数量；
- symbol 是否匹配；
- 每只股票的记录数；
- 是否存在空 DataFrame；
- 是否存在字段缺失。

---

### 避坑三：复权参数不要隐藏

如果策略依赖前复权数据，我会直接写：

**adjust="forward"**

而不是依赖默认设置。

原因很简单：

半年以后再看代码，我仍然能知道当时的数据口径。

对于量化系统来说：

**可追溯性比少写一个参数更加重要。**

---

## 六、总结与思考

我以前认为量化数据工程最重要的是“把数据拿回来”。

现在我的理解完全不一样。

真正可靠的数据工程应该解决三个问题：

**数据拿得到。**

**数据定义说得清楚。**

**数据进入策略之前能够自动验证。**

尤其当量化策略从单市场扩展到多个市场之后，数据质量的重要性会越来越明显。

我在实践中使用统一数据接口来降低数据访问层的复杂度，同时把数据质量规则留在自己的工程体系里。这样即使未来更换数据来源，我也不会把策略和某一个具体接口强耦合在一起。

所以如果让我重新搭建一套量化回测系统，我会把数据质量检查放在最前面。

不是因为数据比策略更“高级”，而是因为：

> **策略是在数据之上运行的。数据输入错了，后面的收益曲线再漂亮，也没有办法证明策略真的有效。**

## Q&A

### Q1: A 股、美股、港股可以直接使用同一套回测逻辑吗？

A: 可以共享很多研究层代码，但不能默认所有市场的交易规则、交易时间和数据语义完全相同。统一接口主要解决数据访问和标的表示问题，交易规则仍然应该在回测系统中独立处理。

### Q2: 为什么要在回测前检查标的信息？

A: 因为股票代码本身就是数据上下文的一部分。尤其是跨市场研究时，同样的字符串可能在不同数据体系里代表完全不同的标的。确认 symbol、交易所和标的类型，可以避免从数据源层面就发生错误。

我以前把数据获取理解成“调用 API 拿到 DataFrame”，后来真正维护多市场量化系统以后才发现，数据获取只是第一步，真正困难的是验证数据能不能进入回测。本文从数据工程角度，介绍我如何检查标的代码、字段完整性、时间连续性、OHLC 关系、复权口径和批量数据异常，并通过 Python 建立一套可以复用的数据质量检查流程。

量化回测前的数据质量检查：如何建立一套可复用的数据验证流程

编程语言

数据库

后端

量化回测数据质量决定策略有效性。本文从数据工程角度，系统讲解跨市场行情数据的校验方法，包括标的代码映射、字段完整性、时间连续性、OHLC逻辑、复权口径及批量异常检测，并给出基于Python与QuantDash SDK的可复用检查流程，助你构建稳健的量化数据管道。

结构化数据

数据分析

数据接口

数据转换

数据结构

代码解释

Python

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

社区新版编辑器体验调研

量化回测前的数据质量检查：如何建立一套可复用的数据验证流程-腾讯云开发者社区-腾讯云

量化回测前的数据质量检查：如何建立一套可复用的数据验证流程

量化回测前的数据质量检查：如何建立一套可复用的数据验证流程

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐