一、问题场景：我第一次做多市场策略时，代码很快失控

我最早做量化策略的时候，主要关注 A 股。

后来策略需要增加美股和港股。

但真正接进去以后，我发现问题远比想象中复杂。

不同数据接口的代码格式不同。

有的使用交易所后缀，有的使用前缀，有的直接使用数字代码。

如果策略里面直接写这些字符串，后面每增加一个市场，都需要修改大量代码。

多市场量化系统最先应该统一的不是策略，而是标的标识体系。

1. 标的代码是最容易被忽视的基础设施

600519.SH
000001.SZ
920047.BJ
AAPL.US
00700.HK


QuantDash 的标的代码就是采用这种市场后缀方式。

这对我最大的帮助不是少写几行代码，而是让策略层不需要关心不同市场的原始代码格式。

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


从策略的角度来看，它们都只是 symbol。

多市场策略另一个常见坑是周期。

但一旦进入分钟级数据，交易时间、K 线频率和数据时间字段就会变得非常重要。

我现在通常会把数据模型设计成：

symbol
trade_date / trade_time
open
high
low
close
volume


3. 单只股票查询和股票池查询应该分开

如果只研究一只股票，我倾向于使用单标的接口。

如果策略每天需要扫描整个市场，则更适合股票池或者批量查询。

import os

from quantdash import QuantDash


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

    if not api_key:
        print("请设置 QUANTDASH_API_KEY")
        return

    qd = QuantDash(api_key=api_key)

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

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

        if not dfs:
            print("没有返回任何数据")
            return

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

            if df.empty:
                print("数据为空")
                continue

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

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


if __name__ == "__main__":
    main()


这时候我的策略层只需要拿到一个字典：

而不需要知道底层请求是怎么完成的。

我之前也考虑过自己写一层统一适配。

A股数据源
美股数据源
港股数据源
       ↓
SymbolAdapter
       ↓
统一 DataFrame


但实际开发以后会发现，真正需要维护的不只是代码转换。

如果项目规模不大，我不太喜欢把大量时间花在重复的数据适配工作上。

后来我在项目中使用 QuantDash 后，一个比较明显的变化是，A 股、美股和港股可以采用统一的标的代码形式，数据又可以直接返回 Pandas。

策略
 ↓
统一 DataFrame
 ↓
QuantDash
 ↓
数据接口


四、具体实现：建立一个多市场数据读取器

我现在比较推荐把数据访问封装成一个简单类。

import os

import pandas as pd
from quantdash import QuantDash


class MarketData:
    def __init__(self) -> None:
        api_key = os.getenv("QUANTDASH_API_KEY")

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

        self.qd = QuantDash(api_key=api_key)

    def daily(
        self,
        symbol: str,
        count: int = 60,
        adjust: str = "forward",
    ) -> pd.DataFrame:
        try:
            df = self.qd.klines.get(
                symbol,
                period="1d",
                count=count,
                adjust=adjust,
                to_dataframe=True,
            )
        except Exception as exc:
            raise RuntimeError(
                f"{symbol} 日线请求失败: {exc}"
            ) from exc

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

        return df

    def realtime(
        self,
        symbols: list[str],
    ) -> pd.DataFrame:
        try:
            df = self.qd.quotes.get(
                symbols=symbols,
                to_dataframe=True,
            )
        except Exception as exc:
            raise RuntimeError(
                f"实时行情请求失败: {exc}"
            ) from exc

        if df.empty:
            raise ValueError(
                "实时行情返回为空"
            )

        return df


def main() -> None:
    try:
        data = MarketData()

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

        for symbol in symbols:
            try:
                df = data.daily(
                    symbol,
                    count=10,
                )

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

            except Exception as exc:
                print(
                    f"{symbol} 处理失败: {exc}"
                )

    except Exception as exc:
        print(f"程序初始化失败: {exc}")


if __name__ == "__main__":
    main()


这段代码对我来说有一个很重要的价值：

策略代码不需要判断这是 A 股、美股还是港股。

如果策略需要扫描整个 A 股市场，我会进一步使用标的池。

import os

from quantdash import QuantDash


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

    if not api_key:
        print("请设置 QUANTDASH_API_KEY")
        return

    qd = QuantDash(api_key=api_key)

    try:
        df = qd.quotes.get(
            universes=["CN_Stock"],
            to_dataframe=True,
        )

        if df.empty:
            print("股票池没有返回数据")
            return

        print(f"股票数量: {len(df)}")

        columns = [
            "symbol",
            "region",
            "last_price",
            "prev_close",
            "volume",
        ]

        available = [
            column
            for column in columns
            if column in df.columns
        ]

        print(
            df[available]
            .head(20)
            .to_string(index=False)
        )

    except Exception as exc:
        print(f"股票池查询失败: {exc}")


if __name__ == "__main__":
    main()


如果我要做一个简单的横截面选股，这类接口比一只股票一只股票请求更加适合。

多市场策略还有一个经常被忽略的问题：

股票代码不是完整的标的信息。

我通常还需要知道交易所、证券类型、上市日期等信息。

import os

from quantdash import QuantDash


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

    if not api_key:
        print("请设置 QUANTDASH_API_KEY")
        return

    qd = QuantDash(api_key=api_key)

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

    try:
        instruments = qd.instruments.get(symbols)

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

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

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


if __name__ == "__main__":
    main()


我会把这些元数据放在数据层，而不是让策略自己维护一份股票信息表。

避坑 1：不要在策略里写市场判断

我不推荐出现大量这样的逻辑：

如果是 A 股……
如果是美股……
如果是港股……


市场差异应该尽量被封装在数据层。

避坑 2：批量接口不是越多越好

批量获取可以减少请求次数，但也要考虑：

我的习惯是把批量接口放在数据准备阶段，而不是让策略运行时一次性请求所有东西。

避坑 3：分钟线一定要单独做数据质量测试

我曾经认为日线正常，分钟线自然也不会有问题。

import os

from quantdash import QuantDash


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

    if not api_key:
        print("请设置 QUANTDASH_API_KEY")
        return

    qd = QuantDash(api_key=api_key)

    try:
        df = qd.klines.get(
            "600519.SH",
            period="5m",
            count=100,
            to_dataframe=True,
        )

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

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

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


if __name__ == "__main__":
    main()


我现在越来越倾向于把“多市场数据”理解成一个数据建模问题，而不仅仅是一个 API 调用问题。

如果 A 股、美股、港股分别使用不同的数据模型，那么策略层迟早会充满各种 if/else。

反过来，如果在数据层统一 symbol、时间、OHLCV 和元数据，策略层就可以真正做到市场无关。

我目前在部分项目中使用 QuantDash 来承担这层数据访问工作。对我而言，它最大的价值并不是“接口数量多”，而是让我能够把数据访问与策略逻辑隔离开。

对于量化系统来说，我更看重的始终是这一点：

数据层负责复杂，策略层负责简单。

Q1：A股、美股、港股可以直接使用同一套策略代码吗？

A：数据模型统一后，大部分计算逻辑可以复用，但交易日历、交易时段、市场规则等仍然需要在交易执行层单独处理。

Q2：为什么推荐统一使用带市场后缀的代码？

A：因为单纯使用证券代码可能产生歧义。统一格式可以让数据访问层更明确地识别市场和交易所。

Q3：多市场策略应该如何选择数据源？

A：我建议优先看统一性，而不是只看市场数量。至少应该比较代码体系、字段定义、历史数据、分钟数据、实时行情、复权方式和批量能力。

A股、美股、港股数据怎么统一？我踩过代码格式和数据口径的坑

编程语言

后端

建立统一Python量化数据访问层，统一A股、美股、港股标的代码与数据模型，封装日K、实时行情及批量获取接口，避免策略层市场判断，提升多市场策略开发效率。

数据建模

数据接口

Python

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

A股、美股、港股数据怎么统一？我踩过代码格式和数据口径的坑

A股、美股、港股数据怎么统一？我踩过代码格式和数据口径的坑

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐