一、问题场景：我曾经被“回测和实盘不一致”折腾过

我做量化系统的时候，有一个场景让我印象非常深。

当时一个日频策略已经完成回测。

回测逻辑非常简单：每天使用过去一段时间的价格计算指标，生成交易信号。

但策略上线以后，我发现某些交易日的信号和回测结果对不上。

我甚至把同一份策略代码在服务器上重新跑了一遍。

后来我把某一天的输入数据逐字段打印出来，才发现真正的问题：

回测使用的是历史 K 线，而实盘使用的是实时行情，两者在时间点和复权口径上并没有完全统一。

这次之后，我对“数据源应该看哪些指标”有了更具体的答案。

我认为对于量化系统来说，数据源最重要的指标之一，就是：

同一个数据定义能不能贯穿研究、回测和实盘。

1. 最大的问题：你以为自己在比较同一个时间点

回测阶段，我可能使用当天最终收盘价。

但实盘阶段，14:55 时当天还没有收盘。

这两个数据从数学上就不是同一个东西。

因此我现在设计策略时，会首先定义：

这个问题没有定义清楚，后面所有回测结果都可能出现偏差。

2. 复权方式不同，也会造成信号差异

我以前见过一个特别容易忽略的问题：

历史回测使用前复权数据，而实时行情返回的是当前市场价格。

如果策略直接把两个价格序列放到一起计算，就需要明确它们的价格定义是否一致。

QuantDash 的 K 线接口支持不复权、前复权、后复权，以及比例复权和差值复权。

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:
        none_adjust = qd.klines.get(
            "600519.SH",
            period="1d",
            count=10,
            adjust="none",
            to_dataframe=True,
        )

        forward_adjust = qd.klines.get(
            "600519.SH",
            period="1d",
            count=10,
            adjust="forward",
            to_dataframe=True,
        )

        if none_adjust.empty or forward_adjust.empty:
            print("没有获取到完整数据")
            return

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

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

    except Exception as exc:
        print(f"获取历史数据失败: {exc}")


if __name__ == "__main__":
    main()


我现在做策略研究时，会明确记录数据的复权方式，而不是依赖接口默认值。

3. 字段名称相同，不代表字段含义相同

这是数据工程里特别容易被低估的问题。

例如 volume 看起来就是成交量。

但如果两个数据源的单位、统计时间或市场定义不同，直接拼接可能产生非常隐蔽的错误。

因此我在数据层一般会建立自己的标准字段：

symbol
trade_date / trade_time
open
high
low
close
volume


然后所有上层策略只使用这一层标准数据。

我以前尝试过把不同来源的数据分别放进不同目录。

短期看起来没问题，长期一定会出现麻烦。

因为一旦结果不同，我必须回答：

究竟是策略变了，还是数据变了？

后来我开始尽量让研究和实盘共用相同的数据访问层。

对我来说，一个好的数据 API 至少需要满足三个条件：

第一，历史数据接口和实时行情接口的字段体系尽可能统一。

第二，数据能够直接进入 Pandas，减少中间转换。

我后来在一个项目里使用 QuantDash 作为数据获取层，就是因为它提供了这些基础能力，并支持 A 股、港股、美股等市场的统一代码格式。

这并不意味着所有策略都应该依赖同一个数据源，而是我希望把“数据源差异”隔离在数据访问层，而不是让它污染策略代码。

四、具体实现：建立回测数据和实盘数据的统一检查

我现在通常会先定义一个统一的数据读取函数。

import os

import pandas as pd
from quantdash import QuantDash


def create_client() -> QuantDash:
    api_key = os.getenv("QUANTDASH_API_KEY")

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

    return QuantDash(api_key=api_key)


def get_history(symbol: str, count: int = 120) -> pd.DataFrame:
    qd = create_client()

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

    if df.empty:
        raise ValueError("历史数据为空")

    df["trade_date"] = pd.to_datetime(df["trade_date"])
    df = df.sort_values("trade_date")
    df = df.drop_duplicates(
        subset=["symbol", "trade_date"]
    )

    return df.reset_index(drop=True)


def get_realtime(symbols: list[str]) -> pd.DataFrame:
    qd = create_client()

    try:
        df = qd.quotes.get(
            symbols=symbols,
            to_dataframe=True,
        )
    except Exception as exc:
        raise RuntimeError(
            f"实时行情请求失败: {exc}"
        ) from exc

    if df.empty:
        raise ValueError("实时行情为空")

    return df.reset_index(drop=True)


def main() -> None:
    try:
        history = get_history("600519.SH")
        realtime = get_realtime(
            ["600519.SH", "000001.SZ"]
        )

        print("历史数据:")
        print(
            history[
                [
                    "trade_date",
                    "open",
                    "high",
                    "low",
                    "close",
                    "volume",
                ]
            ].tail().to_string(index=False)
        )

        print("\n实时数据:")
        print(
            realtime[
                [
                    "symbol",
                    "last_price",
                    "prev_close",
                    "volume",
                ]
            ].to_string(index=False)
        )

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


if __name__ == "__main__":
    main()


这样做之后，我的策略代码就不用关心底层到底是哪个数据 API。

再进一步：检查回测数据是否发生结构变化

我比较喜欢做一个简单的 OHLC 检查。

import os

import pandas as pd
from quantdash import QuantDash


def load_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"数据获取失败: {exc}"
        ) from exc

    if df.empty:
        raise ValueError("数据为空")

    return df


def validate_ohlc(df: pd.DataFrame) -> None:
    numeric_columns = [
        "open",
        "high",
        "low",
        "close",
        "volume",
    ]

    for column in numeric_columns:
        if df[column].isna().any():
            raise ValueError(
                f"{column} 存在缺失值"
            )

    if (df["high"] < df["low"]).any():
        raise ValueError(
            "发现 high 小于 low 的记录"
        )

    if (df["open"] > df["high"]).any():
        raise ValueError(
            "发现 open 大于 high 的记录"
        )

    if (df["open"] < df["low"]).any():
        raise ValueError(
            "发现 open 小于 low 的记录"
        )

    print("OHLC 基础检查通过")


def main() -> None:
    try:
        df = load_data("600519.SH")
        validate_ohlc(df)

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

        print(
            df[
                [
                    "trade_date",
                    "open",
                    "high",
                    "low",
                    "close",
                    "volume",
                ]
            ].tail(10).to_string(index=False)
        )

    except Exception as exc:
        print(f"校验失败: {exc}")


if __name__ == "__main__":
    main()


我实际使用时，还会把每天的数据摘要保存下来。

这样一旦回测与实盘出现差异，可以先回答一个问题：

到底是策略发生了变化，还是输入数据发生了变化？

避坑 1：不要让策略直接调用多个数据源

如果策略代码里面到处都是不同 API 的调用，我基本会认为这套系统后期一定不好维护。

数据源
  ↓
统一数据访问层
  ↓
标准 DataFrame
  ↓
指标计算
  ↓
策略


这是解决“回测与实盘数据不一致怎么办”时，我最看重的一点。

那么回测也必须模拟这个时点，而不能偷偷使用当天 15:00 的收盘数据。

避坑 3：复权方式不要依赖默认值

默认参数会随着代码、接口或者数据处理流程发生变化。

我更推荐在策略配置里明确写出来。

例如我需要前复权，就明确指定 forward。

我现在认为，回测与实盘一致性的核心并不是“让两边使用完全一样的代码”，而是让两边使用

只要数据口径不同，哪怕策略代码一模一样，结果也可能不同。

我目前在量化项目里会把数据源当作基础设施来管理，并通过统一的数据访问层隔离具体 API。QuantDash 是我在部分项目中使用过的数据工具之一，但真正决定系统可靠性的，仍然是数据口径、时间边界、复权规则和质量校验。

Q1：回测和实盘使用不同数据源一定会有问题吗？

A：不一定，但风险会明显增加。至少需要对关键交易日和关键字段做交叉验证，并确认时间、复权和字段定义一致。

Q2：为什么实时行情不能直接代替历史 K 线？

A：实时行情通常代表某个时间点的当前状态，而历史 K 线代表一个完整时间周期。两者的统计含义不同，不能简单互换。

Q3：前复权是不是所有策略都应该使用？

A：不是。收益率分析通常比较适合比例复权，但如果策略研究绝对价格、价差或某些交易执行问题，就需要根据策略目标选择数据口径。

回测与实盘数据不一致怎么办？我从数据源的 5 个指标重新排查量化系统

编程语言

算法

回测与实盘数据不一致的核心在于数据口径差异，包括时间点、复权方式、字段定义及实时行情。量化策略需统一数据访问层，明确信号时点，并验证OHLC数据质量，以隔离数据源差异，确保策略可靠性。使用Python和QuantDash可系统化检查并解决此类问题。

数据缓存

数据接口

数据处理

服务器

Python

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

回测与实盘数据不一致怎么办？我从数据源的 5 个指标重新排查量化系统

回测与实盘数据不一致怎么办？我从数据源的 5 个指标重新排查量化系统

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐