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

每天收盘后，策略从股票池中计算过去一段时间的收益率和波动率，然后生成第二天的交易列表。

但策略上线之后，我发现一个非常反常的现象：

同一个股票、同一个时间区间，回测计算出来的收益率和实盘系统记录的收益率对不上。

后来我直接把某只股票的历史行情逐行导出来，才发现一个关键问题：

回测使用的是复权价格，而交易系统记录的是实际成交价格。

再往下查，又发现历史上刚好存在除权事件。

1. 我后来发现，量化系统里其实存在三种“价格”

在这个问题上，我现在会刻意区分三类价格。

它描述某个交易日真实发生的开盘、最高、最低和收盘价格。

它通过公司行为相关的因子，对历史价格序列进行调整，让价格序列更加适合收益率和技术指标研究。

这是策略真正下单时使用的价格。

这三个价格可以有关联，但不能简单当成同一个字段。

2. 为什么回测和实盘特别容易产生差异

我的经验是，问题经常发生在数据链路的边界上。

历史行情 → 前复权 → 计算指标 → 产生信号

实时行情 → 原始价格 → 产生订单 → 成交

如果我直接拿两个结果比较，就已经存在语义差异。

尤其是长期策略，一旦时间区间跨越除权事件，差异会更加明显。

3. 我排查问题时会先检查数据，而不是先改策略

出现回测与实盘不一致时，我一般按照下面的顺序排查：

现在想想，花几个小时重新审查策略逻辑，结果最后发现是行情口径不同，是非常典型的无效排查。

方案一：所有数据全部使用原始价格

结果是短周期策略问题不明显，但长期趋势策略马上出现了问题。

因为公司行为会让原始价格序列出现不连续变化。

方案二：所有数据全部使用前复权

这个方案解决了历史收益率计算问题，但又产生了另一个问题。

如果交易执行系统也直接使用复权价格，就会让成交模拟失去现实意义。

方案三：研究价格和交易价格分层

这套架构虽然多了几个字段，但排查问题的时候非常省时间。

我现在通常会先把行情访问封装起来。

import os
import sys
from quantdash import QuantDash


def create_client():
    """从环境变量读取 API Key 并创建客户端。"""
    if not os.getenv("QUANTDASH_API_KEY"):
        raise RuntimeError("请先设置 QUANTDASH_API_KEY 环境变量")

    return QuantDash()


def load_history(qd, symbol, adjust):
    """读取指定复权方式的日K线。"""
    df = qd.klines.get(
        symbol,
        period="1d",
        count=100,
        adjust=adjust,
        to_dataframe=True
    )

    if df.empty:
        raise RuntimeError(
            f"{symbol} 未返回历史行情，请检查标的代码或数据权限"
        )

    return df


def main():
    try:
        qd = create_client()

        raw = load_history(qd, "600519.SH", "none")
        forward = load_history(qd, "600519.SH", "forward")

        print("原始行情最后5条：")
        print(
            raw[
                ["trade_date", "open", "close", "volume"]
            ].tail(5).to_string(index=False)
        )

        print("\n前复权行情最后5条：")
        print(
            forward[
                ["trade_date", "open", "close", "volume"]
            ].tail(5).to_string(index=False)
        )

    except Exception as exc:
        print(f"程序执行失败: {exc}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()


我会把这类函数放在数据访问层，而不是散落在策略代码里面。

这样以后切换数据源或者调整复权方式，影响范围会小很多。

如果我已经知道某次回测异常发生在哪个月，就没必要拉全部历史数据。

import os
import sys
import datetime
from quantdash import QuantDash


def to_milliseconds(dt):
    """将本地时间转换为毫秒时间戳。"""
    return int(dt.timestamp() * 1000)


def main():
    if not os.getenv("QUANTDASH_API_KEY"):
        print("未检测到 QUANTDASH_API_KEY", file=sys.stderr)
        sys.exit(1)

    try:
        qd = QuantDash()

        start = to_milliseconds(
            datetime.datetime(2026, 5, 1)
        )
        end = to_milliseconds(
            datetime.datetime(2026, 5, 31)
        )

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

        if df.empty:
            print("指定时间区间没有数据")
            return

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

    except Exception as exc:
        print(f"读取历史区间失败: {exc}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()


因为数据问题的定位，本质上是缩小搜索范围。

3. 同时读取实时行情和历史行情

如果问题出现在“今天”和“历史”之间，我还会把实时行情一起拿出来。

import os
import sys
from quantdash import QuantDash


def main():
    if not os.getenv("QUANTDASH_API_KEY"):
        print("未检测到 QUANTDASH_API_KEY", file=sys.stderr)
        sys.exit(1)

    try:
        qd = QuantDash()

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

        quotes = qd.quotes.get(
            symbols=symbols,
            to_dataframe=True
        )

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

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

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

    except Exception as exc:
        print(f"实时行情读取失败: {exc}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()


这一步的意义并不是拿实时价格做历史回测。

实盘系统看到的价格，到底和历史系统里的哪一种价格对应。

我现在不喜欢只使用一个 close 字段。

在内部数据模型中，我会明确区分：

名字清楚以后，很多错误根本不会发生。

避坑二：把除权事件加入回测测试集

我以前测试行情系统，喜欢随机抽日期。

现在我会主动加入公司行为附近的日期。

因为随机测试不一定能覆盖最容易出错的数据。

避坑三：回测与实盘必须使用相同的标的编码规则

当我开始做多市场策略以后，这一点变得更加明显。

如果一个系统用 600519，另一个系统用 600519.SH，看似只是字符串不同，实际可能导致数据映射失败。

我现在处理“回测与实盘数据不一致怎么办”这类问题时，已经不会第一时间怀疑策略。

回测与实盘数据不一致怎么办：从除权因子排查量化策略收益偏差

算法

大数据

职业发展

回测与实盘收益偏差常源于数据口径不一致，而非策略逻辑错误。本文以实际案例剖析原始价格、复权价格与交易执行价的区别，提供按序排查方法，并演示用Python统一行情读取、区间复现及实时校验，建立研究层与执行层分离的可靠数据架构。

数据架构

数据审计

数据处理

Python

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

回测与实盘数据不一致怎么办：从除权因子排查量化策略收益偏差

回测与实盘数据不一致怎么办：从除权因子排查量化策略收益偏差

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐