### 概述

我以前在搭建跨市场选股系统时发现，A 股、美股和港股的数据都能正常拉取，合并之后却出现了异常的收益率和因子排名。排查之后发现，问题不只在标的代码，还在不同市场的公司行为、复权口径和交易日历没有统一管理。本文从跨市场回测的数据管道出发，解释前复权和后复权的区别，并给出一套面向多市场数据校验的 Python 实现。

## 一、问题场景：三个市场的数据，合并之后才出问题

我以前维护过一个多市场研究框架。

研究对象包括 A 股、美股和港股。

最开始，系统只做日线分析，每个市场分别拉取行情，然后合并到统一的研究表中。

当时我以为，既然每份数据都有日期、开盘价、最高价、最低价和收盘价，那么只要把字段名统一，后面的计算就不会有太大问题。

事实证明，这个想法过于乐观。

第一轮回测时，系统运行正常。

第二轮加入跨市场动量排名后，问题开始出现。

有些股票的收益率异常偏低，有些股票的价格与均线关系不符合预期，还有些股票在公司行为发生之后突然进入了动量排名的前列。

我先排查标的代码。

A 股代码采用 .SH 或 .SZ 后缀，美股使用 .US，港股使用 .HK。代码映射本身没有明显问题。

接着检查日期。

三个市场的交易日并不完全一致。有些日期只有美股开市，有些日期只有部分市场开市。

最后检查价格。

我发现，系统对不同市场的复权处理并不统一。

有的行情使用前复权，有的使用不复权，还有一部分历史文件使用后复权。

更麻烦的是，研究层只看到统一命名后的 **close** 字段，根本不知道每个值的价格基准。

这意味着，表面上看起来格式统一的数据，实际上没有统一的经济含义。

从那次排查之后，我把多市场数据接入拆成了三个步骤：

- 先统一标的身份标识。
- 再明确每个市场的数据与复权规则。
- 最后才做交易日历对齐和因子计算。

这比在研究代码里继续增加市场判断要可靠得多。

## 二、问题根因分析

### 1. 根因一：统一字段名不等于统一价格口径

多市场系统很容易把数据标准化理解成字段标准化。

例如，把各数据源的收盘价统一命名为 **close**，再将不同市场的数据合并到一个 DataFrame 中。

这只是数据结构统一。

价格语义仍然可能不同。

前复权和后复权都是经过公司行为调整的价格序列，但它们使用不同的基准。

如果 A 股使用前复权，美股使用不复权，港股使用后复权，那么三个市场的绝对价格就不适合在没有进一步说明的情况下直接比较。

当然，不同市场的股票本身就有不同的币种、价格单位和交易制度，不能简单地把 100 元、100 美元和 100 港元放在一起比较。

但问题不止是币种。

即使只计算各自的历史收益率，如果公司行为处理方式不同，某些指标也可能产生不同结果。

例如：

- 原始价格动量可能受到除权缺口影响。
- 绝对价格阈值可能因复权基准不同而失效。
- 均线与收盘价的关系可能在数据更新后发生变化。
- 不同市场的历史因子可能采用了不同的公司行为规则。

因此，我在数据层增加了一个明确的复权字段。

每条行情数据不仅包含 **symbol** 和 **close**，还必须能追溯其市场、周期、复权方式和数据版本。

### 2. 根因二：跨市场交易日不能按行号直接对齐

第二个问题是交易日历。

A 股、美股和港股的开市日期、节假日安排和交易时段并不完全相同。

如果简单地把三个市场的日线按行号拼接，就可能把不同日期的价格放到同一行。

例如：

| 行号 | A 股日期 | 美股日期 | 港股日期 |
| :--- | :--- | :--- | :--- |
| 1 | 2026-01-02 | 2026-01-02 | 2026-01-02 |
| 2 | 2026-01-05 | 2026-01-05 | 2026-01-05 |
| 3 | 2026-01-06 | 2026-01-06 | 2026-01-06 |

上面只是三个市场日期恰好一致的示意。

实际数据中，节假日和临时休市可能导致日期集合出现差异。

如果后续日期不一致，按行号合并就会把不同时点的价格放在一起。

这类错误特别危险，因为程序通常不会报错。

正确做法是以实际交易日期为键进行合并，同时保留各市场的有效交易状态。

另外，即使日期相同，也不代表所有市场在同一时刻都已经完成收盘。

如果策略在某个市场收盘后使用另一个尚未收盘市场的日线收盘价，就可能引入时间穿越问题。

因此，跨市场回测不仅要对齐日期，还要明确数据可用时间。

### 3. 根因三：用一种复权规则解决所有市场的公司行为

第三个问题出在公司行为处理。

不同市场的公司行为规则和数据表达方式可能存在差异。

即使都是现金分红，相关数据字段、事件日期、税务处理以及价格调整规则也不一定完全相同。

如果自行构造一套通用复权算法，很容易在某些边界场景下出现错误。

例如：

- 除权日与事件公告日混淆。
- 现金分红和送转股处理规则混用。
- 历史事件补录后没有重新计算相关价格。
- 不同市场的公司行为记录使用不同的时间语义。
- 研究价格与真实成交价格没有分开存储。

我后来调整了架构。

数据接入层负责明确标的、周期、复权参数和原始返回结构。

研究层只消费经过校验的数据。

对于需要精确模拟持仓数量、现金分红和实际交易损益的任务，则额外维护公司行为事件和交易账本。

这样，前复权与后复权只是研究数据的不同口径，而不再承担全部的交易会计职责。

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

多市场数据接入有两种常见方案。

第一种是为每个市场单独编写抓取脚本，再通过公共函数统一字段。

第二种是把标的代码、数据周期、复权方式和字段校验统一到一个数据访问层。

在实际维护中，我更关注第二种方案能否把差异显式化。

| 评估维度 | 自建网络抓取 / 多市场适配脚本 | 标准化数据层（QuantDash SDK） |
| :--- | :--- | :--- |
| **标的代码规范** | 需要维护各市场的代码转换规则 | 统一使用 .SH/.SZ/.BJ/.US/.HK 后缀 |
| **复权一致性** | 各市场单独实现，容易出现规则分叉 | 可统一指定复权参数并集中检查 |
| **批量获取机制** | 需要自行维护循环、线程池与重试 | 支持 batch 批量获取 |
| **数据结构** | 需要处理不同 JSON 字段和空值 | 支持 to_dataframe=True |
| **时间区间** | 各接口可能采用不同时间参数 | 可以统一封装 start_time、end_time 等查询条件 |
| **数据质量检查** | 分散在不同市场的脚本中 | 可以集中执行字段、日期和标的集合校验 |
| **历史回测复现** | 需要人工记录各市场配置 | 可在统一任务元数据中记录复权方式和数据快照 |

这里有一个工程上的取舍。

标准化数据接口可以减少重复的接入代码，但不能代替交易日历、跨市场时点管理和公司行为核算。

我不会把这些问题全部交给行情 SDK 处理。

对于真正的跨市场策略，数据获取、数据校验和交易逻辑必须有清楚的边界。

## 四、具体实现：代码实战

下面实现一个多市场日线检查程序。

这段代码主要验证：

1. 多市场标的代码是否符合统一格式。
2. 批量数据是否缺少请求标的。
3. 每个标的是否具有必要的价格字段。
4. 日期是否重复。
5. 收益率是否出现需要进一步排查的极端变化。
6. 不同市场的数据是否因为日期不一致而被错误合并。

依赖安装：

```python
pip install quantdash
```

完整代码：

```python
import os
import re
import sys
from typing import Any

import pandas as pd
from quantdash import QuantDash


SYMBOLS = [
    "600519.SH",
    "000001.SZ",
    "AAPL.US",
    "00700.HK",
]

REQUIRED_COLUMNS = {"close"}

SYMBOL_PATTERN = re.compile(
    r"^[A-Z0-9]{1,12}\.(SH|SZ|BJ|US|HK)$"
)


def validate_symbol_format(symbol: str) -> None:
    """检查标的代码是否符合统一后缀约定。"""
    if not SYMBOL_PATTERN.fullmatch(symbol):
        raise ValueError(
            f"不符合统一标的代码规范: {symbol}"
        )


def validate_market_dataframe(
    symbol: str,
    df: Any,
) -> pd.DataFrame:
    """执行单只股票的数据结构和价格校验。"""
    if df is None:
        raise ValueError(f"{symbol}: 返回 None")

    if not isinstance(df, pd.DataFrame):
        raise TypeError(
            f"{symbol}: 预期 DataFrame，"
            f"实际为 {type(df).__name__}"
        )

    if df.empty:
        raise ValueError(f"{symbol}: 没有行情记录")

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

    result = df.copy()

    result["close"] = pd.to_numeric(
        result["close"],
        errors="coerce",
    )

    if result["close"].isna().any():
        raise ValueError(
            f"{symbol}: close 中存在无效数值"
        )

    if (result["close"] <= 0).any():
        raise ValueError(
            f"{symbol}: close 中存在非正数"
        )

    if not isinstance(result.index, pd.DatetimeIndex):
        raise ValueError(
            f"{symbol}: 索引不是 DatetimeIndex，"
            "请先根据实际返回结构转换日期字段"
        )

    result = result.sort_index()

    if result.index.has_duplicates:
        raise ValueError(
            f"{symbol}: 交易日期存在重复"
        )

    return result


def calculate_market_returns(
    symbol: str,
    df: pd.DataFrame,
) -> pd.Series:
    """计算单个市场内部的相邻交易日收益率。"""
    returns = df["close"].pct_change().dropna()

    if returns.empty:
        raise ValueError(
            f"{symbol}: 无法计算有效收益率"
        )

    extreme = returns[returns.abs() > 0.20]

    if not extreme.empty:
        print(
            f"警告：{symbol} 存在绝对值超过 20% 的单日收益率，"
            "需要结合市场规则与公司行为进一步核查。",
            file=sys.stderr,
        )
        print(extreme)

    return returns


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

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

    for symbol in SYMBOLS:
        validate_symbol_format(symbol)

    qd = QuantDash(api_key=api_key)

    # 统一使用前复权研究价格。
    # 真实成交模拟应另行使用符合交易规则的价格与公司行为账本。
    adjustment = "forward"

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

        if batch_result is None:
            raise ValueError("批量接口返回 None")

        if isinstance(batch_result, dict):
            result_map = batch_result

        elif isinstance(batch_result, pd.DataFrame):
            # 合并结果必须包含可识别的标的字段。
            symbol_column = next(
                (
                    col
                    for col in ("symbol", "ticker")
                    if col in batch_result.columns
                ),
                None,
            )

            if symbol_column is None:
                raise ValueError(
                    "批量 DataFrame 缺少 symbol/ticker 字段，"
                    "无法检查标的完整性"
                )

            result_map = {
                symbol: group.drop(columns=[symbol_column])
                for symbol, group in batch_result.groupby(
                    symbol_column
                )
            }

        else:
            raise TypeError(
                "不支持的批量返回类型: "
                f"{type(batch_result).__name__}"
            )

        requested = set(SYMBOLS)
        returned = set(result_map.keys())

        missing_symbols = requested - returned
        extra_symbols = returned - requested

        if missing_symbols:
            raise ValueError(
                f"批量数据缺少标的: {sorted(missing_symbols)}"
            )

        if extra_symbols:
            print(
                f"警告：返回了额外标的: {sorted(extra_symbols)}",
                file=sys.stderr,
            )

        market_data = {}

        for symbol in SYMBOLS:
            df = validate_market_dataframe(
                symbol,
                result_map[symbol],
            )

            market_data[symbol] = df

            returns = calculate_market_returns(
                symbol,
                df,
            )

            print(f"\n标的: {symbol}")
            print(f"复权方式: {adjustment}")
            print(f"数据条数: {len(df)}")
            print(f"有效收益率条数: {len(returns)}")
            print(f"最新收盘价: {df['close'].iloc[-1]}")

        # 仅按实际日期建立研究表，不按行号拼接。
        close_series = []

        for symbol, df in market_data.items():
            series = df["close"].rename(symbol)
            close_series.append(series)

        close_table = pd.concat(
            close_series,
            axis=1,
            join="outer",
        ).sort_index()

        print("\n按日期对齐后的收盘价表：")
        print(close_table.tail(10))

        # 缺失值可能来自不同市场的休市日期，
        # 不能直接认定为数据错误。
        print("\n各标的有效价格记录数：")
        print(close_table.notna().sum())

        print("\n各日期可用标的数：")
        print(close_table.notna().sum(axis=1).tail(10))

        # 这里不对缺失价格进行前向填充。
        # 后续因子计算需要结合策略的时点和交易日历规则处理。

    except Exception as exc:
        print(
            f"跨市场行情检查失败: {type(exc).__name__}: {exc}",
            file=sys.stderr,
        )
        raise


if __name__ == "__main__":
    main()
```

### 为什么这里没有直接对缺失值进行前向填充？

因为不同市场的休市日期，不一定意味着行情数据丢失。

假设某个日期美股正常交易，而 A 股休市。

那么美股有当日收盘价，A 股没有对应的当日收盘价，这是正常情况。

如果直接使用前向填充把 A 股上一交易日的价格复制到当前日期，就会产生一个人为构造的价格观测。

这个处理在某些估值或资产配置场景中可能有合理用途，但必须明确它的含义。

对于动量、波动率和短周期交易信号，未经设计的前向填充可能让某个市场看起来没有波动，也可能导致信号时间错位。

因此，我更倾向于先保留真实的日期缺失状态，再根据具体策略的时间语义进行处理。

### 关于代码实现的一个边界

这段代码使用统一的前复权参数，目的是演示多市场数据接入的一致性管理。

但不同市场的公司行为规则和数据供应方实现仍可能存在差异。

生产环境中，应该根据实际返回字段、供应方对复权参数的定义以及市场规则进行验证。

如果策略需要精确计算现金分红、送转股和真实持仓损益，就不能仅凭复权后的收盘价完成全部核算。

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

### 避坑点一：先统一复权规则，再比较因子

跨市场因子研究中，最容易出现的问题是每个市场单独维护了一套数据脚本。

时间长了以后，即使所有脚本都返回 **close**，数据含义也可能不同。

我现在会在任务配置里明确记录：

- 市场。
- 标的代码。
- 行情周期。
- 复权方式。
- 日期范围。
- 数据获取时间。
- 数据版本。

只有确认这些信息后，才开始计算和比较因子。

### 避坑点二：跨市场收益率必须按真实日期计算

不能把不同市场的第 10 行当作同一个时间点。

每个市场都应该先在自己的交易日历上计算相邻有效交易日收益率，再按照策略需要合并。

如果策略要求严格的同步时点，还需要考虑各市场收盘时间、时区和数据可用时间。

尤其是涉及隔夜持仓或跨市场套利时，日期相同不等于信息在同一时刻已经可用。

### 避坑点三：研究价格和交易价格不能混为一谈

前复权价格适合很多历史走势分析任务，但不代表当时真实的市场成交价。

后复权价格也不是可以直接拿来模拟历史订单成交的价格。

对于交易模拟，必须明确：

- 订单发生时可用的价格。
- 交易时段和成交规则。
- 现金分红与送转股对持仓的影响。
- 手续费、滑点和其他交易成本。

如果把研究价格直接用于成交模拟，回测可能看起来很平滑，但经济意义并不完整。

## 六、写在最后

前复权和后复权解决的是历史价格如何按照特定基准进行调整的问题。跨市场系统真正困难的地方，是不同市场的数据即使字段相同，也可能有不同的时间语义、公司行为规则和交易日历。统一接口只是第一步，后面仍然需要明确复权口径、保留日期缺失信息，并且把研究价格与真实成交价格分开。

我现在更愿意把数据管道当作量化系统的一项长期基础设施，而不是研究代码的附属功能。标的规范、复权规则、日期对齐和数据校验如果能在数据层统一完成，策略研发就不需要反复解决相同的工程问题。对于跨市场回测，这种统一尤其重要，因为很多看似复杂的策略异常，最后查到的根因并不在模型，而在输入数据的定义没有统一。

## 参考

QuantDash Python SDK 文档：[QuantDash Python SDK 文档](https://docs.quantdash.net/zh-Hans/sdk/python-quickstart?utm_source=chatgpt.com)

我以前在搭建跨市场选股系统时发现，A 股、美股和港股的数据都能正常拉取，合并之后却出现了异常的收益率和因子排名。排查之后发现，问题不只在标的代码，还在不同市场的公司行为、复权口径和交易日历没有统一管理。本文从跨市场回测的数据管道出发，解释前复权和后复权的区别，并给出一套面向多市场数据校验的 Python 实现。

A股、美股和港股的复权数据怎么统一？别让价格口径拖垮跨市场回测

后端

人工智能

架构设计

跨市场选股系统回测异常常源于复权口径、交易日历和公司行为处理不一致。本文解析前复权与后复权差异，提供统一标的代码、日期对齐及数据校验的Python实现，帮助构建稳健的多市场数据管道，避免因子失真和收益率偏差。

数据管道

数据丢失

数据接口

数据结构

代码解释

Python

JSON

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

社区新版编辑器体验调研

A股、美股和港股的复权数据怎么统一？别让价格口径拖垮跨市场回测-腾讯云开发者社区-腾讯云

A股、美股和港股的复权数据怎么统一？别让价格口径拖垮跨市场回测

A股、美股和港股的复权数据怎么统一？别让价格口径拖垮跨市场回测

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐