我曾经花了一周时间排查一个很奇怪的问题：

回测赚钱，实盘却完全没有那么赚钱。

最后甚至重新检查了信号代码。

后来我把回测数据和实盘每天收到的数据放到一起比较，发现一个很容易被忽视的问题：

这件事让我后来形成了一个非常明确的工程原则：

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

当时我的策略是一套中低频趋势策略。

但是上线后，我发现某些股票的信号时间点与研究环境不完全一致。

回测：
close = A

实盘：
close = B


两者差异并不是因为实时行情延迟。

进一步追踪之后，我发现历史行情的复权方式不同。

数据源一致，并不代表数据口径一致。

1. 回测策略真正依赖的是“价格序列”

这个 close 是什么？

不复权？
前复权？
后复权？
差值复权？


如果这个问题回答不清楚，后面的指标计算其实没有稳定的数据基础。

2. 为什么趋势策略容易受到除权影响

然后因为权益事件发生价格调整。

如果直接使用新的名义价格，序列可能出现：

对于人眼来说，我们知道这里可能发生了除权。

看到的就是一次巨大的负收益。

那么这个异常点就会进入因子计算。

我后来使用前复权的主要原因并不是为了得到“真实价格”。

我知道它不是历史真实成交价格。

但它可以把历史价格调整到更连续的价格体系。

但问题是，我不仅需要历史行情，还需要可靠的除权因子。

而且一旦数据更新，我还需要重新计算历史序列。

对于一个需要持续运行的数据系统来说，维护成本不低。

但是我的趋势策略很快暴露出问题。

因为一些权益事件会造成价格序列不连续。

方案三：在数据获取阶段明确复权方式

数据获取层：
明确 adjust

↓
Pandas DataFrame

↓

因子计算层

↓

策略层


这样策略本身不需要知道底层数据到底是怎么处理的。

我后来在项目里使用 QuantDash 的 K 线接口，可以直接指定：

这样复权逻辑就可以放在数据入口。

下面是我现在比较常用的一种写法。

import os
import pandas as pd
from quantdash import QuantDash


def get_market_data(symbol: str, count: int = 250) -> pd.DataFrame:
    """获取前复权日线数据。"""
    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.get(
            symbol,
            period="1d",
            count=count,
            adjust="forward",
            to_dataframe=True,
        )

        if df.empty:
            raise ValueError(
                f"{symbol} 没有返回任何行情数据"
            )

        required_columns = {
            "symbol",
            "trade_date",
            "open",
            "high",
            "low",
            "close",
            "volume",
        }

        missing = required_columns - set(df.columns)

        if missing:
            raise ValueError(
                f"行情数据缺少字段: {sorted(missing)}"
            )

        return df.copy()

    except Exception as exc:
        raise RuntimeError(
            f"获取行情失败: {exc}"
        ) from exc


def calculate_factors(df: pd.DataFrame) -> pd.DataFrame:
    """计算简单趋势因子。"""
    result = df.copy()

    result["return_20d"] = (
        result["close"] / result["close"].shift(20) - 1
    )

    result["ma20"] = (
        result["close"].rolling(20).mean()
    )

    result["ma60"] = (
        result["close"].rolling(60).mean()
    )

    result["trend_signal"] = (
        result["ma20"] > result["ma60"]
    )

    return result


def main():
    symbol = "600519.SH"

    try:
        df = get_market_data(symbol)
        result = calculate_factors(df)

        print(
            result[
                [
                    "trade_date",
                    "close",
                    "return_20d",
                    "ma20",
                    "ma60",
                    "trend_signal",
                ]
            ].tail(10).to_string(index=False)
        )

    except Exception as exc:
        print(f"策略数据处理失败: {exc}")


if __name__ == "__main__":
    main()


我比较喜欢这种写法的原因是：

复权方式在函数入口就固定了。

策略层不需要再写一遍复权逻辑。

再进一步，我会做多股票批量计算

当股票池扩大以后，我不会给每只股票单独写请求。

import os
from quantdash import QuantDash


def load_universe(symbols):
    api_key = os.getenv("QUANTDASH_API_KEY")

    if not api_key:
        raise RuntimeError("缺少 QUANTDASH_API_KEY")

    qd = QuantDash(api_key=api_key)

    try:
        dfs = qd.klines.batch(
            symbols,
            period="1d",
            count=120,
            adjust="forward",
            to_dataframe=True,
            show_progress=True,
        )

        if not dfs:
            raise ValueError("批量请求没有返回数据")

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

            df["ma20"] = df["close"].rolling(20).mean()
            df["momentum20"] = (
                df["close"] / df["close"].shift(20) - 1
            )

            print(f"\n--- {symbol} ---")
            print(
                df[
                    [
                        "trade_date",
                        "close",
                        "ma20",
                        "momentum20",
                    ]
                ].tail(5).to_string(index=False)
            )

    except Exception as exc:
        print(f"批量行情处理失败: {exc}")


if __name__ == "__main__":
    load_universe([
        "600519.SH",
        "000001.SZ",
        "000858.SZ",
    ])


这种结构比较适合我的多股票趋势研究。

避坑一：不要让数据源自己“默认决定”复权方式

我现在看到任何量化数据接口，都会先确认：

默认是什么？
能不能显式指定？
前复权到底是什么定义？


因为默认值一旦变化，历史策略结果可能受到影响。

我后来越来越觉得，量化系统里真正容易出问题的地方，往往不是复杂模型，而是这些看起来不起眼的数据定义。

为什么我的回测结果总和实盘对不上？我从复权数据开始排查

算法

开发工具

职业发展

回测与实盘收益差异常源于前复权数据口径不一致。本文通过Python示例，展示如何在数据入口统一复权方式，避免除权事件扭曲趋势、动量信号，并明确回测与实盘的数据契约，提升策略可靠性。

数据一致性

批量计算

数据接口

数据处理

Python

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

为什么我的回测结果总和实盘对不上？我从复权数据开始排查

为什么我的回测结果总和实盘对不上？我从复权数据开始排查

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐