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

我曾经遇到过一个很典型的问题。

一个策略回测的时候表现非常漂亮，年化收益、最大回撤、换手率都在合理范围内。代码本身没有报错，数据也能正常加载，K 线数量甚至比策略需要的还多。

但策略进入模拟盘以后，结果明显不一样。

我一开始怀疑的是策略逻辑，后来才发现真正的问题出在数据。

当时我把数据按照交易日重新排序，逐个检查以后发现，有些交易日的数据存在缺失；部分股票在公司行为发生以后，历史价格与当前价格的衔接也没有按照同一种复权口径处理。

这件事情让我意识到一个问题：

“API 能返回数据”和“这批数据可以用于量化研究”，完全是两回事。

后来我把股票数据质量拆成几个可以量化检查的指标：

我现在判断一个股票数据源，基本不会先看它“有多少字段”，而是先看它能不能经得起这些检查。

拿到一份 DataFrame，只要行数够，就认为数据没问题。

例如某只股票理论上应该有 20 个交易日，但接口返回了 18 条。如果直接计算收益率，Python 不会报错。

问题只会在最终结果里体现出来。

下面是我现在比较常用的基础检查函数。

import pandas as pd


def check_basic_quality(df: pd.DataFrame) -> dict:
    """检查股票日线数据的基础质量。"""

    if df is None or df.empty:
        return {
            "passed": False,
            "reason": "数据为空"
        }

    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:
        return {
            "passed": False,
            "reason": f"缺少字段: {missing_columns}"
        }

    data = df.copy()

    # 检查重复交易日
    duplicate_count = data["trade_date"].duplicated().sum()

    # 检查关键字段空值
    null_count = data[
        ["open", "high", "low", "close", "volume"]
    ].isnull().sum().sum()

    # 检查日期是否已经排序
    dates = pd.to_datetime(data["trade_date"])
    sorted_ok = dates.is_monotonic_increasing

    return {
        "passed": duplicate_count == 0 and null_count == 0 and sorted_ok,
        "rows": len(data),
        "duplicate_dates": int(duplicate_count),
        "null_values": int(null_count),
        "sorted": bool(sorted_ok)
    }


if __name__ == "__main__":
    print("请将接口返回的 DataFrame 传入 check_basic_quality()")


我在实际项目中不会只看一个 passed，而是会把每个指标都记录下来。

因为数据质量不是简单的“好”或者“坏”，而是需要知道

2. 我第二个检查的是 OHLC 的逻辑关系

股票日线数据至少有一个非常基本的数学约束：

最高价不能低于开盘价、收盘价和最低价，最低价也不能高于开盘价、收盘价。

这个规则非常简单，但特别适合发现数据污染。

import pandas as pd


def check_ohlc(df: pd.DataFrame) -> pd.DataFrame:
    """返回所有 OHLC 逻辑异常的记录。"""

    data = df.copy()

    data["invalid_ohlc"] = (
        (data["high"] < data["open"]) |
        (data["high"] < data["close"]) |
        (data["high"] < data["low"]) |
        (data["low"] > data["open"]) |
        (data["low"] > data["close"])
    )

    return data[data["invalid_ohlc"]].copy()


if __name__ == "__main__":
    sample = pd.DataFrame({
        "trade_date": ["2026-01-02", "2026-01-05"],
        "open": [100, 101],
        "high": [105, 99],
        "low": [98, 100],
        "close": [103, 102],
    })

    bad_rows = check_ohlc(sample)

    if bad_rows.empty:
        print("OHLC 检查通过")
    else:
        print("发现 OHLC 异常:")
        print(bad_rows.to_string(index=False))


这类检查看起来基础，但我认为非常重要。

因为真正进入生产以后，数据问题往往不是“整张表全部错误”，而是某几个时间点出现异常。

如果没有自动质检，这种异常很容易混进回测数据。

3. 我第三个检查的是异常收益率

假设昨天收盘 100，今天收盘 150。

这并不一定是错误，因为股票可能发生重大事件。

但是如果一个普通股票连续几个交易日出现几十甚至几百个百分点的变化，我至少应该让数据质量系统把它标记出来。

我一般不会直接把异常收益当成错误，而是标记成“需要人工复核”。

import pandas as pd


def detect_price_jumps(df: pd.DataFrame, threshold: float = 0.20):
    """检测收盘价日收益超过阈值的数据。"""

    data = df.copy()

    data["trade_date"] = pd.to_datetime(data["trade_date"])
    data = data.sort_values("trade_date").reset_index(drop=True)

    data["return"] = data["close"].pct_change()

    abnormal = data[
        data["return"].abs() > threshold
    ].copy()

    return abnormal


if __name__ == "__main__":
    sample = pd.DataFrame({
        "trade_date": [
            "2026-01-02",
            "2026-01-05",
            "2026-01-06",
            "2026-01-07"
        ],
        "close": [100, 101, 130, 131]
    })

    abnormal = detect_price_jumps(sample)

    print("异常价格变化:")
    if abnormal.empty:
        print("没有发现超过阈值的异常")
    else:
        print(abnormal.to_string(index=False))


异常值检测不是简单删除异常值。

如果我看到一个异常价格变化，我首先要问：

这是市场真实发生的事情，还是数据源的问题？

这两者在量化系统里的处理方式完全不同。

方案一：自己维护多个免费数据源

不同数据源的字段命名、交易日、复权方式和更新节奏都可能不同。

最麻烦的是，同一个股票同一天的价格可能存在微小差异。

策略研究阶段这些差异看起来不大，但经过长期复利以后，可能影响最终结果。

一开始感觉非常灵活，但维护成本很快超过了预期。

网页结构一变化，解析逻辑就需要修改。

而且爬虫解决的是“怎么拿数据”，并没有自动解决“数据是否可靠”。

所以我后来逐渐把关注点从“数据获取”转向“数据质量”。

方案三：统一数据接口 + 自己做质量层

我在项目里使用过一个叫 QuantDash 的工具作为数据获取层。它支持 A 股、美股和港股，标的代码使用统一格式，例如 

对我来说，更重要的不是“一个接口支持多少股票”，而是拿到 DataFrame 以后，我可以把自己的数据质量检查层接在后面。

数据源负责提供数据，我自己的系统负责决定数据能不能进入回测。

四、具体实现：用 Python 建立股票数据质量检查

下面是我现在比较推荐的一种最小实现。

首先通过环境变量读取 API Key，然后获取数据。

import os
import sys

import pandas as pd
from quantdash import QuantDash


def get_daily_data(symbol: str, count: int = 100) -> pd.DataFrame:
    """获取指定股票的日线数据。"""

    api_key = os.getenv("QUANTDASH_API_KEY")

    if not api_key:
        raise RuntimeError(
            "未设置 QUANTDASH_API_KEY 环境变量"
        )

    qd = QuantDash(api_key=api_key)

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

    if df is None or df.empty:
        raise RuntimeError(
            f"{symbol} 未获取到日线数据"
        )

    return df


def quality_report(df: pd.DataFrame) -> dict:
    """生成股票数据质量报告。"""

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

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

    if missing_columns:
        return {
            "passed": False,
            "errors": [
                f"缺少字段: {missing_columns}"
            ]
        }

    data = df.copy()

    data["trade_date"] = pd.to_datetime(
        data["trade_date"]
    )

    errors = []

    if data.empty:
        errors.append("数据为空")

    if data["trade_date"].duplicated().any():
        errors.append("存在重复交易日")

    if not data["trade_date"].is_monotonic_increasing:
        errors.append("交易日期没有按升序排列")

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

    if data[price_columns].isnull().any().any():
        errors.append("价格字段存在空值")

    if data["volume"].isnull().any():
        errors.append("成交量存在空值")

    invalid_ohlc = (
        (data["high"] < data["open"]) |
        (data["high"] < data["close"]) |
        (data["high"] < data["low"]) |
        (data["low"] > data["open"]) |
        (data["low"] > data["close"])
    )

    if invalid_ohlc.any():
        errors.append(
            f"发现 {int(invalid_ohlc.sum())} 条 OHLC 异常"
        )

    negative_volume = data["volume"] < 0

    if negative_volume.any():
        errors.append("发现负成交量")

    data["return"] = data["close"].pct_change()

    abnormal_return = data["return"].abs() > 0.50

    if abnormal_return.any():
        errors.append(
            f"发现 {int(abnormal_return.sum())} 个超过 50% 的价格变化，"
            "建议人工复核"
        )

    return {
        "passed": len(errors) == 0,
        "rows": len(data),
        "start": str(data["trade_date"].min().date()),
        "end": str(data["trade_date"].max().date()),
        "errors": errors
    }


def main():
    symbol = "600519.SH"

    try:
        df = get_daily_data(symbol, count=100)

        report = quality_report(df)

        print(f"标的: {symbol}")
        print(f"数据条数: {report.get('rows', 0)}")
        print(f"开始日期: {report.get('start', '')}")
        print(f"结束日期: {report.get('end', '')}")
        print(f"质量检查通过: {report['passed']}")

        if report["errors"]:
            print("发现的问题:")
            for error in report["errors"]:
                print(f"- {error}")

        print("\n最近 5 条数据:")
        columns = [
            "trade_date",
            "open",
            "high",
            "low",
            "close",
            "volume"
        ]

        print(
            df[columns]
            .tail(5)
            .to_string(index=False)
        )

    except Exception as exc:
        print(f"数据检查失败: {exc}")
        sys.exit(1)


if __name__ == "__main__":
    main()


这个程序的核心思想非常简单：

数据源 → 原始数据 → 质量检查 → 标准化 → 回测数据集 → 策略。

复权是股票数据质量里最容易被低估的问题之一。

我以前遇到过一个非常典型的情况：

同一策略在不同机器上回测，代码一样，数据时间范围也一样，但结果却不同。

最后定位到的原因之一就是复权口径不同。

import os

from quantdash import QuantDash


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

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

    qd = QuantDash(api_key=api_key)

    try:
        symbol = "600519.SH"

        raw_df = qd.klines.get(
            symbol,
            period="1d",
            count=20,
            adjust="none",
            to_dataframe=True
        )

        forward_df = qd.klines.get(
            symbol,
            period="1d",
            count=20,
            adjust="forward",
            to_dataframe=True
        )

        if raw_df.empty or forward_df.empty:
            print("未获取到完整数据")
            return

        print("不复权:")
        print(
            raw_df[
                ["trade_date", "close"]
            ].tail(5).to_string(index=False)
        )

        print("\n前复权:")
        print(
            forward_df[
                ["trade_date", "close"]
            ].tail(5).to_string(index=False)
        )

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


if __name__ == "__main__":
    main()


收益率计算和绝对价格分析，不一定应该使用同一种数据。

比例复权更适合历史收益率计算；如果研究绝对价差，则需要明确自己采用的是哪一种复权方式。

最重要的是，整个回测过程必须固定口径。

从“能下载”到“敢回测”：股票数据源质量到底该怎么判断？

编程语言

算法

量化策略中数据可信度比数据获取更关键，本文从完整性、OHLC逻辑、异常收益率、复权一致性等维度，提供Python实现的股票数据质检方案，并展示如何用QuantDash搭建可落地的数据质量检查流程，确保回测与实盘数据口径统一。

数据一致性

数据完整性

数据治理

数据接口

数据流

Python

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

从“能下载”到“敢回测”：股票数据源质量到底该怎么判断？

从“能下载”到“敢回测”：股票数据源质量到底该怎么判断？

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐