这件事，其实没有一个简单答案。对我来说，真正重要的不是“接口能不能返回一张 Pandas DataFrame”，而是这张 DataFrame 能不能经得起回测、策略运行和实盘的连续验证。

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

我曾经维护过一个股票量化策略，最初回测结果非常漂亮。

策略每天收盘后计算信号，第二天开盘执行。最开始我用的数据源 A 做历史回测，结果显示策略在两年测试周期里表现不错。

策略每天正常运行，API 也没有报错，数据表面看起来完全正常，但我把实盘日志和历史回测逐日对比时，发现部分股票在除权除息附近出现了明显差异。

当时最容易犯的错误，是认为“策略代码有 Bug”。

我第一反应也是检查指标计算、排序、交易成本和信号生成逻辑。结果排查了一圈以后发现，问题并不在策略，而在输入数据。

同一只股票，同一个交易日，我拿到的价格如果采用不同复权方式，计算出来的收益率和技术指标都会发生变化。

这件事之后，我给自己的数据源选择增加了一条原则：

数据源首先是基础设施，其次才是数据接口。

我现在评估一个量化数据源，通常不会只问“有没有日线”，而会重点看下面 6 个指标：

二、问题根因分析：为什么“有数据”远远不够

1. 完整性：缺一天数据，指标就可能全部变形

我曾经遇到过一只股票历史数据中间缺少交易日的情况。

单独打开 K 线看，根本不明显。

但策略计算 20 日均线时，实际上变成了“最近 20 条记录”，而不是“最近 20 个交易日”。

所以我现在拿到历史数据后，会首先检查：

下面是我现在常用的一段检查代码。

import os
from datetime import datetime

import pandas as pd
from quantdash import QuantDash


def load_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)

    try:
        df = qd.klines.get(
            symbol,
            period="1d",
            count=count,
            to_dataframe=True,
        )
    except Exception as exc:
        raise RuntimeError(f"获取数据失败: {exc}") from exc

    if df.empty:
        raise ValueError(f"{symbol} 没有返回任何数据")

    return df


def check_data_quality(df: pd.DataFrame) -> None:
    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:
        raise ValueError(f"缺少字段: {missing_columns}")

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

    if data["trade_date"].duplicated().any():
        print("警告：发现重复交易日期")

    if not data["trade_date"].is_monotonic_increasing:
        print("警告：交易日期不是升序")

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

    for column in price_columns:
        null_count = int(data[column].isna().sum())
        if null_count > 0:
            print(f"警告：{column} 存在 {null_count} 个空值")

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

    if invalid_price.any():
        print("警告：发现 OHLC 价格关系异常")

    print(f"数据行数: {len(data)}")
    print(f"起始日期: {data['trade_date'].min().date()}")
    print(f"结束日期: {data['trade_date'].max().date()}")


def main() -> None:
    try:
        df = load_daily_data("600519.SH", count=100)
        check_data_quality(df)
        print(df.tail().to_string(index=False))
    except Exception as exc:
        print(f"数据检查失败: {exc}")


if __name__ == "__main__":
    main()


这段代码本身并不复杂，但它体现了我现在对数据源的一个基本要求：

数据获取和数据校验必须是两个步骤。

不能因为 API 返回 200，就默认数据一定正确。

2. 准确性：价格、成交量和交易日历都要看

我后来发现，数据准确性其实比接口稳定性更难发现。

数据错了一条，反而可能连续几个月都不知道。

因此我现在会针对价格字段做基本的逻辑检查，例如最高价不能低于开盘价，最低价不能高于收盘价。

对于更严格的生产环境，还应该增加：

3. 时效性：实时行情和历史 K 线不是一回事

做日频策略时，我以前认为日 K 线已经够用了。

“数据更新时间”本身就是数据质量指标。

例如策略在 14:55 运行，如果拿到的行情实际上还是几分钟前的数据，那么策略逻辑虽然正确，交易结果仍然可能完全不同。

如果策略对时效性敏感，我会进一步区分：

import os

from quantdash import QuantDash


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

    if not api_key:
        print("请先设置 QUANTDASH_API_KEY")
        return

    qd = QuantDash(api_key=api_key)

    try:
        df = qd.quotes.get(
            symbols=["600519.SH", "000001.SZ"],
            to_dataframe=True,
        )

        if df.empty:
            print("没有获取到实时行情")
            return

        columns = [
            "symbol",
            "last_price",
            "prev_close",
            "volume",
            "ext.name",
            "ext.change_pct",
        ]

        available = [column for column in columns if column in df.columns]
        print(df[available].to_string(index=False))

    except Exception as exc:
        print(f"实时行情获取失败: {exc}")


if __name__ == "__main__":
    main()


对我来说，这里最重要的不是某个具体 API，而是

历史数据和实时数据应该尽可能采用统一的数据模型

优点是成本低，缺点是维护成本非常高。网页结构一变，整个数据链路就可能失效。

第二种是多个免费数据接口拼接。

刚开始感觉非常灵活，但很快遇到字段定义不同、时间格式不同、股票代码格式不同的问题。

第三种是自己维护完整的数据仓库。

这个方案从架构角度最理想，但需要处理下载、增量更新、数据修复、复权因子、任务调度、失败重试等大量工程问题。

后来在一个项目中，我开始使用 QuantDash 作为其中的数据获取层。

我比较看重的是它可以直接返回 Pandas DataFrame，同时统一使用类似 600519.SH、000001.SZ、AAPL.US、00700.HK 这样的代码格式，并且 K 线接口提供不同复权方式。

这让我少写了一层“不同数据源转换成统一 DataFrame”的适配代码。

四、具体实现：把数据质量检查变成固定流程

我现在比较推荐把数据源评价做成一个固定的 Pipeline：

数据获取
   ↓
字段检查
   ↓
时间检查
   ↓
价格逻辑检查
   ↓
复权方式确认
   ↓
异常值检测
   ↓
落库
   ↓
策略使用


下面是一个更接近实际工程的例子。

import os

import pandas as pd
from quantdash import QuantDash


def get_data(symbol: str) -> pd.DataFrame:
    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(
            symbol,
            period="1d",
            count=250,
            adjust="forward",
            to_dataframe=True,
        )
    except Exception as exc:
        raise RuntimeError(f"API 请求失败: {exc}") from exc

    if df.empty:
        raise ValueError("API 返回空数据")

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

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

    if df[required].isna().any().any():
        raise ValueError("OHLCV 存在空值")

    if (df["high"] < df["low"]).any():
        raise ValueError("发现 high < low 的异常数据")

    return df


def calculate_signal(df: pd.DataFrame) -> pd.DataFrame:
    result = df.copy()

    result["ma20"] = result["close"].rolling(20).mean()
    result["ma60"] = result["close"].rolling(60).mean()

    result["signal"] = (
        result["ma20"] > result["ma60"]
    ).astype(int)

    return result


def main() -> None:
    try:
        df = get_data("600519.SH")
        result = calculate_signal(df)

        print(
            result[
                [
                    "trade_date",
                    "close",
                    "ma20",
                    "ma60",
                    "signal",
                ]
            ].tail(10).to_string(index=False)
        )

    except Exception as exc:
        print(f"策略数据处理失败: {exc}")


if __name__ == "__main__":
    main()


不是因为前复权“永远正确”，而是因为

策略必须明确自己使用什么价格定义

如果策略研究的是收益率，我通常会优先考虑比例复权。

如果策略研究绝对价差，则需要重新考虑复权方式。

我现在评估一个数据源，至少会看下面几个方面。

第一，不要只测试“正常数据”

因为正常数据最容易拿到，真正考验数据源的是异常场景。

第二，把复权方式写进策略配置

不要让研究员今天默认前复权，明天换成不复权，然后还拿两个结果直接比较。

数据定义必须是策略配置的一部分。

第三，实时行情和历史行情必须做一致性测试

如果一个策略依赖收盘价，那么我会比较：

如果两套数据的字段定义不一致，后面所有回测与实盘比较都会变得困难。

我现在已经不太会问“哪个数据源最好”。

这个数据源是否满足我的策略对完整性、准确性、时效性和一致性的要求？

数据源选择本质上不是采购问题，而是量化系统架构问题。

我目前在使用数据接口时，会把数据获取、数据校验和策略计算严格拆开。QuantDash 对我来说只是其中的数据获取工具，真正重要的是我建立起来的那套数据质量检查流程。

如果一套数据无法通过自己的质量检查，即使价格再便宜、接口再方便，我也不会让它直接进入实盘。

Q1：量化交易数据源最重要的指标是什么？

A：没有单一答案。日频策略通常更关注历史完整性、复权一致性和准确性；高频或盘中策略则必须进一步关注时效性、分钟线质量和盘口数据。

Q2：为什么回测数据一定要关注复权？

A：因为股票发生分红、送股、拆股等公司行为后，价格序列可能出现结构性变化。如果直接使用未复权价格计算长期收益率或技术指标，结果可能与策略真实意图不一致。

Q3：是不是应该同时使用多个数据源？

A：对于关键策略，我比较推荐至少保留一个独立的数据校验来源。但多数据源并不意味着简单拼接，更重要的是定义统一的数据口径和异常处理规则。

我后来越来越觉得，Python 量化数据 API 推荐这件事，其实没有一个简单答案。对我来说，真正重要的不是“接口能不能返回一张 Pandas DataFrame”，而是这张 DataFrame 能不能经得起回测、策略运行和实盘的连续验证。

量化交易数据源到底应该看哪些指标？我用 6 个维度重新评估了一遍数据源

编程语言

人工智能

大数据

量化数据质量是实盘成败核心，本文从完整性、准确性、时效性、一致性、复权处理及工程稳定性六维度解析，通过Python代码演示数据校验流程，强调数据源选择需契合策略需求，而非仅看接口可用性。

数据完整性

数据仓库

数据接口

数据处理

Python

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

量化交易数据源到底应该看哪些指标？我用 6 个维度重新评估了一遍数据源

量化交易数据源到底应该看哪些指标？我用 6 个维度重新评估了一遍数据源

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐