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

去年下半年，我在做AH股溢价套利策略的回测时栽了一个大跟头。

策略逻辑很简单：同时持有贵州茅台（600519.SH）和腾讯控股（

），当AH溢价率偏离历史均值超过两个标准差时进行调仓。我在本地用akshare拿A股数据、用yfinance拿港股数据，费了整整两天把字段名对齐、时区转换、代码映射表写完后，回测结果漂亮得吓人——年化收益38%，夏普比率1.6。

然后实盘跑了一个月，亏了7%。

我当时的第一反应是策略过拟合了，但把回测日志和实盘日志逐日对比后，发现了一个致命问题：

我的回测在2025年9月18日（中秋节假期前一天）提前“看到”了港股的收盘价

。那天A股正常交易，港股因为中秋节翌日休市。但在我的数据合并逻辑中，由于没有正确处理交易日历，导致A股当天的决策用到了港股前一天的收盘价——这在逻辑上相当于偷看了未来数据。

更隐蔽的问题是复权。贵州茅台在2025年6月有过一次分红除权，我本地计算的复权因子因为除权基准日取错了，导致2025年6月之前的回测价格全部偏高。回测年化被虚高了将近5个百分点。

这两个问题叠加，让一个本来勉强能盈利的策略在回测中被包装成了明星策略。

深入排查后，我发现跨境多资产数据合并回测有三个层次的技术陷阱：

A股、港股、美股的休市安排完全不同。中秋节A股放假但港股可能正常交易，感恩节美股休市但A股和港股照常运行。直接做 

 会产生大量NaN，而很多初学者会用 

一旦用了后向填充（bfill），就等于用未来的数据填充了今天的空缺，这在回测中是致命的未来函数

这是最隐蔽的坑。北京时间比港股时间快0小时（同属UTC+8），但比美股慢12-13小时。如果在数据合并时直接用物理日期（date）对齐而不带时区信息，可能会出现在A股盘中（北京时间14:00）用到了当晚才开盘的美股收盘价。虽然港股通标的不涉及美股，但同样的逻辑适用于A股与港股之间的时间边界处理——比如港股16:00收盘，A股15:00收盘，如果不对齐时间戳，尾盘数据的归属就会错乱。

不同数据源返回的复权价格基准不同。有的默认前复权、有的默认后复权、有的干脆不复权。更麻烦的是，

同一只股票在不同数据源的复权因子计算方式可能不一致

——有的是比例复权（乘法），有的是差值复权（加减法）。把贵州茅台从akshare拿的前复权和腾讯控股从yfinance拿的前复权直接合并，本质上是在用两套不同的价格体系做回测。

发现问题后，我尝试过三种方案。

方案一：自建数据管道 + 本地复权

第一反应是自己写一套ETL，把akshare和yfinance的数据统一清洗、统一复权。我写了一个300多行的数据清洗模块，包含交易日历映射表、时区转换函数、复权因子计算器。跑了两个月，维护成本越来越高——akshare的接口字段变了两次、yfinance的时区处理逻辑也调整过，每次都要改代码。更关键的是，

本地计算复权因子这件事本身就容易出错

，尤其是遇到多次除权除息叠加的情况。

方案二：爬虫 + 手动维护交易日历

有人建议我直接爬交易所官网的交易日历，然后手工维护一个全局日历表。我试了一下——A股、港股、美股三套日历，节假日每年都在变，还要处理台风天临时休市这种突发事件。这套方案在投入两周后被我放弃了，因为我发现自己变成了一个“交易日历运维工程师”，而不是量化策略研究员。

方案三：统一数据接口 + 服务器端复权

后来我在找替代方案时发现了一个叫 

 的工具。它的核心思路很简单：

所有市场用同一套API、同一套代码格式、同一套字段规范，复权在服务端完成

。这意味着我不再需要维护一堆 

更重要的是复权逻辑交给服务端——我只需要在请求时指定 

，服务器返回的就是已经复权好的价格。这彻底消除了本地计算复权因子引入未来函数的风险。

下面是我最终采用的方案——用 QuantDash 统一获取A股和港股通数据，并通过标准化的时序对齐完成合并回测。

# 依赖安装: pip install quantdash pandas

import os
import datetime
import pandas as pd
from quantdash import QuantDash

# 通过环境变量读取 API Key（不要硬编码在代码里）
# export QUANTDASH_API_KEY="your-api-key-here"
qd = QuantDash()  # 自动读取环境变量 QUANTDASH_API_KEY[reference:15]

def fetch_multimarket_data(symbols: list, start_date: str, end_date: str):
    """
    批量获取多市场K线数据，统一返回DataFrame字典
    支持 A股(.SH/.SZ) 和港股(.HK)
    """
    # 将日期字符串转换为毫秒时间戳[reference:16]
    start_dt = datetime.datetime.strptime(start_date, "%Y-%m-%d")
    end_dt = datetime.datetime.strptime(end_date, "%Y-%m-%d")
    start_ts = int(start_dt.timestamp() * 1000)
    end_ts = int(end_dt.timestamp() * 1000)
    
    try:
        # 批量获取，服务端自动完成前复权（比例复权）[reference:17]
        dfs = qd.klines.batch(
            symbols=symbols,
            period="1d",
            start_time=start_ts,
            end_time=end_ts,
            adjust="forward",  # 前复权-比例复权，适合计算收益率
            to_dataframe=True,
            show_progress=True
        )
        return dfs
    except Exception as e:
        print(f"数据获取失败: {e}")
        return {}

# 定义股票池：A股 + 港股通标的
symbols = [
    "600519.SH",  # 贵州茅台（A股）
    "000858.SZ",  # 五粮液（A股）
    "00700.HK",   # 腾讯控股（港股通）
    "09988.HK",   # 阿里巴巴（港股通）
]

dfs = fetch_multimarket_data(symbols, "2024-01-01", "2026-06-30")

# 查看各标的的数据量
for sym, df in dfs.items():
    if not df.empty:
        name = df['name'].iloc[0] if 'name' in df.columns else sym
        print(f"{sym} ({name}): {len(df)} 条日线")

4.3 时序对齐：解决交易日历不一致

这是整个合并过程中最关键的一步。

 不能简单地对所有日期取并集或交集，需要根据策略需求选择合适的对齐策略。

def align_multimarket_data(dfs: dict, align_method: str = "union"):
    """
    多市场数据时序对齐
    
    参数:
        dfs: {symbol: DataFrame} 字典，每个DataFrame需包含 trade_date 列
        align_method: 
            - "union": 取所有市场交易日的并集（适合需要每日判断仓位的策略）
            - "intersection": 取交集（适合要求所有市场同时交易的策略）
    
    返回:
        对齐后的DataFrame，列为各标的的收盘价
    """
    # 提取每个标的的收盘价序列，以 trade_date 为索引
    price_series = {}
    for sym, df in dfs.items():
        if df.empty:
            continue
        # trade_date 已经是日期字符串，转换为 datetime 索引[reference:19]
        df_copy = df.copy()
        df_copy['trade_date'] = pd.to_datetime(df_copy['trade_date'])
        df_copy.set_index('trade_date', inplace=True)
        price_series[sym] = df_copy['close'].rename(sym)
    
    # 构建全局交易日历
    if align_method == "union":
        # 并集：只要有一个市场开盘，该日就是有效交易日[reference:20]
        all_dates = pd.DatetimeIndex([])
        for series in price_series.values():
            all_dates = all_dates.union(series.index)
        all_dates = all_dates.sort_values()
    else:  # intersection
        # 交集：所有市场都必须开盘
        all_dates = None
        for series in price_series.values():
            if all_dates is None:
                all_dates = series.index
            else:
                all_dates = all_dates.intersection(series.index)
        all_dates = all_dates.sort_values()
    
    # 合并所有价格序列到统一的DataFrame
    aligned_df = pd.DataFrame(index=all_dates)
    for sym, series in price_series.items():
        aligned_df[sym] = series
    
    # ⚠️ 关键：只使用前向填充（ffill），绝不使用后向填充（bfill）
    # 前向填充：用前一个交易日的价格填充今日（如果今日休市）
    # 后向填充：会用未来的价格填充今日——这是未来函数！[reference:21]
    aligned_df = aligned_df.ffill()
    
    return aligned_df

# 执行对齐
aligned_prices = align_multimarket_data(dfs, align_method="union")
print(f"\n对齐后数据形状: {aligned_prices.shape}")
print(f"时间范围: {aligned_prices.index[0]} 至 {aligned_prices.index[-1]}")
print(aligned_prices.tail(10))

有了对齐后的价格矩阵，就可以计算策略所需的各类指标了：

def calculate_returns(aligned_df: pd.DataFrame):
    """
    基于对齐后的价格计算日收益率
    """
    # 日收益率
    returns = aligned_df.pct_change().dropna()
    
    # 计算AH溢价率（示例：茅台 vs 腾讯）
    if '600519.SH' in aligned_df.columns and '00700.HK' in aligned_df.columns:
        # 注意：币种不同，这里需要做汇率转换，示例中省略
        # 实际使用中需要引入USD/CNY或HKD/CNY汇率
        aligned_df['ah_spread'] = (
            aligned_df['600519.SH'] / aligned_df['00700.HK'] - 1
        ) * 100
        print(f"AH溢价率均值: {aligned_df['ah_spread'].mean():.2f}%")
        print(f"AH溢价率标准差: {aligned_df['ah_spread'].std():.2f}%")
    
    return returns

returns = calculate_returns(aligned_prices)

# 计算策略的累计收益（等权重组合示例）
portfolio_returns = returns.mean(axis=1)
cumulative_ret = (1 + portfolio_returns).cumprod()
print(f"\n等权重组合累计收益: {(cumulative_ret.iloc[-1] - 1) * 100:.2f}%")

将以上代码整合为一个完整的可运行脚本：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
跨市场（A股 + 港股通）数据合并回测示例
依赖: pip install quantdash pandas
环境变量: QUANTDASH_API_KEY
"""

import os
import datetime
import pandas as pd
from quantdash import QuantDash

# ============ 配置 ============
SYMBOLS = [
    "600519.SH",  # 贵州茅台
    "000858.SZ",  # 五粮液
    "00700.HK",   # 腾讯控股
    "09988.HK",   # 阿里巴巴
]
START_DATE = "2024-01-01"
END_DATE = "2026-06-30"

# ============ 1. 初始化 ============
api_key = os.getenv("QUANTDASH_API_KEY")
if not api_key:
    raise ValueError("请设置环境变量 QUANTDASH_API_KEY")
qd = QuantDash(api_key=api_key)

# ============ 2. 获取数据 ============
def fetch_data():
    start_dt = datetime.datetime.strptime(START_DATE, "%Y-%m-%d")
    end_dt = datetime.datetime.strptime(END_DATE, "%Y-%m-%d")
    start_ts = int(start_dt.timestamp() * 1000)
    end_ts = int(end_dt.timestamp() * 1000)
    
    try:
        dfs = qd.klines.batch(
            symbols=SYMBOLS,
            period="1d",
            start_time=start_ts,
            end_time=end_ts,
            adjust="forward",
            to_dataframe=True,
            show_progress=True
        )
        return dfs
    except Exception as e:
        print(f"❌ 数据获取失败: {e}")
        return {}

print("📊 正在获取多市场数据...")
dfs = fetch_data()

if not dfs:
    print("❌ 未获取到任何数据，请检查 API Key 和网络")
    exit(1)

# ============ 3. 数据预览 ============
print("\n📋 数据概览:")
for sym, df in dfs.items():
    if not df.empty:
        name = df['name'].iloc[0] if 'name' in df.columns else sym
        print(f"  {sym} ({name}): {len(df)} 条")

# ============ 4. 时序对齐 ============
print("\n🔄 正在进行时序对齐...")
price_series = {}
for sym, df in dfs.items():
    if df.empty:
        continue
    df_copy = df.copy()
    df_copy['trade_date'] = pd.to_datetime(df_copy['trade_date'])
    df_copy.set_index('trade_date', inplace=True)
    price_series[sym] = df_copy['close'].rename(sym)

# 并集对齐
all_dates = pd.DatetimeIndex([])
for series in price_series.values():
    all_dates = all_dates.union(series.index)
all_dates = all_dates.sort_values()

aligned_df = pd.DataFrame(index=all_dates)
for sym, series in price_series.items():
    aligned_df[sym] = series
aligned_df = aligned_df.ffill()  # 仅前向填充

print(f"✅ 对齐完成: {len(aligned_df)} 个交易日")
print(aligned_df.head(10))

# ============ 5. 简单策略回测 ============
print("\n📈 计算策略收益...")
# 等权重组合
returns = aligned_df.pct_change().dropna()
portfolio_returns = returns.mean(axis=1)
cumulative = (1 + portfolio_returns).cumprod()

print(f"  累计收益: {(cumulative.iloc[-1] - 1) * 100:.2f}%")
print(f"  年化收益: {((cumulative.iloc[-1]) ** (252 / len(cumulative)) - 1) * 100:.2f}%")
print(f"  最大回撤: {((cumulative / cumulative.cummax() - 1).min() * 100):.2f}%")

print("\n✅ 回测完成！")

改用这套方案后，我的数据工程工作量减少了约80%。原本需要维护的300行数据清洗代码被压缩到了不到100行，而且

再也不用担心复权因子算错、时区搞混、交易日历漏更新

以下是三个我在实践中总结的避坑建议：

避坑1：永远只用前向填充（ffill），绝不用后向填充（bfill）

在合并多市场数据时，遇到某市场休市导致的NaN，

只能用前一天的数值填充（ffill），绝对不能用后一天的数值填充（bfill）

。后向填充等于用未来的数据填补今天的空白，这在回测中会引入致命的未来函数。

避坑2：复权方式的选择要匹配策略类型

（前复权-比例）做回测，因为它能让历史价格向后调整，保证收益率计算的准确性。但如果策略涉及价格绝对值（如突破策略），可能需要考虑使用 

避坑3：港股通数据注意币种和汇率

港股通标的是港币计价，A股是人民币计价。如果策略涉及两个市场的相对价值（如AH溢价），

，否则计算出的溢价率是严重失真的。我一般用每日的HKD/CNY中间价来做转换，这部分数据可以通过 QuantDash 的 

 接口获取汇率相关标的，或者接入独立的汇率数据源。

回顾这次从“回测年化38%”到“实盘亏7%”的经历，我最大的教训是：

跨境多资产回测的坑不在策略逻辑本身，而在数据工程层面

交易日历不一致、时区差异、复权基准不统一——这三个问题单独看都不复杂，但叠加在一起就会产生各种“看起来合理、实际上致命”的数据错误。而最可怕的是，

这些错误不会让程序报错，它们只会让回测结果变得异常漂亮

，然后在实盘中让你付出真金白银的代价。

我现在在用的 QuantDash 帮我解决的是“数据获取和复权”这一层的问题——统一接口、服务器端复权、标准化的字段格式。但“时序对齐”和“策略逻辑的时序一致性”这一层，仍然需要开发者自己根据策略需求来设计。没有银弹，但好的工具至少能让你把精力聚焦在真正重要的事情上。

如果你也在做跨境多资产的量化策略，我的建议是：

先把数据工程的地基打牢，再谈策略优化

。地基不牢，楼盖得越高，摔得越惨。

Q1: 港股通标的的代码后缀是什么？和普通港股有区别吗？

A: QuantDash 对港股通标的和普通港股使用相同的代码后缀 

。港股通标的和普通港股在数据接口层面没有区别，你只需要知道标的代码即可。是否属于港股通标的需要在策略层面自行维护一个白名单。

Q2: 如果我只想做A股和港股通的合并回测，需要处理美股时区问题吗？

A: 不需要。A股和港股同属UTC+8时区，没有时差问题。但要注意

：A股15:00收盘，港股16:00收盘。如果使用日线数据（1d），这个问题不大，因为日线对应的是全天数据。但如果使用分钟线，就需要注意时间边界的对齐。

Q3: 前复权和后复权在回测中应该选哪个？

。前复权让历史价格向后调整，使得历史收益率计算更准确，且最新价格就是实际价格，便于与实盘对照。后复权更适合观察价格的长期绝对走势，但最新价格不是实际价格，在回测中容易造成混淆。我个人的经验是：回测用前复权，实盘信号用实时价格，两者分开处理。

策略逻辑很简单：同时持有贵州茅台（600519.SH）和腾讯控股（00700.HK），当AH溢价率偏离历史均值超过两个标准差时进行调仓。我在本地用akshare拿A股数据、用yfinance拿港股数据，费了整整两天把字段名对齐、时区转换、代码映射表写完后，回测结果漂亮得吓人——年化收益38%，夏普比率1.6。

港股通数据与A股数据怎么合并回测？——跨境多资产时序对齐的工程实践

数据库

算法

架构设计

港股通与A股合并回测的核心陷阱在于交易日历不一致、时区差导致未来函数及复权基准不统一。本文从AH溢价策略回测失真案例出发，分享通过统一数据接口、服务器端复权及时序对齐，安全完成跨境多资产回测，并强调仅用前向填充避免未来数据泄露。

数据清洗

数据管道

数据接口

服务器

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

港股通数据与A股数据怎么合并回测？——跨境多资产时序对齐的工程实践

港股通数据与A股数据怎么合并回测？——跨境多资产时序对齐的工程实践

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐