### 概述

我以前在做跨市场选股时发现，最初为 A 股编写的数据处理逻辑，一旦扩展到港股和美股，就会出现代码映射、字段类型、交易时段和行情时间不一致的问题。本文从一次跨市场选股重构出发，讨论如何通过统一标的身份、标准化行情字段和独立的市场规则层，构建能够持续扩展的实时选股数据层。

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

最开始做选股时，我只处理 A 股。

当时的数据接口比较简单，股票代码、行情字段和交易时间都围绕一个市场设计。

策略需要的输入也很明确：

股票代码、最新价格、成交量、成交额，再加上前几个时间片的行情数据。

后来，业务需要扩展到港股和美股。

这时候，我才发现原先那套数据层并没有想象中那么通用。

A 股股票可能使用 **600519.SH** 这样的标识，港股则可能使用 **00700.HK**，美股则可能使用 **AAPL.US**。

如果继续让每个策略直接处理原始代码，就会出现大量市场判断。

有些地方根据代码后缀判断市场，有些地方根据数字长度判断，有些地方则依赖自己维护的映射表。

最初只有一个策略时，这些判断还算容易管理。

但当行情、因子、回测和实盘监控都开始使用同一批标的时，问题就逐渐暴露出来。

我曾经见过同一只股票在两个模块中使用不同的代码表示。

行情模块把它识别为一个标的，因子模块却使用另一种格式查找历史数据。

程序没有崩溃，只是连接历史行情时找不到对应记录。

还有一次，行情数据本身没有问题，但不同市场的数据时间戳没有统一处理。

A 股和美股的行情记录被放进同一个 DataFrame 后，程序按照时间排序，随后直接计算横截面排名。

结果看起来很合理。

但如果不区分市场交易时段，这个排名就可能混合了不同时间语义的数据。

我当时以为，只要把代码统一一下，再把字段名称改成一样，问题就解决了。

实际做完才发现，统一代码和统一字段只是第一步。

真正需要统一的，是数据进入策略之前的处理边界。

例如：

- 标的身份应该由什么规则确定？
- 同一字段在不同市场中的含义是否完全一致？
- 当前行情与上一交易日行情应该如何区分？
- 不同市场的非交易时段，应该返回什么状态？
- 跨市场因子计算是否可以直接拼接所有行情？

如果这些问题没有明确约定，数据层就会逐渐变成一堆市场判断语句。

每增加一个市场，维护成本都会继续上升。

## 二、问题根因分析

### 1. 根因一：标的代码只是字符串，但系统把它当成了完整的标的身份

很多系统最初都用字符串保存股票代码。

这没有问题。

问题出在不同模块对字符串的解释方式不一致。

例如，**700.HK** 和 **00700.HK** 可能被不同模块当作两个独立标识。

如果历史行情使用一种格式，实时行情使用另一种格式，两个数据集就无法直接关联。

更麻烦的是，一些内部映射逻辑会自动删除前导零，或者通过数字转换把字符串变成数值。

这样处理之后，原始代码的身份信息就可能丢失。

我后来把标的代码规范统一成：

- A 股：**600519.SH**、**000001.SZ**、**920047.BJ**
- 港股：**00700.HK**
- 美股：**AAPL.US**

这里的关键不在于后缀本身，而在于系统中只有一个标准化入口。

所有外部代码都需要先经过规范化处理，才能进入行情存储、因子计算和策略模块。

同时，我不再允许策略自己猜测代码属于哪个市场。

市场身份应该由标的主数据层负责确认。

如果某个标的无法匹配到已知的市场和证券身份，数据层应该明确拒绝，而不是自动猜测。

### 2. 根因二：字段名称相同，不代表数据语义完全一致

跨市场数据处理中，还有一个容易被忽略的问题。

字段统一了，含义却不一定完全统一。

例如，**close** 都可以表示价格字段，但不同接口返回的价格可能对应不同的时间定义。

有的接口返回最近一笔成交价格，有的返回某个时间片的收盘价，有的则返回经过特定处理的行情字段。

如果把这些数据直接拼在一起，字段名称虽然一致，实际语义却可能不同。

成交量和成交额也需要注意。

不同市场的计量单位、证券交易规则和行情更新频率可能存在差异。

如果不检查这些定义，就直接把原始数值输入统一因子，结果未必具有可比性。

因此，我现在会区分两个层次。

第一层是字段规范化。

例如，将经过确认的最新价格统一映射到 **price**，将经过确认的行情时间统一映射到 **event_time**。

第二层是数据语义确认。

每个字段都需要知道它表示什么、采用什么单位、对应什么时间，以及是否经过复权处理。

统一字段名称，不等于自动统一字段含义。

### 3. 根因三：市场规则与策略逻辑混在一起

最初的代码经常这样写：

如果是 A 股，就使用某种行情处理方式。

如果是港股，就使用另一种处理方式。

如果是美股，再增加一组判断。

当时看起来很直接。

但市场判断会逐渐渗透到因子计算、交易时间处理、缺失值填充和选股排序中。

结果就是，策略本身越来越依赖市场细节。

后来，如果想调整一个通用动量因子，就必须先检查它是否包含市场判断。

如果想增加一个新市场，又需要修改多个模块。

这种结构最难维护的地方在于，代码虽然可以运行，但没有清晰的责任边界。

我更倾向于把系统拆成四层：

1. **标的主数据层**：确认标的身份、市场和标准代码。
2. **行情接入层**：负责获取行情并统一返回结构。
3. **市场规则层**：负责交易日历、时区和市场特定的行情语义。
4. **策略计算层**：只处理已经通过标准化和质量检查的数据。

这样一来，策略不需要知道行情是从哪个市场接口获取的。

它只需要知道输入数据满足什么契约。

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

我最终没有继续给每个市场单独增加一套行情适配代码，而是把接入方式统一起来。

数据拉取可以通过 QuantDash SDK 完成，标的身份和市场规则则由自己的业务层维护。

这里要注意：统一 SDK 并不意味着所有市场的交易规则都相同。数据接入可以统一，交易日历、时区和市场语义仍然需要单独管理。

| 评估维度 | 自建网络抓取 / 多市场适配脚本 | 标准化数据层（QuantDash SDK） |
| :--- | :--- | :--- |
| **标的代码规范** | 各市场单独维护代码转换和映射表 | 使用 .SH、.SZ、.HK、.US、.BJ 等统一后缀 |
| **行情接入** | 每个市场实现独立请求与解析逻辑 | 通过统一的行情接口获取数据 |
| **字段转换** | 每个接口单独维护 JSON 到 Pandas 的映射 | 通过统一的数据获取方式减少重复转换 |
| **交易时间处理** | 容易把市场判断散落在各个策略中 | 将交易日历和时区规则保留在独立的市场规则层 |
| **历史行情获取** | 不同市场维护不同的历史数据函数 | 通过统一 K 线接口组织历史数据请求 |
| **扩展新市场** | 需要修改多个策略模块 | 主要扩展标的主数据和市场规则适配层 |
| **维护方式** | 接口与策略高度耦合 | 接入层、规则层与策略层分离 |

对我来说，最重要的收益不是少写几个接口函数，而是能够控制市场特定逻辑的传播范围。

行情接入负责把数据取回来。

市场规则层负责解释数据的时间和交易含义。

策略层只消费符合约定的数据。

这三个职责不能混在一起。

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

下面通过一个简化的跨市场行情获取示例，展示如何在接入层统一标的格式，并把市场信息与行情字段分开处理。

这个示例主要覆盖：

- 环境变量读取。
- 多市场标的配置。
- 批量行情获取。
- 必需字段校验。
- 标的身份检查。
- 统一字段输出。

需要强调的是，代码不会尝试自行推断所有市场的代码规则，也不会假设不同市场可以直接按同一交易时间进行比较。

安装依赖：

**pip install quantdash**

```python
import os
import sys
import logging
from typing import Any

import pandas as pd
from quantdash import QuantDash


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


INSTRUMENTS = {
    "600519.SH": {"market": "CN", "timezone": "Asia/Shanghai"},
    "000001.SZ": {"market": "CN", "timezone": "Asia/Shanghai"},
    "00700.HK": {"market": "HK", "timezone": "Asia/Hong_Kong"},
    "AAPL.US": {"market": "US", "timezone": "America/New_York"},
}

REQUIRED_COLUMNS = {"symbol", "close", "volume"}


def normalize_symbol(symbol: Any) -> str:
    """标准化输入形式，不擅自猜测缺失的交易所后缀。"""
    if not isinstance(symbol, str):
        raise ValueError(f"标的代码必须是字符串: {symbol!r}")

    value = symbol.strip().upper()

    if not value or "." not in value:
        raise ValueError(f"标的代码缺少交易所后缀: {symbol!r}")

    suffix = value.rsplit(".", 1)[1]
    allowed_suffixes = {"SH", "SZ", "BJ", "HK", "US"}

    if suffix not in allowed_suffixes:
        raise ValueError(f"不支持的交易所后缀: {symbol!r}")

    return value


def build_instrument_master() -> pd.DataFrame:
    """构建最小化标的主数据表。"""
    rows = []

    for raw_symbol, metadata in INSTRUMENTS.items():
        symbol = normalize_symbol(raw_symbol)

        rows.append({
            "symbol": symbol,
            "market": metadata["market"],
            "timezone": metadata["timezone"],
        })

    master = pd.DataFrame(rows)

    if master["symbol"].duplicated().any():
        raise RuntimeError("标的主数据中存在重复代码")

    return master


def normalize_quotes(
    df: pd.DataFrame,
    expected_symbols: list[str],
    master: pd.DataFrame,
) -> pd.DataFrame:
    """将行情映射到统一字段，并检查标的覆盖情况。"""
    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)}"
        )

    result = df.copy()

    result["symbol"] = result["symbol"].map(normalize_symbol)

    if result["symbol"].duplicated().any():
        raise RuntimeError("行情结果包含重复标的")

    expected = set(expected_symbols)
    actual = set(result["symbol"])

    missing = expected - actual
    unexpected = actual - expected

    if missing or unexpected:
        raise RuntimeError(
            f"标的集合不一致，缺失={sorted(missing)}，"
            f"非预期={sorted(unexpected)}"
        )

    result["close"] = pd.to_numeric(
        result["close"],
        errors="coerce",
    )
    result["volume"] = pd.to_numeric(
        result["volume"],
        errors="coerce",
    )

    if result[["close", "volume"]].isna().any().any():
        raise RuntimeError("行情关键字段存在无效数值")

    if (result["close"] <= 0).any():
        raise RuntimeError("行情包含非正价格")

    if (result["volume"] < 0).any():
        raise RuntimeError("行情包含负成交量")

    # 标的主数据提供市场身份，不通过代码长度猜测市场。
    result = result.merge(
        master,
        on="symbol",
        how="left",
        validate="one_to_one",
        indicator=True,
    )

    if (result["_merge"] != "both").any():
        raise RuntimeError("部分行情无法关联到标的主数据")

    result = result.drop(columns=["_merge"])

    return result[
        ["symbol", "market", "timezone", "close", "volume"]
    ]


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

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

    master = build_instrument_master()
    symbols = master["symbol"].tolist()

    qd = QuantDash(api_key=api_key)

    try:
        logger.info(
            "开始获取跨市场行情，标的数量: %d",
            len(symbols),
        )

        raw_quotes = qd.quotes.get(
            symbols=symbols,
            to_dataframe=True,
        )

        normalized = normalize_quotes(
            raw_quotes,
            symbols,
            master,
        )

        # 注意：此处仅展示标准化结果。
        # 跨市场因子计算前，还需按行情时间、交易日历和字段语义
        # 完成进一步校验，不能直接把所有市场的数据视为同一时点。
        print(normalized.to_string(index=False))

        logger.info("跨市场行情标准化完成")

    except Exception:
        logger.exception("跨市场行情处理失败")
        raise


if __name__ == "__main__":
    try:
        main()
    except Exception:
        sys.exit(1)
```

这段实现有一个有意保留的限制。

它没有根据代码字符串自动推断所有可能的市场，也没有尝试把不同市场的行情时间强行统一成一个时间点。

这是为了避免把不确定性藏在看似方便的工具函数中。

在真实系统中，标的主数据应该进一步包含证券名称、交易所、交易状态和必要的证券属性。市场规则层则需要维护各市场的交易日历和时区信息。

此外，实时行情接口返回的字段名称应以实际 SDK 响应为准。如果接口返回的是其他字段名称，应当在接入适配层进行明确映射，而不是在策略中不断增加兼容分支。

### 让因子计算不再依赖市场接口

行情标准化之后，因子模块可以只接受统一结构。

例如，动量计算可以单独定义为一个函数：

```python
import pandas as pd


def calculate_cross_sectional_score(
    df: pd.DataFrame,
) -> pd.DataFrame:
    """计算单个已对齐时间截面的简单价格排名。"""
    required = {"symbol", "market", "close"}

    missing = required - set(df.columns)

    if missing:
        raise ValueError(
            f"因子输入缺少字段: {sorted(missing)}"
        )

    if df.empty:
        raise ValueError("因子输入不能为空")

    result = df.copy()

    result["close"] = pd.to_numeric(
        result["close"],
        errors="coerce",
    )

    if result["close"].isna().any():
        raise ValueError("价格字段存在无效值")

    if (result["close"] <= 0).any():
        raise ValueError("价格必须大于零")

    result["score"] = result.groupby("market")[
        "close"
    ].rank(
        method="average",
        ascending=False,
        pct=True,
    )

    return result.sort_values(
        ["market", "score"],
        ascending=[True, False],
    )
```

这里有一个重要细节。

示例按照市场分组排名，而不是直接将所有市场的原始价格混合排序。

原因很简单：AAPL 的价格、港股的价格和 A 股的价格并不能直接用于跨市场强弱比较。

即使已经统一了数据格式，也不代表价格具有可比性。

真正的跨市场选股，需要使用具有一致定义的收益率、标准化动量或者其他可比较的因子，并确保输入数据对应可解释的时间区间。

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

### 避坑点一：不要用代码长度判断市场

代码字符串可以帮助校验格式，但不应该成为完整的证券身份数据库。

尤其是扩展到不同市场后，代码长度、前导零和交易所后缀都可能影响数据关联。

我更倾向于通过标准化标的代码关联独立的标的主数据。

无法匹配的代码应该进入异常处理流程，而不是自动转换成某种猜测结果。

### 避坑点二：同一个字段必须有明确的数据语义

字段名统一之后，还要确认单位、时间和计算口径。

例如，价格是最新成交价还是某根 K 线的收盘价，成交量采用什么单位，行情时间对应哪个时区。

如果这些定义没有统一，所谓的标准化只是把不同的数据包装成了相同的列名。

### 避坑点三：跨市场选股不能按行号对齐

不同市场的交易时间和交易日历并不相同。

即使两组数据都存在 100 行，也不能认为第 50 行对应同一个时间截面。

跨市场因子计算应当依据明确的时间戳、市场交易日历和策略的时间对齐规则完成。

如果某个市场当前没有交易，也不能直接把最近一次价格当作当前市场价格参与比较。

## 六、写在最后

我现在看待多市场数据层时，首先关注的不是能够接入多少个市场，而是新增市场会不会迫使现有策略跟着修改。一个稳定的数据层应该让标的身份、行情字段和市场规则各自承担明确的职责。行情获取方式可以统一，市场之间的交易规则却不应该被强行抹平。

对策略研发而言，真正有价值的标准化不是让所有数据看起来一样，而是让数据的差异变得可见、可解释、可管理。这样做以后，新增市场主要影响标的主数据和市场适配逻辑，因子模块则可以在明确的数据契约下持续复用。维护成本能否长期控制，往往取决于这些边界是否在最初就划清楚。

## 参考

- QuantDash Python SDK 文档：[https://docs.quantdash.net/zh-Hans/sdk/python-quickstart](https://cloud.tencent.com/developer/tools/blog-entry?target=https%3A%2F%2Fdocs.quantdash.net%2Fzh-Hans%2Fsdk%2Fpython-quickstart&objectId=2762046&objectType=1&contentType=markdown)

我以前在做跨市场选股时发现，最初为 A 股编写的数据处理逻辑，一旦扩展到港股和美股，就会出现代码映射、字段类型、交易时段和行情时间不一致的问题。本文从一次跨市场选股重构出发，讨论如何通过统一标的身份、标准化行情字段和独立的市场规则层，构建能够持续扩展的实时选股数据层。

从 A 股扩展到美股和港股后，我才发现实时选股最难维护的是数据接口

大数据

数据接口

数据处理

代码解释

数据库

Python

JSON

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

社区新版编辑器体验调研

从 A 股扩展到美股和港股后，我才发现实时选股最难维护的是数据接口-腾讯云开发者社区-腾讯云

从 A 股扩展到美股和港股后，我才发现实时选股最难维护的是数据接口

从 A 股扩展到美股和港股后，我才发现实时选股最难维护的是数据接口

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐