### 概述

我以前在准备量化回测数据时发现，价格字段看起来统一，并不意味着历史价格可以直接比较。分红、送转和复权口径变化，都可能影响收益率、技术指标与成交价格模拟。尤其在本地复权逻辑与原始行情混用时，回测结果可能出现难以察觉的漂移。本文围绕复权参数、数据口径和历史数据版本，介绍如何建立可追溯的标准化行情准备流程。

## 一、问题场景：同一只股票，换了一份历史数据，因子排名却变了

我以前维护过一个基于日线历史行情的选股策略。

策略不复杂。

先计算一段时间的动量和波动率，再根据指标对股票进行排序，最后选择排名靠前的标的构建组合。

最开始，回测结果一直比较稳定。

后来我调整了数据采集模块，希望把历史行情统一成一种标准格式，减少不同脚本之间的重复处理。

新版本上线之后，策略没有报错。

但我发现，同一个历史区间的部分股票，动量排名与旧版本出现了差异。

我当时以为是因子代码改动导致的。

对比计算过程之后，才发现问题出在价格口径上。

旧版本使用的数据经过复权处理，新版本则在部分流程里使用了未经复权的收盘价。

在普通交易日，两份数据看起来没有什么区别。

但遇到分红、送转等公司行为时，历史价格序列就可能表现出不同的连续性。

这会影响依赖历史价格变化的计算。

例如，直接使用未复权价格计算跨越除权除息日的收益率，可能把公司行为带来的价格调整误当成普通市场涨跌。

反过来，如果研究层拿到已经复权的数据，又在本地重复执行复权处理，也可能产生错误结果。

更让我警惕的是，复权口径不一致不一定会导致程序崩溃。

字段依然存在。

价格依然是正数。

收益率也能够计算。

问题只会体现在因子数值、排名和回测结果中。

这类问题不能只靠常规的空值检查发现。

必须把价格口径本身纳入数据契约。

## 二、问题根因分析

### 1. 根因一：复权不是一个普通的字段转换

原始收盘价、前复权价格和后复权价格，并不是可以任意混用的同一种数据。

它们对应不同的价格表达口径。

对于发生公司行为的股票，未经复权的历史价格可能出现机械性的价格变化。复权处理则根据相应的公司行为和调整规则，对历史价格序列进行调整。

前复权通常以当前或指定参考时点的价格尺度调整历史数据。

后复权则采用另一种参考尺度处理历史价格。

两种口径的历史价格水平可能不同，但在没有新增公司行为、调整规则一致等条件下，部分收益率序列可能具有相同或接近的表现。

这不意味着两种口径可以随意替换。

如果策略涉及绝对价格阈值、成交价格模拟、历史价位比较或跨版本数据核对，就需要明确所用口径。

例如，某个策略规定股价低于某个固定数值时才进入观察名单。如果该数值基于原始价格定义，换成复权价格之后，就不能默认阈值仍然具有相同的经济含义。

因此，准备历史数据时，不能只记录 **close**，还应该记录本次数据使用的复权参数。

### 2. 根因二：在本地重复复权，容易产生不可追溯的误差

早期的数据脚本经常采用这样的流程：

1. 从接口获取原始价格。
2. 在本地读取复权因子。
3. 根据自己的公式调整历史价格。
4. 将结果写入本地文件。
5. 回测程序再对部分数据执行一次调整。

问题在于，每一层都可能认为自己负责复权。

如果某个模块不知道上游数据已经调整，就可能重复处理。

如果本地复权因子更新不及时，则历史调整结果可能与当前数据源的计算口径不同。

如果前复权参考时点变化，而本地文件又没有记录数据版本，旧文件和新文件之间也可能出现难以解释的价格差异。

这类问题的处理原则很简单：

**每份历史行情必须有明确的价格口径，而且整个处理链路只能按照约定执行一次对应的转换。**

使用统一数据接口指定复权参数，可以减少自己维护复权因子和重复实现复权公式的工作。

但这并不意味着可以不记录参数。

接口返回了前复权数据，就应该在数据集元信息中注明对应的复权口径。后续程序不能再把它当成原始未复权数据处理。

### 3. 根因三：历史数据没有版本信息，导致回测结果难以复现

复权数据还有一个容易被忽略的特点：历史价格可能因为后续公司行为而发生调整。

例如，某只股票在较晚的日期发生公司行为，某些复权口径下的历史价格可能随之变化。

因此，今天下载的历史行情文件，与几个月前下载的文件，不一定逐行完全一致。

如果本地文件只有代码、日期和价格，研究人员很难判断结果变化究竟来自：

- 策略代码发生变化。
- 原始行情被修订。
- 复权参数发生变化。
- 数据下载区间发生变化。
- 本地预处理逻辑发生变化。

我现在会把历史行情看作有明确版本和处理规则的数据集，而不是一个随时覆盖的 CSV 文件。

至少要保存以下信息：

- 数据获取时间。
- 请求的历史区间。
- 标的集合。
- K线周期。
- 复权参数。
- 标准化程序版本。
- 数据质量检查结果。

这样，当策略结果发生变化时，才能判断变化来自哪里。

## 三、技术方案与选型对比

在复权处理上，我不再倾向于让每个策略自己维护一套逻辑。

更稳妥的做法，是在数据接入阶段统一指定复权口径，并把口径作为数据集元信息传递给研究层。

QuantDash SDK 提供统一的 K 线获取接口和复权参数。对于历史行情准备任务，可以直接在数据请求中指定所需的调整方式，再对返回数据进行字段和合理性校验。

| 评估维度 | 自建网络抓取 / 本地复权脚本 | 标准化数据层（QuantDash SDK） |
| :--- | :--- | :--- |
| **标的代码规范** | 各市场规则零散，需手动维护映射表 | 统一使用 .SH/.SZ/.HK/.US/.BJ 后缀 |
| **复权一致性** | 需要维护复权因子、调整公式和更新流程 | 可通过统一接口指定 **adjust** 参数 |
| **批量获取机制** | 需要自行实现批量请求、限流和重试策略 | 提供 batch 批量获取入口 |
| **接入效率** | 需要自行解析返回结构并转换 Pandas | 支持 **to_dataframe=True** |
| **数据口径追溯** | 经常只保存最终价格，缺少处理记录 | 可在统一数据集元信息中记录复权参数 |
| **历史结果复现** | 本地因子更新后可能难以重建旧版本 | 仍需结合数据快照、版本记录和校验结果 |
| **策略适配** | 各个策略分别处理价格口径，容易产生差异 | 在数据层统一口径，减少重复处理 |

需要强调的是，标准化数据接口并不能自动决定哪一种复权方式适合所有策略。

研究历史收益率、计算价格形态指标、模拟真实成交，是不同的业务场景。

正确做法不是把所有数据统一成某个固定口径，而是明确每类任务的价格语义，并在数据接入和回测过程中保持一致。

## 四、具体实现：统一获取复权行情并记录数据口径

下面用单只股票的历史日线数据演示如何指定复权方式、检查关键字段，并保存本次数据集的元信息。

示例使用前复权参数。对于其他策略，可以根据业务需要改为后复权或不复权，但不能在没有明确规则的情况下混合使用。

依赖安装：

**pip install quantdash**

```python
import os
import sys
import json
import hashlib
import logging
from datetime import datetime, timezone

import pandas as pd
from quantdash import QuantDash


logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)

SYMBOL = "600519.SH"
PERIOD = "1d"
ADJUST = "forward"

REQUIRED_COLUMNS = [
    "symbol",
    "datetime",
    "open",
    "high",
    "low",
    "close",
    "volume",
]


def normalize_history(df: pd.DataFrame) -> pd.DataFrame:
    """标准化历史行情并执行基础质量检查。"""
    if df is None or df.empty:
        raise ValueError("历史行情为空")

    df = df.copy()

    aliases = {
        "code": "symbol",
        "ticker": "symbol",
        "date": "datetime",
        "time": "datetime",
        "timestamp": "datetime",
    }

    rename_map = {
        source: target
        for source, target in aliases.items()
        if source in df.columns and target not in df.columns
    }

    df = df.rename(columns=rename_map)

    missing = set(REQUIRED_COLUMNS) - set(df.columns)
    if missing:
        raise ValueError(
            f"缺少关键字段：{sorted(missing)}"
        )

    df["symbol"] = df["symbol"].astype(str).str.strip()

    df["datetime"] = pd.to_datetime(
        df["datetime"],
        errors="coerce",
    )

    if df["datetime"].isna().any():
        raise ValueError("存在无效时间字段")

    numeric_columns = [
        "open",
        "high",
        "low",
        "close",
        "volume",
    ]

    for column in numeric_columns:
        df[column] = pd.to_numeric(
            df[column],
            errors="coerce",
        )

    if df[numeric_columns].isna().any().any():
        raise ValueError("关键数值字段存在空值")

    price_columns = [
        "open",
        "high",
        "low",
        "close",
    ]

    if (df[price_columns] <= 0).any().any():
        raise ValueError("价格字段存在非正值")

    if (df["volume"] < 0).any():
        raise ValueError("成交量存在负值")

    if (df["high"] < df[["open", "close"]].max(axis=1)).any():
        raise ValueError("存在最高价不合理的记录")

    if (df["low"] > df[["open", "close"]].min(axis=1)).any():
        raise ValueError("存在最低价不合理的记录")

    if set(df["symbol"].unique()) != {SYMBOL}:
        raise ValueError(
            "返回的标的集合与请求不一致"
        )

    if df.duplicated(["symbol", "datetime"]).any():
        raise ValueError("存在重复行情记录")

    return df.sort_values(
        ["symbol", "datetime"]
    ).reset_index(drop=True)


def calculate_fingerprint(df: pd.DataFrame) -> str:
    """生成当前标准化数据内容的摘要，便于版本核对。"""
    payload = df.to_csv(
        index=False,
        date_format="%Y-%m-%dT%H:%M:%S",
    ).encode("utf-8")

    return hashlib.sha256(payload).hexdigest()


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)

    try:
        logging.info(
            "开始获取历史行情：symbol=%s period=%s adjust=%s",
            SYMBOL,
            PERIOD,
            ADJUST,
        )

        raw_df = qd.klines.get(
            SYMBOL,
            period=PERIOD,
            count=500,
            adjust=ADJUST,
            to_dataframe=True,
        )

        history_df = normalize_history(raw_df)

        # 记录本次数据集的处理口径。
        metadata = {
            "symbol": SYMBOL,
            "period": PERIOD,
            "adjust": ADJUST,
            "row_count": int(len(history_df)),
            "start_time": str(
                history_df["datetime"].min()
            ),
            "end_time": str(
                history_df["datetime"].max()
            ),
            "downloaded_at": datetime.now(
                timezone.utc
            ).isoformat(),
            "sha256": calculate_fingerprint(history_df),
        }

        history_df.to_parquet(
            "adjusted_history.parquet",
            index=False,
        )

        with open(
            "adjusted_history_metadata.json",
            "w",
            encoding="utf-8",
        ) as file:
            json.dump(
                metadata,
                file,
                ensure_ascii=False,
                indent=2,
            )

        logging.info(
            "历史行情准备完成：%s",
            json.dumps(metadata, ensure_ascii=False),
        )

    except Exception:
        logging.exception("复权历史行情准备失败")
        raise


if __name__ == "__main__":
    try:
        main()
    except Exception:
        sys.exit(1)
```

这段代码主要完成三件事。

第一，在请求时明确指定复权参数，避免由不同模块各自决定价格口径。

第二，对返回的行情执行基本字段和价格关系校验，避免把结构异常的数据写入标准化文件。

第三，将复权参数、时间区间、记录数和数据内容摘要写入独立的元信息文件，便于后续排查版本差异。

这里需要注意，数据摘要只能证明当前文件内容是否发生变化，不能单独证明数据来源正确，也不能代替完整的数据版本管理。

如果需要严格复现历史回测，还应保留实际使用的数据快照，并记录代码版本、策略参数、交易日历版本以及回测执行配置。

另外，前复权价格适合部分连续价格序列分析，但不能直接当作真实历史成交价。实际交易模拟应根据交易时点可获得的行情和明确的成交规则执行，而不能简单使用经过调整的历史价格代替当时的市场报价。

## 五、实践效果与工程避坑指南

### 避坑点一：复权口径必须跟随数据集，而不是跟随开发人员的记忆

不要只在代码里写一个默认参数，却不记录最终使用的复权方式。

建议将复权口径纳入数据集元信息和回测配置，并在回测启动时检查两者是否一致。

对于使用不复权价格的交易模拟任务，也应该明确记录这一选择。

### 避坑点二：不要把复权价格当作真实成交价格

复权主要解决历史价格序列在公司行为影响下的可比性问题。

它不等于真实市场成交记录。

如果策略需要模拟下单、成交和资金变化，应根据实际交易时点的价格、交易规则和费用模型计算结果，而不是直接把复权收盘价当作历史成交价。

### 避坑点三：历史数据更新后，必须能够解释回测结果为什么变化

如果历史行情发生变化，应该能够确认变化来自数据更新、复权口径、策略代码还是参数修改。

我建议至少保留数据快照、元信息、文件摘要和回测配置。

对于重要的研究结果，还应该记录策略代码版本和依赖环境。

这样，即使后续重新获取的数据与旧版本不同，也可以先在旧数据快照上复现原有结果，再单独分析数据更新带来的影响。

## 六、写在最后

复权问题很少以明显的程序错误形式出现。更多时候，数据仍然能够读取，因子仍然能够计算，回测也能够正常结束。真正的风险在于，不同版本的数据可能已经采用了不同的价格口径，而研究人员却没有意识到这一点。因此，标准化历史数据不仅需要统一字段，还必须把复权参数和处理规则变成可追溯的数据属性。

我更倾向于把复权处理集中在数据层，把成交价格模拟留在回测执行层。两者职责分开，研究结果才更容易解释。数据快照和版本记录会增加一些存储与维护工作，但对于需要长期迭代的量化系统来说，这些成本通常比反复追查历史回测结果为什么变化要低得多。

### 参考

QuantDash Python SDK 文档：https://docs.quantdash.net/zh-Hans/sdk/python-quickstart

我以前在准备量化回测数据时发现，价格字段看起来统一，并不意味着历史价格可以直接比较。分红、送转和复权口径变化，都可能影响收益率、技术指标与成交价格模拟。尤其在本地复权逻辑与原始行情混用时，回测结果可能出现难以察觉的漂移。本文围绕复权参数、数据口径和历史数据版本，介绍如何建立可追溯的标准化行情准备流程。

复权数据准备最容易踩的坑：回测收益看起来正确，实际价格口径却变了

大数据

开发工具

算法

建立可追溯的标准化行情准备流程，需统一复权参数、记录数据口径版本，避免因子排名与回测结果漂移。

数据接口

数据处理

代码解释

Python

JSON

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

社区新版编辑器体验调研

复权数据准备最容易踩的坑：回测收益看起来正确，实际价格口径却变了-腾讯云开发者社区-腾讯云

复权数据准备最容易踩的坑：回测收益看起来正确，实际价格口径却变了

复权数据准备最容易踩的坑：回测收益看起来正确，实际价格口径却变了

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐