### 概述

我以前在做跨市场量化选股时发现，行情接入最难维护的部分，经常不是因子公式，而是标的代码、字段口径和交易时间规则。A 股、美股和港股的行情如果分别接入、分别转换，策略层很容易积累大量市场判断。本文从统一数据管道的角度分析全市场实时行情的必要性，并通过 Python 示例展示如何用标准化标的代码、批量行情接口和数据校验降低跨市场选股的维护成本。

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

最早做多市场选股时，我没有一开始就搭建统一的数据层。

当时觉得市场不多，每个市场单独写一个行情获取模块就可以了。

A 股一套代码，美股一套代码，港股再加一套。

单独看，这样做并不复杂。

问题是，策略不断增加之后，市场差异开始向上层蔓延。

某个因子需要读取收盘价，我得先判断当前标的属于哪个市场，再决定调用哪个接口、解析哪个字段。

某个策略需要过滤成交额不足的股票，又要考虑不同接口对成交额字段的定义是否一致。

到了批量选股阶段，代码里开始出现越来越多的市场分支。

更难处理的是标的身份。

例如，程序只用数字代码标识股票时，不同市场可能存在形式相同或容易混淆的代码。即使当前股票池没有发生冲突，未来加入新的市场后，也可能需要重新整理历史数据、缓存键和任务配置。

当时我以为，只要维护一份股票代码映射表，就可以解决问题。

后来发现，代码映射只是第一层。

数据时间、交易日历、复权方式和字段含义仍然需要单独管理。如果这些差异一直留在策略代码里，那么每增加一个市场，就会增加一批需要测试的分支。

这也是我后来开始把行情接入和策略计算分开的原因。

## 二、问题根因分析

### 1. 标的代码是数据身份的一部分，不只是一个字符串

跨市场系统里，单纯的数字代码不够稳妥。

同一个字符串可能在不同市场中代表不同标的，也可能因为前导零丢失而变成另一个形式。

例如，港股代码 **00700.HK** 如果被转换成整数，就可能丢失前导零。

对于 A 股、美股和港股，也不能依赖一套没有市场后缀的代码规则。

我倾向于使用统一格式表示标的：

- A 股：**600519.SH**、**000001.SZ**。
- 北交所股票：**920047.BJ**。
- 美股：**AAPL.US**。
- 港股：**00700.HK**。

这样做的价值，不是字符串变得更漂亮，而是数据管道有了统一的标的身份。

缓存键、日志、批量请求参数、行情记录和因子结果，都可以使用同一套标识。

但也需要强调，代码后缀并不能替代完整的标的主数据。

如果系统需要处理退市、代码变更、证券类型变化或者历史代码复用，还应维护稳定的内部标的标识，并保留代码有效期和市场元数据。

统一代码规范解决的是接入层的基本一致性，不是所有证券主数据问题。

### 2. 全市场实时行情需要统一字段口径，但不能抹平市场差异

跨市场选股时，常见的做法是把各市场的行情全部转换成一个 DataFrame。

这本身没有问题。

问题出在转换完成之后，系统可能误以为字段含义也已经完全一致。

例如，价格字段都叫 **close**，并不代表它们都适合直接比较。

A 股的价格可能经过前复权处理，美股使用未复权价格，港股又使用另一套默认处理方式。即使字段名称相同，数值口径仍然可能不同。

再比如，成交额、成交量和报价时间，虽然都可以转换成数值或时间类型，但背后的市场定义可能存在差异。

如果直接将它们送进统一因子公式，代码虽然简洁了，数据语义却可能不一致。

所以，我对标准化的理解分成两层：

第一层是统一结构，包括标的代码、字段类型、数据表结构和异常状态。

第二层是保留市场语义，包括交易日历、币种、复权方式、交易状态以及数据可用时间。

统一结构不意味着所有市场可以不加区分地进行比较。

对于需要跨市场横截面排序的策略，还需要明确币种转换、时区对齐和交易日历规则。否则，同一张表里的数值未必具有直接可比性。

### 3. 数据接入层和策略层混在一起，会让每次扩展都变成一次重构

当策略数量较少时，直接在策略函数中请求行情看起来很方便。

但随着市场和策略增加，这种方式会逐渐产生维护问题。

例如：

- 策略 A 自己维护一份标的代码映射。
- 策略 B 自己实现行情重试。
- 策略 C 使用另一套字段转换规则。
- 策略 D 在本地单独计算复权价格。

同一个问题会在不同策略中被重复解决，而且解决方式可能并不一致。

一旦行情接口、标的代码或字段定义发生变化，维护人员就需要检查所有使用方。

我后来把系统拆成了几个职责明确的部分：

1. **标的主数据层**：维护标的身份、市场和基本属性。
2. **行情接入层**：负责请求行情、转换格式和记录请求状态。
3. **数据质量层**：检查字段、缺失标的、重复记录和时间口径。
4. **因子计算层**：只处理通过校验的输入数据。
5. **策略决策层**：负责股票筛选、排序和风险约束。

这种拆分不会自动解决所有问题，但可以减少策略代码对底层行情接口的依赖。

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

对于多市场量化系统，我不太赞成把所有问题都归结为“自己写还是使用 SDK”。

真正需要比较的是：哪些通用能力值得集中维护，哪些市场规则必须由业务明确处理。

| 评估维度 | 自建网络抓取 / 多市场适配脚本 | 标准化数据层（QuantDash SDK） |
| :--- | :--- | :--- |
| **标的代码规范** | 各市场分别处理，映射表容易分散 | 可统一使用 .SH/.SZ/.HK/.US/.BJ 后缀 |
| **批量行情接入** | 需要自行处理请求分批、并发和限流 | 可使用统一 batch 接口组织批量 K 线请求 |
| **数据结构** | 不同市场的 JSON 结构需要分别解析 | 支持通过 to_dataframe=True 转换为 DataFrame |
| **复权参数** | 需要自行维护各类复权逻辑 | 可通过统一接口指定复权参数 |
| **实时行情接入** | 不同市场往往维护不同调用路径 | 可以通过统一接口组织行情请求 |
| **交易日历** | 需要自行建立并维护各市场规则 | 仍需由业务层管理交易日历与跨市场对齐 |
| **币种与单位** | 容易在策略代码中混用不同口径 | 标准化接入后仍需明确币种和单位规则 |
| **策略扩展** | 新市场通常需要修改多处逻辑 | 可将市场适配集中在数据接入与校验层 |

这里需要避免一个误区：统一 SDK 并不会自动消除跨市场的所有差异。

它可以帮助统一调用方式和基础数据结构，但市场交易时间、货币单位、证券属性以及策略的比较规则仍然需要显式定义。

对于日频策略，交易日历对齐尤其重要。美股休市而 A 股开市时，不能为了方便，直接拿美股上一个交易日的收盘价与 A 股当天的收盘价做同步截面比较，然后把它们当成同一时刻的数据。

对于盘中策略，还要考虑各市场是否处于交易时段，以及行情是否满足当前决策时点的时效要求。

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

下面用一个简化的多市场数据接入任务说明如何组织代码。

示例选择 A 股、港股和美股的三个标的，统一通过行情接口获取数据，并进行基础校验。

为了避免混淆，示例中的标的列表只是说明接口组织方式，不代表它们在当前时点都处于交易状态。

安装依赖：

```python
pip install quantdash
```

完整代码：

```python
import os
import sys
import logging
from typing import Dict, List

import pandas as pd
from quantdash import QuantDash


logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)
logger = logging.getLogger(__name__)


MARKET_UNIVERSE: Dict[str, List[str]] = {
    "CN": [
        "600519.SH",
        "000001.SZ",
    ],
    "HK": [
        "00700.HK",
    ],
    "US": [
        "AAPL.US",
    ],
}

REQUIRED_COLUMNS = {
    "symbol",
    "last_price",
}


def flatten_universe(
    universe: Dict[str, List[str]],
) -> List[str]:
    symbols = []

    for market, market_symbols in universe.items():
        if not market_symbols:
            raise ValueError(
                f"市场 {market} 的股票池为空"
            )

        symbols.extend(market_symbols)

    if len(symbols) != len(set(symbols)):
        raise ValueError("跨市场股票池存在重复标的")

    return symbols


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

    symbols = flatten_universe(MARKET_UNIVERSE)

    qd = QuantDash(api_key=api_key)

    try:
        df = qd.quotes.get(
            symbols=symbols,
            to_dataframe=True,
        )
    except Exception:
        logger.exception("跨市场行情请求失败")
        raise

    if df is None or df.empty:
        raise RuntimeError("行情接口返回空数据")

    missing_columns = REQUIRED_COLUMNS - set(df.columns)
    if missing_columns:
        raise RuntimeError(
            f"缺少必要行情字段: {sorted(missing_columns)}"
        )

    df = df.copy()
    df["symbol"] = df["symbol"].astype(str)

    expected_symbols = set(symbols)
    returned_symbols = set(df["symbol"])

    missing_symbols = sorted(
        expected_symbols - returned_symbols
    )
    unexpected_symbols = sorted(
        returned_symbols - expected_symbols
    )

    if missing_symbols:
        logger.warning(
            "以下标的未返回行情: %s",
            missing_symbols,
        )

    if unexpected_symbols:
        logger.warning(
            "以下标的不属于预期股票池: %s",
            unexpected_symbols,
        )

    if df["symbol"].duplicated().any():
        duplicated = df.loc[
            df["symbol"].duplicated(keep=False),
            "symbol",
        ].tolist()

        raise RuntimeError(
            f"发现重复行情记录: {duplicated}"
        )

    df["last_price"] = pd.to_numeric(
        df["last_price"],
        errors="coerce",
    )

    invalid_price = (
        df["last_price"].isna()
        | (df["last_price"] <= 0)
    )

    if invalid_price.any():
        logger.warning(
            "存在无效价格记录:\n%s",
            df.loc[
                invalid_price,
                ["symbol", "last_price"],
            ].to_string(index=False),
        )

    # 市场分类应使用明确的主数据规则，
    # 不建议在整个策略系统中重复解析代码后缀。
    def classify_market(symbol: str) -> str:
        suffix_to_market = {
            ".SH": "CN",
            ".SZ": "CN",
            ".BJ": "CN",
            ".HK": "HK",
            ".US": "US",
        }

        for suffix, market in suffix_to_market.items():
            if symbol.endswith(suffix):
                return market

        raise ValueError(
            f"无法识别标的所属市场: {symbol}"
        )

    df["market"] = df["symbol"].map(
        classify_market
    )

    # 按市场分组进行数据质量统计。
    quality_summary = (
        df.assign(invalid_price=invalid_price)
        .groupby("market")
        .agg(
            symbol_count=("symbol", "nunique"),
            row_count=("symbol", "size"),
            invalid_price_count=(
                "invalid_price",
                "sum",
            ),
        )
        .reset_index()
    )

    logger.info(
        "行情质量统计:\n%s",
        quality_summary.to_string(index=False),
    )

    # 不同市场的数据可以统一存储，
    # 但不应直接忽略交易日历、币种和时间差异。
    valid_df = df.loc[~invalid_price].copy()

    if valid_df.empty:
        raise RuntimeError("没有可用行情记录")

    # 后续进入市场级校验、时区处理和因子计算。
    # 在完成相应处理前，不直接生成跨市场排名。
    print(
        valid_df[
            ["symbol", "market", "last_price"]
        ].to_string(index=False)
    )


if __name__ == "__main__":
    try:
        main()
    except Exception:
        logger.exception("跨市场行情任务失败")
        sys.exit(1)
```

这份代码有意把重点放在接入层，而不是直接计算跨市场综合排名。

原因很简单：拿到价格，不等于这些价格已经具备直接比较的条件。

在生产环境中，我还会补上以下几个模块。

**第一，标的主数据校验。**

代码后缀只能作为基础分类信息。更完整的系统应通过证券主数据确认标的是否存在、是否处于有效状态，以及当前证券类型是否符合策略要求。

**第二，行情时间与交易日历校验。**

需要明确每个市场的交易时段和数据可用时间。不同市场的节假日并不完全一致，不能只用同一个日期字符串就认为行情已经对齐。

**第三，币种与计量单位处理。**

如果因子涉及价格、成交额或市值，需要明确各市场的数据单位。若策略要比较不同市场的成交额或市值，还需要采用明确的汇率和估值时点规则。

**第四，复权口径管理。**

QuantDash 的 K 线接口支持通过 **adjust** 指定复权方式，例如 **forward**、**backward** 和 **none**。但不同策略不能随意混用复权口径。

如果使用前复权数据计算历史收益率，应明确复权数据的用途。如果涉及实盘下单价格、成交回报和真实持仓成本，则应与实际交易价格体系保持清晰的边界。

另外，示例采用的是实时行情接口。如果策略需要历史 K 线，可以使用对应的 K 线接口；如果需要跨市场历史回测，还应额外管理交易日历和历史数据的时间对齐。

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

### 避坑点 1：统一标的编码之后，仍然要维护证券主数据

使用 **600519.SH** 或 **00700.HK** 这样的代码形式，可以减少很多低级错误。

但不能只靠字符串后缀处理所有证券身份问题。

代码变更、退市、证券类型变化以及历史标的映射，都应该由主数据层负责。

如果策略直接从代码字符串推断所有业务属性，后面还是会积累大量隐式规则。

### 避坑点 2：不要把不同市场的最新行情直接当成同一时点的数据

这是跨市场选股最容易出现的时间问题。

A 股和美股可能处于完全不同的交易时段。

港股与其他市场的节假日安排也可能不同。

即使每个市场都返回了最新行情，也不代表这些行情能够组成一个严格同步的截面。

如果策略需要进行跨市场比较，就必须定义决策时点、允许的数据延迟、交易日历规则以及休市期间数据的处理方式。

对于某些策略，使用上一个有效交易日的数据是合理的；对于另一些策略，这样做可能完全不符合信号定义。

关键是规则必须明确，而不是由数据接口的默认行为决定。

### 避坑点 3：统一数据结构，不等于统一所有字段的业务含义

把所有行情转成 DataFrame 只是第一步。

如果不同市场的成交额单位、复权口径或时间字段语义不一致，简单拼接仍然可能造成计算错误。

我的做法是先统一基础结构，再在市场适配层明确处理单位、币种、复权方式和交易时间。

当策略需要使用某个字段时，应该能够追溯这个字段来自哪个市场、采用什么口径、对应哪个时间点。

这比单纯让所有接口返回相同的列名更重要。

## 六、写在最后

量化选股需要全市场实时行情，不仅因为策略需要更大的候选范围，还因为市场覆盖、标的身份和数据时间共同决定了选股输入是否可靠。对于跨市场系统来说，统一行情接入层的价值，是把重复的网络请求、字段转换和基础校验集中管理，同时保留每个市场真正需要的业务规则。

我不太建议一开始就把数据架构设计得过于复杂。先把标的代码统一起来，把行情接入和因子计算分开，再补齐数据质量检查、交易日历和时间口径，通常就能解决一大批维护问题。真正重要的不是代码看起来有多统一，而是系统在增加市场、增加策略或者遇到行情异常时，仍然能够清楚地知道自己拿到了什么数据，以及这些数据能不能用于当前决策。

参考
- QuantDash Python SDK 文档：https://docs.quantdash.net/zh-Hans/sdk/python-quickstart

我以前在做跨市场量化选股时发现，行情接入最难维护的部分，经常不是因子公式，而是标的代码、字段口径和交易时间规则。A 股、美股和港股的行情如果分别接入、分别转换，策略层很容易积累大量市场判断。本文从统一数据管道的角度分析全市场实时行情的必要性，并通过 Python 示例展示如何用标准化标的代码、批量行情接口和数据校验降低跨市场选股的维护成本。

跨市场量化选股为什么不能只维护一份股票列表？从代码映射到统一行情数据层

编程语言

跨市场量化选股时，行情接入的维护成本常集中在标的代码、字段口径与交易时间规则上。本文提出通过统一数据管道，以标准化代码、批量接口及数据校验，降低A股、港股、美股跨市场选股的维护复杂度，并提供Python实战示例，重点解决代码冲突、字段语义差异与时间对齐问题，确保策略输入数据的可靠性与一致性。

数据管道

数据架构

数据接口

数据结构

代码解释

Python

JSON

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

社区新版编辑器体验调研

跨市场量化选股为什么不能只维护一份股票列表？从代码映射到统一行情数据层-腾讯云开发者社区-腾讯云

跨市场量化选股为什么不能只维护一份股票列表？从代码映射到统一行情数据层

跨市场量化选股为什么不能只维护一份股票列表？从代码映射到统一行情数据层

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐