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

如果让我选择量化数据里最容易被忽略的问题，我会把“复权口径”排在很靠前的位置。

因为它有一个非常危险的特点：

甚至你随机打印几行数据，都很难发现问题。

我曾经维护过一个中短周期的趋势策略。

回测阶段使用的是经过复权处理的历史行情，而实盘监控程序拿到的是另一套价格口径。

结果某一天，策略在研究环境里认为某只股票已经突破阈值。

两个系统比较的根本不是同一种价格。

1. “没有缺数据”不等于“数据可以直接比较”

这是我后来对数据完整性的理解发生变化的地方。

以前我理解的数据质量主要是：

有没有数据
有没有空值
有没有重复


对于股票历史价格而言，复权方式就是一个典型的数据口径。

假设某股票发生分红、送股或者其他公司行为。

如果我直接使用原始价格序列，那么前后价格可能发生明显跳变。

这个跳变未必代表市场真实收益变化。

过去 N 日收益率
移动平均
动量
突破价格
波动率


那么公司行为造成的价格变化可能直接进入指标。

因此，历史研究通常需要根据具体策略选择适合的价格口径。

3. 我后来把复权分成两个维度理解

其中比例复权使用乘法因子还原价格，更适合收益率计算；差值复权使用加减方式调整价格，更适合观察绝对价差。

我后来发现，真正容易出问题的不是“选错一种复权”。

研究阶段选了一种，实盘阶段又用了另一种。

第一种方案：全部使用不复权数据

但是当我要计算长期收益率、移动平均或者趋势指标时，公司行为带来的价格变化会明显影响指标。

所以这种方式并不适合我所有的研究策略。

它比较适合很多历史收益率和技术指标计算。

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


并不是所有研究任务都应该使用同一种复权方式。

例如我如果要分析绝对价格变化，那么比例复权和差值复权的意义并不完全一样。

所以我最后没有建立“默认永远前复权”的规则，而是建立：

研究目标
   ↓
确定价格口径
   ↓
保存口径信息
   ↓
计算指标


第三种方案：原始数据和研究数据分层保存

我的数据层不会只保存一份价格。

原始行情
    ↓
复权行情
    ↓
因子计算


这样未来如果策略需要重新研究不同复权口径，不需要重新猜测当时使用的是什么数据。

下面的程序会同时获取不复权和前复权数据，然后检查日期是否一致，并比较收盘价序列。

这里我特别强调一个工程原则：

比较不同复权数据时，不能把价格相等作为质量标准。

import os
import datetime
from typing import Tuple

import pandas as pd
from quantdash import QuantDash


def date_to_timestamp(date_text: str) -> int:
    """将日期转换为毫秒时间戳。"""
    dt = datetime.datetime.strptime(
        date_text,
        "%Y-%m-%d",
    )
    return int(dt.timestamp() * 1000)


def validate_dataframe(
    name: str,
    df: pd.DataFrame,
) -> Tuple[bool, str]:
    """检查基础 DataFrame 数据质量。"""
    if df.empty:
        return False, f"{name} 返回为空"

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

    missing = [
        column
        for column in required
        if column not in df.columns
    ]

    if missing:
        return False, (
            f"{name} 缺少字段: "
            + ", ".join(missing)
        )

    if df["trade_date"].duplicated().any():
        return False, f"{name} 存在重复交易日期"

    if df[required[1:]].isna().any().any():
        return False, f"{name} 存在核心字段空值"

    return True, f"{name} 基础检查通过"


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

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

    qd = QuantDash(api_key=api_key)

    start_time = date_to_timestamp("2026-05-01")
    end_time = date_to_timestamp("2026-06-01")

    symbol = "600519.SH"

    try:
        raw_df = qd.klines.get(
            symbol,
            period="1d",
            start_time=start_time,
            end_time=end_time,
            adjust="none",
            to_dataframe=True,
        )

        forward_df = qd.klines.get(
            symbol,
            period="1d",
            start_time=start_time,
            end_time=end_time,
            adjust="forward",
            to_dataframe=True,
        )

        raw_ok, raw_message = validate_dataframe(
            "不复权数据",
            raw_df,
        )

        forward_ok, forward_message = validate_dataframe(
            "前复权数据",
            forward_df,
        )

        print(raw_message)
        print(forward_message)

        if not raw_ok or not forward_ok:
            return

        raw_dates = set(
            raw_df["trade_date"].astype(str)
        )

        forward_dates = set(
            forward_df["trade_date"].astype(str)
        )

        only_raw = sorted(
            raw_dates - forward_dates
        )

        only_forward = sorted(
            forward_dates - raw_dates
        )

        print("\n日期一致性检查")

        if only_raw:
            print(
                "仅不复权数据存在的日期:",
                only_raw,
            )
        else:
            print("不复权数据没有额外日期")

        if only_forward:
            print(
                "仅前复权数据存在的日期:",
                only_forward,
            )
        else:
            print("前复权数据没有额外日期")

        print("\n数据规模")
        print("不复权:", len(raw_df))
        print("前复权:", len(forward_df))

        comparison = pd.merge(
            raw_df[
                ["trade_date", "close"]
            ].rename(
                columns={"close": "close_raw"}
            ),
            forward_df[
                ["trade_date", "close"]
            ].rename(
                columns={"close": "close_forward"}
            ),
            on="trade_date",
            how="inner",
        )

        comparison["price_difference"] = (
            comparison["close_forward"]
            - comparison["close_raw"]
        )

        print("\n价格口径对比:")
        print(
            comparison.tail(10).to_string(
                index=False
            )
        )

    except Exception as exc:
        print(f"请求或处理失败: {exc}")


if __name__ == "__main__":
    main()


这个例子里，我没有把不复权和前复权数据简单地说成“谁正确”。

这是我在实际工程里很看重的一个原则。

复权方式不是数据质量问题本身，口径不明确才是。

如何避免回测与实盘数据不一致？

我后来专门给研究环境增加了一项元数据：

symbol
period
adjust
start_time
end_time


也就是说，当我生成一个研究数据集时，我不会只保存：

而会把数据的来源和口径一起记录下来。

因为几个月以后，如果回测结果出现变化，我需要能够回答：

这件事看起来和策略没有关系，但实际上对研究复现非常重要。

避坑 1：研究收益率和绝对价格分析不要混为一谈

如果我的目标是计算历史收益率，比例复权通常更方便。

量化研究中如何处理数据缺口：复权错误为什么比少一天数据更危险

编程语言

人工智能

数据库

量化回测与实盘信号对不上的核心原因，往往是数据复权口径不一致而非数据缺失。本文剖析前复权与不复权数据差异，展示如何通过原始数据与复权数据分层、保存口径元数据、建立统一数据检查流程，确保因子计算可复现，避免策略结果悄悄偏离。

数据完整性

数据治理

数据接口

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

量化研究中如何处理数据缺口：复权错误为什么比少一天数据更危险

量化研究中如何处理数据缺口：复权错误为什么比少一天数据更危险

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐