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

我做过一个非常典型的动量策略。

每天收盘后计算过去一段时间的价格变化，然后决定第二天的股票池。

但真正进入模拟实盘之后，我发现一个很奇怪的现象：

策略每天晚上算出来的股票名单，第二天重新回看时，有几只股票的历史信号竟然消失了。

最开始我以为是代码缓存的问题。

后来我把当天策略运行时保存的数据和第二天重新下载的数据逐行比较，发现价格发生了变化。

股票发生分红、送转等公司行为以后，历史价格的复权数据可能会发生变化。

这件事让我吃了一个很大的教训：

回测与实盘一致性，本质上首先是数据口径一致性。

1. 时间边界是最容易犯错的地方

“使用截至当天收盘的数据生成第二天信号。”

那么数据截止时间必须非常明确。

如果我误把当天盘中的数据算进去了，就产生了未来函数。

如果我把下一交易日的数据混入历史窗口，同样会产生未来信息。

我现在处理时间窗口时，会尽量显式指定起止时间。

import datetime
import os
from quantdash import QuantDash


def timestamp_ms(year, month, day, hour=0, minute=0):
    value = datetime.datetime(
        year,
        month,
        day,
        hour,
        minute,
    )
    return int(value.timestamp() * 1000)


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:
        start = timestamp_ms(2026, 5, 1)
        end = timestamp_ms(2026, 5, 31, 23, 59)

        df = qd.klines.get(
            "600519.SH",
            period="1d",
            start_time=start,
            end_time=end,
            to_dataframe=True,
        )

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

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


if __name__ == "__main__":
    main()


我现在特别喜欢这种写法，因为它比“取最近 N 条数据”更容易审计。

尤其在回测系统里，时间窗口应该是一个明确的实验参数。

假设研究阶段使用前复权数据计算均线，实盘阶段却使用了不复权数据。

那么两个系统计算出来的信号出现差异几乎是必然的。

我现在做策略研究时，会把复权方式写进配置。

import os
from quantdash import QuantDash


def load_price_data(qd, symbol, adjust):
    df = qd.klines.get(
        symbol,
        period="1d",
        count=100,
        adjust=adjust,
        to_dataframe=True,
    )

    if df.empty:
        raise ValueError(
            f"{symbol} 未获取到 {adjust} 数据"
        )

    return df


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:
        forward = load_price_data(
            qd,
            "600519.SH",
            "forward",
        )

        none_adjust = load_price_data(
            qd,
            "600519.SH",
            "none",
        )

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

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

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


if __name__ == "__main__":
    main()


QuantDash 支持比例复权和差值复权。

我个人在计算收益率和趋势类指标时，会特别关注比例复权；如果策略研究的是绝对价差，则会另外评估差值复权是否更合适。

关键不是“哪一种复权永远正确”，而是：

研究、回测和实盘必须使用同一套定义。

后来我不再满足于只拿最终复权价格。

因为如果复权后的价格发生异常变化，我希望能够进一步追踪原因。

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:
        df = qd.klines.ex_factors(
            ["600519.SH"],
            to_dataframe=True,
        )

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

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

    except Exception as exc:
        print(f"获取除权因子失败: {exc}")


if __name__ == "__main__":
    main()


复权数据出了问题时，先追查因子，再追查最终价格，通常比直接肉眼看 K 线快很多。

我以前解决回测和实盘不一致的方法比较原始。

发现信号不同，就手工找两边的数据。

这种方式最大的问题是不可重复。

今天查明白了，过几天换一只股票，又得重新查。

确认交易日期、OHLCV 和标的代码。

确认研究和实盘使用的是相同复权方式。

确认策略最终拿到的 DataFrame 是否完全一致。

确认相同输入是否产生相同信号。

这样一来，如果信号不同，我就能快速判断：

到底是数据问题，还是代码问题。

四、具体实现：做一套回测数据一致性检查

我现在会先比较两份 DataFrame 的关键字段。

import pandas as pd


def compare_dataframes(left, right):
    columns = [
        "trade_date",
        "open",
        "high",
        "low",
        "close",
        "volume",
    ]

    for column in columns:
        if column not in left.columns:
            raise ValueError(
                f"左侧数据缺少字段: {column}"
            )

        if column not in right.columns:
            raise ValueError(
                f"右侧数据缺少字段: {column}"
            )

    left_view = left[columns].copy()
    right_view = right[columns].copy()

    left_view = left_view.reset_index(drop=True)
    right_view = right_view.reset_index(drop=True)

    if len(left_view) != len(right_view):
        print(
            f"行数不同: "
            f"{len(left_view)} vs {len(right_view)}"
        )

    count = min(
        len(left_view),
        len(right_view),
    )

    differences = []

    for index in range(count):
        for column in columns:
            left_value = left_view.iloc[index][column]
            right_value = right_view.iloc[index][column]

            if pd.isna(left_value) and pd.isna(right_value):
                continue

            if left_value != right_value:
                differences.append(
                    {
                        "row": index,
                        "column": column,
                        "left": left_value,
                        "right": right_value,
                    }
                )

    return differences


def main():
    first = pd.DataFrame(
        {
            "trade_date": [
                "2026-06-16",
                "2026-06-17",
            ],
            "open": [1267.01, 1258.00],
            "high": [1267.88, 1259.77],
            "low": [1255.00, 1238.56],
            "close": [1255.67, 1240.00],
            "volume": [34970, 44803],
        }
    )

    second = first.copy()
    second.loc[1, "close"] = 1241.00

    differences = compare_dataframes(
        first,
        second,
    )

    for item in differences:
        print(item)


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:
        quotes = qd.quotes.get(
            symbols=[
                "600519.SH",
                "000001.SZ",
            ],
            to_dataframe=True,
        )

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

        columns = [
            "symbol",
            "last_price",
            "prev_close",
            "volume",
        ]

        available = [
            column
            for column in columns
            if column in quotes.columns
        ]

        print(
            quotes[available].to_string(
                index=False
            )
        )

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


if __name__ == "__main__":
    main()


我不会把实时行情直接当作策略输入。

在进入策略前，我通常还会做一次：

避坑一：不要默认“重新下载的数据一定和昨天一样”

历史数据可能因为修正、补数或者公司行为而发生变化。

如果策略需要可复现回测，我建议把每次研究使用的数据版本或者快照信息记录下来。

避坑二：明确区分“观察价格”和“计算价格”

我经常看到研究代码里直接把一个 close 字段用于所有计算。

实际上，不复权价格、前复权价格和后复权价格承担的用途可能完全不同。

避坑三：分钟线和日线不要简单互相替代

分钟数据和日线数据虽然都属于 K 线，但聚合规则、时间边界和交易时段都会影响结果。

如果策略是日频策略，就应该明确日线的来源和口径。

如果策略是盘中策略，就不要拿日线字段去推导盘中状态。

我后来发现，所谓“回测和实盘不一致”，其实是一个非常大的问题。

真正稳定的系统必须同时控制时间边界、数据版本、复权方式、行情字段和异常处理。

我在自己的工作流里使用统一的数据接口获取历史 K 线、分钟数据和实时行情，再通过自己的校验层确保策略看到的数据符合预期。QuantDash 对我来说更像是数据获取环节的一块基础设施，而不是质量控制本身。

如果要给量化开发者一个建议，我会说：

不要等策略收益出现问题以后才检查数据。应该在策略运行之前，就让数据自己证明它是可信的。

Q1：回测与实盘数据不一致怎么办？

A：我建议按照时间边界、交易日、复权方式、字段值、数据版本和策略输入六层逐级比较。不要直接从策略结果倒推问题。

Q2：前复权数据适合所有策略吗？

A：不一定。比例复权适合很多收益率和趋势类计算，但具体策略仍然应该根据研究目标选择复权方式。关键是研究、回测和实盘保持一致。

A：首先明确策略信号产生时间，其次严格限定数据截止时间，最后不要使用尚未在信号产生时可获得的数据。

回测与实盘数据不一致怎么办？我从时间、复权和数据口径三个层面排查

编程语言

算法

职业发展

数据一致性

数据接口

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

回测与实盘数据不一致怎么办？我从时间、复权和数据口径三个层面排查

回测与实盘数据不一致怎么办？我从时间、复权和数据口径三个层面排查

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐