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

我第一次真正开始做多市场量化时，原本以为最大的困难是策略。

当时我的研究框架同时覆盖 A 股、港股和美股。

最开始，我分别找了不同的数据接口。

最后策略代码里面出现了大量类似：

“如果是 A 股就这么处理，如果是港股就换一个字段，如果是美股又是另外一个时间格式。”

如果数据层没有统一，策略层迟早会变成条件判断的集合。

而复权问题又让事情更加复杂。

我希望策略研究层拿到的是统一的数据结构，但同时又不能把不同市场的原始价格、复权价格和实时行情混在一起。

所以我后来把数据工程的重点放在了“统一接口 + 明确价格语义”上。

标的代码必须包含交易所或市场后缀。

这样做看起来只是规范字符串，但对于量化系统来说意义很大。

“这个 6 开头的是上海股票。”

这种隐式规则越多，后期维护越困难。

2. 不同市场的价格研究逻辑仍然需要统一

虽然市场不同，但从策略框架看，我最终需要的通常还是：

因此我更愿意让数据访问层负责处理市场差异。

策略层只关心统一的数据结构。

3. 复权是数据工程的一部分，而不是策略临时计算

以前我会在策略代码里临时处理复权。

因为不同策略可能采用不同复权方式。

一个策略使用前复权比例价格。

第三个策略还需要原始行情做成交模拟。

如果所有事情都在策略代码里做，最后很难审计。

所以我现在会把复权作为行情数据层的明确参数。

方案一：每个市场维护一套数据适配器

这个方案一开始看起来很专业。

不同数据源的字段名称、时间格式、股票代码和异常处理方式都会出现差异。

我最终发现，大量时间都花在“把数据变成一样”上。

方案二：自己建设完整数据管道

原始行情 → 公司行为 → 除权因子 → 复权计算 → 数据库存储。

但如果项目本身不是专门的数据供应系统，维护成本非常高。

尤其是数据补发、历史修订和异常重试。

后来我选择了一种更务实的方案：

底层数据获取统一，上层研究逻辑保持独立。

我在 Python 项目中使用 QuantDash 的 K 线、批量 K 线、除权因子和实时行情接口，策略层只接受整理后的 DataFrame。

这样至少解决了一个长期困扰我的问题：

策略代码不再关心数据来自哪个市场。

下面这段代码就是我现在比较常用的基础模式。

import os
import sys
from quantdash import QuantDash


def main():
    if not os.getenv("QUANTDASH_API_KEY"):
        print("未检测到 QUANTDASH_API_KEY", file=sys.stderr)
        sys.exit(1)

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

    try:
        qd = QuantDash()

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

        for symbol, df in dfs.items():
            print(f"\n--- {symbol} ---")

            if df.empty:
                print("没有返回数据")
                continue

            fields = [
                "symbol",
                "name",
                "trade_date",
                "open",
                "high",
                "low",
                "close",
                "volume"
            ]

            available = [
                field for field in fields
                if field in df.columns
            ]

            print(
                df[available].tail(5).to_string(index=False)
            )

    except Exception as exc:
        print(f"批量获取行情失败: {exc}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()


这里我特别推荐使用批量接口。

如果研究的是一个股票池，我不希望策略循环里反复创建客户端和发送零散请求。

多市场系统里，我还会先检查标的基础信息。

import os
import sys
from quantdash import QuantDash


def main():
    if not os.getenv("QUANTDASH_API_KEY"):
        print("未检测到 QUANTDASH_API_KEY", file=sys.stderr)
        sys.exit(1)

    symbols = [
        "600519.SH",
        "000001.SZ"
    ]

    try:
        qd = QuantDash()

        instruments = qd.instruments.get(symbols)

        if not instruments:
            print("没有获取到标的信息")
            return

        for item in instruments:
            ext = item.get("ext", {})

            print(
                f"{item.get('symbol', ''):>10s} "
                f"{item.get('name', ''):<8s} "
                f"交易所={item.get('exchange', ''):<4s} "
                f"类型={item.get('type', ''):<8s} "
                f"上市={ext.get('listing_date', '')}"
            )

    except Exception as exc:
        print(f"获取标的信息失败: {exc}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()


我会把这一步放在数据进入研究数据库之前。

因为很多所谓的“行情异常”，其实是标的映射异常。

当策略研究需要比较不同价格口径时，我会直接通过接口指定复权方式。

import os
import sys
from quantdash import QuantDash


def main():
    if not os.getenv("QUANTDASH_API_KEY"):
        print("未检测到 QUANTDASH_API_KEY", file=sys.stderr)
        sys.exit(1)

    try:
        qd = QuantDash()
        symbol = "600519.SH"

        adjustments = [
            "none",
            "forward",
            "backward",
            "forward_additive",
            "backward_additive"
        ]

        for adjustment in adjustments:
            df = qd.klines.get(
                symbol,
                period="1d",
                count=5,
                adjust=adjustment,
                to_dataframe=True
            )

            if df.empty:
                print(f"{adjustment}: 没有数据")
                continue

            print(
                f"\n复权方式：{adjustment}"
            )

            print(
                df[
                    [
                        "trade_date",
                        "open",
                        "high",
                        "low",
                        "close"
                    ]
                ].tail().to_string(index=False)
            )

    except Exception as exc:
        print(f"读取复权数据失败: {exc}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()


我一般不会让策略开发者自己记住复权公式。

“你要的是哪一种价格序列。”

如果我要审计价格变化，会单独读取除权因子。

import os
import sys
from quantdash import QuantDash


def main():
    if not os.getenv("QUANTDASH_API_KEY"):
        print("未检测到 QUANTDASH_API_KEY", file=sys.stderr)
        sys.exit(1)

    try:
        qd = QuantDash()

        df = qd.klines.ex_factors(
            ["600519.SH"],
            to_dataframe=True
        )

        if df.empty:
            print("没有获取到除权因子")
            return

        print(
            df[
                [
                    "symbol",
                    "trade_date",
                    "ex_factor"
                ]
            ].tail(10).to_string(index=False)
        )

    except Exception as exc:
        print(f"读取除权因子失败: {exc}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()


这也是我认为量化数据系统非常值得保留的一张“审计表”。

当某一天价格突然变化，我至少能知道应该往哪里查。

如果股票池规模变大，我会进一步使用批量接口。

import os
import sys
from quantdash import QuantDash


def main():
    if not os.getenv("QUANTDASH_API_KEY"):
        print("未检测到 QUANTDASH_API_KEY", file=sys.stderr)
        sys.exit(1)

    symbols = [
        "600519.SH",
        "000001.SZ",
        "000858.SZ"
    ]

    try:
        qd = QuantDash()

        result = qd.klines.batch(
            symbols,
            period="1d",
            count=20,
            adjust="forward",
            to_dataframe=True,
            show_progress=True
        )

        for symbol, df in result.items():
            if df.empty:
                print(f"{symbol}: 没有数据")
                continue

            close = df["close"].dropna()

            if close.empty:
                print(f"{symbol}: close 字段为空")
                continue

            print(
                f"{symbol}: "
                f"最新日期={df['trade_date'].iloc[-1]}, "
                f"最新收盘={close.iloc[-1]}"
            )

    except Exception as exc:
        print(f"批量行情处理失败: {exc}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()


我在实际项目里会把这一步进一步封装成定时任务，然后将结果写入自己的研究数据库。

避坑一：不要让策略层处理市场差异

我现在尽可能让策略看到统一的数据结构。

symbol、trade_date、open、high、low、close、volume。

至于这个股票属于沪市、深市、港股还是美股，由数据层处理。

避坑二：不要默认所有市场都可以使用同一个交易日逻辑

这是多市场量化最容易被忽视的问题。

A 股、港股、美股的交易日并不完全相同。

如果策略进行跨市场比较，我会首先建立各市场自己的交易日历，再做时间对齐。

行情 API 能解决数据获取问题，但交易日语义仍然需要策略系统自己明确。

避坑三：复权方式必须写进策略配置

我现在不允许策略代码里出现一个没有解释的默认价格。

Python 量化数据工程实践：如何正确处理 A 股、美股、港股的复权价格

编程语言

大数据

后端

构建多市场量化数据管道，核心在于统一行情访问层与明确价格语义。本文从实战出发，解决A股、港股、美股数据一致性难题，详解如何通过批量接口高效获取K线、除权因子及五种复权数据，建立一套策略层无感、数据层统一的回测数据处理流程，并附Python代码实例与关键避坑指南。

数据一致性

数据清洗

数据管道

数据审计

数据接口

数据结构

数据处理

数据库

Python

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

Python 量化数据工程实践：如何正确处理 A 股、美股、港股的复权价格

Python 量化数据工程实践：如何正确处理 A 股、美股、港股的复权价格

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐