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

我在做股票趋势策略时，曾经碰到过一次让我印象很深的问题。

但是重新跑一遍历史回测，最终收益曲线却和之前保存的结果对不上。

于是我把交易信号重新打印了一遍。

结果发现，同一只股票在某些历史日期上的价格和之前的数据不一样。

股票历史价格并不是永远可以直接拿来比较的。

公司分红、送转等事件会影响历史价格序列。如果不同阶段使用了不同的复权口径，那么同一个策略使用相同代码，也可能产生不同的结果。

这次排查让我形成了一个习惯：

1. K 线数据本身只是结果，复权口径决定了历史价格如何表达

QuantDash 的 K 线接口提供了多种复权方式。

这几个选项不是简单的“哪个价格更真实”。

它们对应的是不同的研究需求。

如果我研究收益率，比例复权通常更适合。

如果我要观察绝对价格变化，则差值复权在某些场景下更有意义。

而如果我要研究原始交易价格，则应该使用不复权数据。

2. 批量股票更容易暴露数据口径问题

单股票回测的时候，我可能很容易发现某个价格异常。

但当股票池变成几百只股票以后，问题就变复杂了。

股票 A：2500 根 K 线
股票 B：2498 根 K 线
股票 C：2300 根 K 线
股票 D：数据为空


如果直接把它们扔进因子计算，很可能直到回测结果异常之后才发现。

所以我后来把数据质量检查提前到了行情获取阶段。

我以前确实考虑过自己维护除权因子，然后在本地计算复权价格。

对于个人研究项目来说，这会增加大量维护成本。

方案二：先下载不复权数据，再本地转换

但是我的经验是，只要复权逻辑不是策略本身的研究重点，就没必要让行情数据管道承担额外复杂度。

方案三：让行情接口直接返回指定复权口径

后来我更倾向于在请求历史 K 线时直接指定 adjust。

这样做最大的价值不是少写几行代码，而是：

为了排查问题，我通常会先把原始价格拉出来：

import os
from quantdash import QuantDash


def get_raw_data():
    if not os.getenv("QUANTDASH_API_KEY"):
        raise RuntimeError(
            "请设置 QUANTDASH_API_KEY"
        )

    qd = QuantDash()

    try:
        df = qd.klines.get(
            "600519.SH",
            period="1d",
            count=100,
            adjust="none",
            to_dataframe=True,
        )
    except Exception as e:
        print(f"获取不复权数据失败: {e}")
        return None

    if df.empty:
        print("没有获取到历史 K 线")
        return None

    return df


def main():
    df = get_raw_data()

    if df is None:
        return

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


if __name__ == "__main__":
    main()


我会把这个结果作为排查基准。

import os
from quantdash import QuantDash


def main():
    if not os.getenv("QUANTDASH_API_KEY"):
        raise RuntimeError(
            "请设置 QUANTDASH_API_KEY"
        )

    qd = QuantDash()

    try:
        df = qd.klines.get(
            "600519.SH",
            period="1d",
            count=100,
            adjust="forward",
            to_dataframe=True,
        )
    except Exception as e:
        print(f"获取前复权数据失败: {e}")
        return

    if df.empty:
        print("没有获取到数据")
        return

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


if __name__ == "__main__":
    main()


如果我的策略计算的是历史收益率，我通常会优先考虑这种比例复权数据。

真正做股票池回测时，我不会只测试一只股票。

import os
from quantdash import QuantDash


def main():
    if not os.getenv("QUANTDASH_API_KEY"):
        raise RuntimeError(
            "请设置 QUANTDASH_API_KEY"
        )

    qd = QuantDash()

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

    try:
        dfs = qd.klines.batch(
            symbols,
            period="1d",
            count=250,
            adjust="forward",
            to_dataframe=True,
            show_progress=True,
        )
    except Exception as e:
        print(f"批量行情请求失败: {e}")
        return

    for symbol, df in dfs.items():
        if df.empty:
            print(f"{symbol}: 空数据")
            continue

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

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

        if missing_columns:
            print(
                f"{symbol}: 缺少字段 "
                f"{missing_columns}"
            )
            continue

        print(
            f"{symbol}: "
            f"{len(df)} 条数据，"
            f"开始日期={df['trade_date'].iloc[0]}，"
            f"结束日期={df['trade_date'].iloc[-1]}"
        )


if __name__ == "__main__":
    main()


如果我要重现某一次历史回测，我不会只写 count。

我会明确记录回测开始和结束时间：

import os
import datetime
from quantdash import QuantDash


def timestamp_ms(year, month, day):
    return int(
        datetime.datetime(
            year,
            month,
            day
        ).timestamp() * 1000
    )


def main():
    if not os.getenv("QUANTDASH_API_KEY"):
        raise RuntimeError(
            "请设置 QUANTDASH_API_KEY"
        )

    qd = QuantDash()

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

    start_time = timestamp_ms(2026, 5, 1)
    end_time = timestamp_ms(2026, 6, 1)

    try:
        dfs = qd.klines.batch(
            symbols,
            period="1d",
            start_time=start_time,
            end_time=end_time,
            adjust="forward",
            to_dataframe=True,
        )
    except Exception as e:
        print(f"请求历史数据失败: {e}")
        return

    for symbol, df in dfs.items():
        if df.empty:
            print(f"{symbol}: 指定时间范围没有数据")
            continue

        print(
            f"{symbol}: {len(df)} 个交易日"
        )

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


if __name__ == "__main__":
    main()


这样以后重新跑回测时，数据边界是确定的。

我后来把回测数据管理成下面这种思路：

symbol
period
start_time
end_time
adjust


这五个参数基本可以描述一次历史 K 线数据任务。

我以前保存回测结果时，只记录：

K 线周期
复权方式
数据获取时间


否则几个月之后很可能无法复现。

避坑二：发现回测异常先比数据

我现在排查策略结果，一般按照：

策略参数
↓
交易信号
↓
行情数据
↓
复权方式
↓
成交假设


不要一看到收益率变化，就马上修改策略。

避坑三：不要混用不同复权口径

比如训练数据使用前复权，测试数据使用不复权。

但是最终收益曲线可能完全变样。

我现在越来越觉得，量化研究里最危险的数据问题，不是接口直接报错，而是

接口正常返回了数据，但数据口径和研究预期不一致

回测收益突然变了？我如何排查批量 K 线与复权数据问题

算法

大数据

后端

量化回测结果异常变化常源于历史K线复权口径不一致，而非策略代码问题。本文详解批量获取股票历史数据的正确方法，对比前复权、后复权及不复权差异，提供代码实战演示，并给出数据可复现性检查清单，帮助量化开发者避免回测数据与研究数据错配的常见陷阱。

数据管道

数据管理

Python

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

回测收益突然变了？我如何排查批量 K 线与复权数据问题

回测收益突然变了？我如何排查批量 K 线与复权数据问题

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐