### 📌 摘要

> 当量化策略从单一 A 股扩展到美股、港股时，真正棘手的问题往往不是 API 调用，而是不同市场的数据结构和代码规则如何统一。本文从我实际搭建跨市场量化数据层的经历出发，介绍如何利用统一的股票代码规则和 DataFrame 输出，把不同市场的数据接入同一套 Pandas 分析流程。

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

我第一次认真做跨市场策略时，最先踩的坑并不是交易规则，而是数据。

当时策略研究人员希望把 A 股、美股和港股放进同一个股票池，做一个非常基础的横截面因子。

理论上，逻辑非常简单：

```text
获取股票
→ 获取历史 K 线
→ 转 DataFrame
→ 计算收益率
→ 横截面比较
```

真正实现以后，我发现第一步就已经开始分叉。

A 股可能写成：

```text
600519
```

另一个数据接口可能要求：

```text
600519.SH
```

美股则是：

```text
AAPL
```

港股又可能是：

```text
00700
```

如果数据访问层不统一，到了策略层，我就不得不写大量市场判断。

后来我甚至遇到过这样的情况：

研究员拿到的是一张 DataFrame，里面股票代码没有市场后缀；另一个程序拿到的数据却包含交易所信息。

两边代码看起来都对，但是合并时无法直接匹配。

所以后来我在跨市场数据层里坚持一个原则：

> **标的身份必须标准化，DataFrame 结构必须标准化，策略代码才有可能标准化。**

---

## 二、问题根因分析

### 1. 股票代码不是一个完整的身份

对量化系统来说，单独一个股票代码并不能永远唯一确定一个标的。

因此我更习惯使用：

```text
代码.交易所后缀
```

目前 SDK 支持的格式包括：

| 市场 | 后缀 | 示例 |
|---|---|---|
| 上海 | .SH | 600519.SH |
| 深圳 | .SZ | 000001.SZ |
| 北京 | .BJ | 920047.BJ |
| 美股 | .US | AAPL.US |
| 港股 | .HK | 00700.HK |

这个设计对我最大的帮助，是策略层不需要自己判断股票属于哪个市场。

### 2. 数据结构统一比接口数量更重要

我以前经常听到一个问题：

“有没有 A 股、美股、港股统一数据接口？”

但我现在认为，这个问题还不够完整。

真正应该问的是：

> **不同市场的数据能不能进入同一个分析模型？**

如果三个市场最终返回三个不同的 DataFrame Schema，那么即使它们都能获取数据，对量化系统来说仍然是三个数据源。

---

## 三、解决方案探索与选型

### 方案一：每个市场使用独立数据源

这是我最早采用的办法。

A 股一个接口，美股一个接口，港股再一个接口。

优点很明显：每个数据源针对自己的市场做优化。

但是随着策略增加，我发现维护成本非常高。

例如：

```text
A股获取函数
美股获取函数
港股获取函数
```

三个函数最终都要做：

```text
请求
→ 异常处理
→ 转 DataFrame
→ 字段统一
→ 时间排序
```

大量重复代码开始出现。

### 方案二：在策略层做市场判断

第二种办法看起来简单：

```python
if symbol.endswith(".SH"):
    ...
elif symbol.endswith(".US"):
    ...
elif symbol.endswith(".HK"):
    ...
```

我后来基本放弃了这种方式。

原因很简单：

**策略不应该关心数据源细节。**

策略应该关心的是 close、volume、trade_date 等数据，而不是数据来自哪个 API。

### 方案三：在数据层统一

最后我把市场差异压缩到数据访问层。

以 QuantDash SDK 为例，我可以使用统一形式获取数据：

```python
df = qd.klines.get(
    "600519.SH",
    period="1d",
    count=10,
    to_dataframe=True,
)
```

同样的访问方式可以用于：

```text
A股：600519.SH
美股：AAPL.US
港股：00700.HK
```

这对我的意义不是“调用 API 更方便”，而是让策略层可以保持稳定。

---

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

### 1. 一个统一的跨市场获取函数

这是我比较推荐的封装方式：

```python
import os
from typing import List
from quantdash import QuantDash


def create_client() -> QuantDash:
    api_key = os.getenv("QUANTDASH_API_KEY")
    if not api_key:
        raise RuntimeError(
            "请设置 QUANTDASH_API_KEY 环境变量"
        )

    return QuantDash()


def load_daily_data(
    qd: QuantDash,
    symbol: str,
    count: int = 10,
):
    if not symbol or "." not in symbol:
        raise ValueError(
            "symbol 必须使用类似 600519.SH、AAPL.US、00700.HK 的格式"
        )

    try:
        df = qd.klines.get(
            symbol,
            period="1d",
            count=count,
            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"{symbol} 缺少字段: {sorted(missing)}"
            )

        return df

    except Exception as exc:
        raise RuntimeError(
            f"获取 {symbol} 数据失败: {exc}"
        ) from exc


def main():
    qd = create_client()

    symbols: List[str] = [
        "600519.SH",
        "AAPL.US",
        "00700.HK",
    ]

    for symbol in symbols:
        try:
            df = load_daily_data(qd, symbol, count=5)

            print(f"\n--- {symbol} ---")
            print(
                df[
                    [
                        "symbol",
                        "trade_date",
                        "open",
                        "close",
                        "volume",
                    ]
                ].to_string(index=False)
            )

        except Exception as exc:
            print(exc)


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

我比较看重这个函数里的一个细节：

**数据格式校验放在函数内部，而不是交给策略代码。**

这样策略调用：

```python
df = load_daily_data(qd, "600519.SH")
```

和：

```python
df = load_daily_data(qd, "AAPL.US")
```

得到的是同一种编程体验。

---

### 2. 批量股票数据进入 DataFrame

当股票池扩大以后，我会优先使用批量接口。

```python
import os
from quantdash import QuantDash


def main():
    if not os.getenv("QUANTDASH_API_KEY"):
        raise RuntimeError(
            "请设置 QUANTDASH_API_KEY 环境变量"
        )

    qd = QuantDash()

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

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

        if not dfs:
            print("批量接口没有返回数据")
            return

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

            print(
                f"\n--- {symbol} ---"
            )
            print(
                df[
                    [
                        "trade_date",
                        "open",
                        "close",
                        "volume",
                    ]
                ].to_string(index=False)
            )

    except Exception as exc:
        print(f"批量获取失败: {exc}")


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

这里我通常不会立刻把所有 DataFrame concat 在一起。

原因是不同市场的交易日历可能不同。

如果直接拼接后再处理，很容易产生大量缺失值。

所以我的习惯是：

```text
先按 symbol 验证
→ 再统一字段
→ 最后根据策略需要对齐日期
```

---

### 3. DataFrame 进入统一的收益率计算

如果我的策略只需要价格收益率，那么可以把市场差异进一步隐藏。

```python
import os
import pandas as pd
from quantdash import QuantDash


def calculate_returns(
    qd: QuantDash,
    symbol: str,
    count: int = 30,
) -> pd.DataFrame:
    df = qd.klines.get(
        symbol,
        period="1d",
        count=count,
        to_dataframe=True,
    )

    if df.empty:
        raise ValueError(f"{symbol} 返回空数据")

    result = df.copy()
    result["trade_date"] = pd.to_datetime(
        result["trade_date"]
    )

    result = result.sort_values("trade_date")
    result["return_1d"] = (
        result["close"].pct_change()
    )

    return result


def main():
    if not os.getenv("QUANTDASH_API_KEY"):
        raise RuntimeError(
            "请设置 QUANTDASH_API_KEY 环境变量"
        )

    qd = QuantDash()

    for symbol in [
        "600519.SH",
        "AAPL.US",
        "00700.HK",
    ]:
        try:
            df = calculate_returns(
                qd,
                symbol,
                count=30,
            )

            print(
                f"\n{symbol} 最近数据:"
            )
            print(
                df[
                    [
                        "trade_date",
                        "close",
                        "return_1d",
                    ]
                ].tail().to_string(index=False)
            )

        except Exception as exc:
            print(
                f"{symbol} 计算失败: {exc}"
            )


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

这里体现的是我比较推崇的一种设计：

> **让数据源适配市场，让 Pandas 承担分析。**

而不是让每个策略重新理解市场。

---

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

### 避坑一：不要默认所有市场交易日一致

这是跨市场策略非常容易忽略的问题。

例如 A 股休市时，美股可能正常交易。

如果我简单使用日期 inner join，很可能直接丢失大量记录。

所以我在构建跨市场 DataFrame 时，会先明确策略到底需要：

- 同一自然日；
- 各市场最近交易日；
- 还是滚动窗口。

不同答案对应不同的数据对齐方式。

---

### 避坑二：不要只用股票代码判断标的

我现在会强制使用带市场后缀的代码。

例如：

```text
600519.SH
000001.SZ
AAPL.US
00700.HK
```

这样在 DataFrame 中保留 symbol 字段时，后续合并数据会更加明确。

---

### 避坑三：复权必须成为数据层参数

如果我要做收益率回测，通常会显式指定复权方式，而不是完全依赖默认值。

例如：

```python
import os
from quantdash import QuantDash


def main():
    if not os.getenv("QUANTDASH_API_KEY"):
        raise RuntimeError(
            "请设置 QUANTDASH_API_KEY 环境变量"
        )

    qd = QuantDash()

    try:
        df = qd.klines.get(
            "600519.SH",
            period="1d",
            count=20,
            adjust="forward",
            to_dataframe=True,
        )

        if df.empty:
            print("没有数据")
            return

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

    except Exception as exc:
        print(f"获取失败: {exc}")


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

比例复权更适合收益率计算，而差值复权更适合观察绝对价差，这是我在策略研究中会特别区分的两个用途。

---

## 六、总结与思考

我后来发现，“A股美股港股统一数据接口”真正难的从来不是把三个市场的数据分别拿回来，而是让它们在进入策略以后看起来像同一种数据。

因此，我现在会尽可能把市场代码、API 差异、异常处理和 DataFrame 转换全部封装在数据层。

QuantDash 在我的数据层中承担的是获取和 DataFrame 输出这一部分工作；至于交易日对齐、因子逻辑、异常数据处理，我仍然会在自己的研究框架里完成。

如果以后更换数据源，我只需要重新实现数据访问层，而不需要把整个策略重写一遍。这是我认为“数据工程”和“写一个能跑的量化脚本”之间非常重要的区别。

### Q&A

**Q1：为什么跨市场策略一定要统一 symbol 格式？**

A：因为股票代码本身不是完整身份。加入交易所或市场后缀后，可以降低不同市场之间代码冲突和合并错误的概率。

**Q2：不同市场的数据能不能直接 concat？**

A：技术上可以，但不代表统计意义正确。首先应该处理交易日历差异，再决定使用 inner、outer 或其他时间对齐方式。

当量化策略从单一 A 股扩展到美股、港股时，真正棘手的问题往往不是 API 调用，而是不同市场的数据结构和代码规则如何统一。本文从我实际搭建跨市场量化数据层的经历出发，介绍如何利用统一的股票代码规则和 DataFrame 输出，把不同市场的数据接入同一套 Pandas 分析流程。

Python 获取股票数据后如何直接转 DataFrame？一套代码统一 A 股、美股和港股

编程语言

跨市场量化策略的核心挑战在于统一不同市场的股票代码与DataFrame结构，确保数据能无缝接入同一套Pandas分析流程。通过标准化代码格式（如600519.SH、AAPL.US）和数据层封装，实现A股、美股、港股的高效整合，提升策略开发与维护效率。

数据接口

数据结构

数据处理

代码解释

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

社区新版编辑器体验调研

Python 获取股票数据后如何直接转 DataFrame？一套代码统一 A 股、美股和港股-腾讯云开发者社区-腾讯云

Python 获取股票数据后如何直接转 DataFrame？一套代码统一 A 股、美股和港股

Python 获取股票数据后如何直接转 DataFrame？一套代码统一 A 股、美股和港股

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐