统一行情 API 查 A 股、港股、美股和数字货币，返回成功了但资产少了一类，量化任务怎么校验？

跨市场任务不能把 code=0 当成批次完整。

>

在 2026 年 6 月 13 日的 TickDB MCP get_ticker 实测中，code=0 表示本次工具调用成功，但不表示请求的 symbol 已经完整返回。

>

下游使用数据前，必须核对请求集合、返回集合和资产类型，并检查意外缺失、意外新增及重复项。发现异常时，应关闭批次或明确降级，不能静默继续。

一、批次成功，不等于横截面完整

量化任务处理跨市场数据时，通常隐含着一个假设：

一次请求拉回来的数据，是一个完整的横截面。

例如，一次查询同时包含：

• A 股
• 港股
• 美股
• 数字货币
• 外汇

如果返回结果中少了一类资产，而程序只检查 code=0，这份不完整数据就可能直接进入下游。

问题在于，任务状态和数据状态不是一回事。

在本轮 TickDB MCP get_ticker 测试中，code=0 只能说明本次工具调用成功。它不能替你回答以下问题：

1. 返回的唯一 symbol 集合是否符合预期？
2. 有没有请求了却没有返回的 symbol？
3. 有没有未请求却出现在结果中的 symbol？
4. 请求或返回结果中有没有重复 symbol？
5. 返回品种的资产类型是否正确？

因此：

批次成功是任务状态，横截面完整是数据状态。两者必须分别判断。

二、资产缺失会怎样污染下游

假设某个任务定时获取五类资产的快照，用于跨市场比价、波动率矩阵或监控面板。

某次请求中，数字货币品种没有返回，但任务仍然显示成功。此时可能出现三类问题。

1. 比率计算失真

如果计算依赖 A 股和数字货币的价格比，数字货币缺失后，这个比率要么无法计算，要么被错误地填入上一期缓存值。

前者应当中断，后者则会制造一条看起来正常、实际上已经失真的结果。

2. 权重发生偏移

如果面板按照资产类别分配权重，少了一类资产后，其余类别的相对权重可能被动提高。

此时波动率、风险暴露等指标的统计口径已经改变，但系统未必会报错。

3. 出现监控盲区

如果告警规则是“任意资产波动超过阈值”，缺失的资产自然不会触发告警。

这不代表市场没有变化，只代表数据没有进入系统。

这些问题的共同特点是：静默。

程序没有崩溃，任务没有报错，结果表也可能正常生成，但计算输入已经不完整。

三、实测：code=0 对应三种不同的数据状态

以下测试于 2026 年 6 月 13 日通过 TickDB MCP get_ticker 完成。

!image.png

测试只核对：

• 返回状态 code
• 请求 symbol
• 返回 symbol
• 返回资产类型 type

价格、成交量和 timestamp 属于动态行情，不作为本文结论依据。

测试一：不传 type

请求五个代表品种：

600519.SH
700.HK
AAPL.US
BTCUSDT
EURUSD

脱敏后的关键结果为：

{
  "code": 0,
  "returned": [
    {"symbol": "700.HK", "type": "stock"},
    {"symbol": "BTCUSDT", "type": "crypto"},
    {"symbol": "EURUSD", "type": "forex"},
    {"symbol": "AAPL.US", "type": "stock"},
    {"symbol": "600519.SH", "type": "stock"}
  ]
}

五个请求品种均有返回。

注意，返回顺序不应被当作请求顺序。集合完整性检查应比较 symbol 集合，而不是依赖数组位置。

测试二：传入 type=stock

使用相同的五个 symbol，并传入：

type=stock

脱敏后的关键结果为：

{
  "code": 0,
  "returned": [
    {"symbol": "700.HK", "type": "stock"},
    {"symbol": "AAPL.US", "type": "stock"},
    {"symbol": "600519.SH", "type": "stock"}
  ]
}

本轮 get_ticker 混合 symbol 测试中，只返回了三只股票：

600519.SH
700.HK
AAPL.US

以下两个品种没有返回：

BTCUSDT
EURUSD

如果任务在请求前已经将这两个 symbol 声明为“针对当前工具验证过的预期排除项”，本批次可以降级放行。

如果程序并不知道它们会被排除，只看到 code=0 就继续运行，那么相同结果就应被视为集合不完整。

这里的关键不是“缺失了什么”，而是：

缺失是否提前声明、是否经过验证、是否被写入批次状态。

测试三：混入无效代码

请求：

AAPL.US
NOTREAL

脱敏后的关键结果为：

{
  "code": 0,
  "returned": [
    {"symbol": "AAPL.US", "type": "stock"}
  ]
}

NOTREAL 没有出现在返回结果中，但本次调用仍然返回 code=0。

如果 NOTREAL 不是事先声明的预期排除项，这就是一次意外缺失。生产任务不应将其视为完整批次。

三组结果可以归纳为：

所以，code=0 之后仍然需要一次独立的数据完整性判断。

四、什么时候允许下游继续

!image.png

可以在 MCP 调用与下游计算之间增加一道完整性闸门。

完整批次至少应满足以下条件：

返回集合 = 请求集合 - 预期排除集合

并且：

• 没有意外缺失；
• 没有意外新增；
• 请求和返回中没有重复 symbol；
• 返回 type 与预期资产类型一致；
• 预期排除项没有意外出现在返回中。

五、把完整性校验写进数据管道

校验可以分为三步。

第一步：请求前定义预期

记录本次请求的 symbol 和对应资产类型。

如果使用了 type 等资产类别消歧参数，还要记录：

• 哪些 symbol 预计不会返回；
• 这种行为是否已针对当前工具验证；
• 本批次是否允许降级继续。

第二步：返回后提取实际集合

从返回记录中提取：

(symbol, type)

不要只提取价格，也不要只检查数组是否为空。

第三步：执行多维比对

至少检查：

• 记录结构；
• 请求重复；
• 返回重复；
• 意外缺失；
• 意外新增；
• 资产类型错配；
• 预期排除项是否意外返回。

任一异常未被解释，都不应进入下游计算。

六、可执行的教学示例

!image.png

下面的代码演示如何完成上述检查。

它是教学示例，不是生产级代码。生产环境还需要补充日志、异常捕获、批次 ID、重试策略和指标上报。

def validate_cross_market_batch(
    requested: list,
    returned: list,
    expected_excluded: set = None,
):
    """教学示例，非生产级代码。

    Args:
        requested:
            请求记录列表，每项为 (symbol, expected_type)。
        returned:
            返回记录列表，每项为 (symbol, actual_type)。
        expected_excluded:
            事先声明且已针对当前工具验证的预期排除 symbol。

    Returns:
        (是否允许继续, 诊断信息)
    """
    if expected_excluded is None:
        expected_excluded = set()

    # 所有记录必须是包含 symbol 和 type 的二元组。
    for index, item in enumerate(requested):
        if not isinstance(item, tuple) or len(item) != 2:
            return False, f"请求记录结构无效: 索引 {index}"
        if not item[0] or not item[1]:
            return False, f"请求记录字段为空: 索引 {index}"

    for index, item in enumerate(returned):
        if not isinstance(item, tuple) or len(item) != 2:
            return False, f"返回记录结构无效: 索引 {index}"
        if not item[0] or not item[1]:
            return False, f"返回记录字段为空: 索引 {index}"

    requested_symbols = [item[0] for item in requested]
    returned_symbols = [item[0] for item in returned]

    requested_set = set(requested_symbols)
    returned_set = set(returned_symbols)

    # 预期排除项必须来自本次请求。
    invalid_exclusions = expected_excluded - requested_set
    if invalid_exclusions:
        return (
            False,
            f"配置错误，预期排除项不在请求集合中: {invalid_exclusions}",
        )

    issues = []

    requested_duplicates = [
        symbol
        for symbol in requested_set
        if requested_symbols.count(symbol) > 1
    ]
    if requested_duplicates:
        issues.append(f"请求重复: {requested_duplicates}")

    returned_duplicates = [
        symbol
        for symbol in returned_set
        if returned_symbols.count(symbol) > 1
    ]
    if returned_duplicates:
        issues.append(f"返回重复: {returned_duplicates}")

    # 预期排除的 symbol 不应出现在返回集合中。
    unexpectedly_included = returned_set & expected_excluded
    if unexpectedly_included:
        issues.append(
            f"预期排除项意外返回: {unexpectedly_included}"
        )

    missing = requested_set - returned_set
    unexpected_missing = missing - expected_excluded
    expected_missing = missing & expected_excluded
    unexpected_extra = returned_set - requested_set

    if unexpected_missing:
        issues.append(f"意外缺失: {unexpected_missing}")

    if unexpected_extra:
        issues.append(f"意外新增: {unexpected_extra}")

    requested_types = dict(requested)
    returned_types = dict(returned)

    type_mismatches = []
    for symbol in requested_set & returned_set:
        expected_type = requested_types[symbol]
        actual_type = returned_types[symbol]

        if expected_type != actual_type:
            type_mismatches.append(
                f"{symbol}: 预期 {expected_type}, 实际 {actual_type}"
            )

    if type_mismatches:
        issues.append(f"类型不一致: {type_mismatches}")

    if issues:
        return False, "批次关闭: " + "; ".join(issues)

    if returned_set == requested_set and not expected_excluded:
        return True, "完整通过"

    if returned_set == requested_set - expected_excluded:
        return True, f"降级继续，预期排除: {expected_missing}"

    return False, "批次关闭: 返回集合状态异常"

使用五类资产构造预期集合：

requested = [
    ("600519.SH", "stock"),
    ("700.HK", "stock"),
    ("AAPL.US", "stock"),
    ("BTCUSDT", "crypto"),
    ("EURUSD", "forex"),
]

returned = [
    ("600519.SH", "stock"),
    ("700.HK", "stock"),
    ("AAPL.US", "stock"),
]

allowed, message = validate_cross_market_batch(
    requested,
    returned,
    expected_excluded={"BTCUSDT", "EURUSD"},
)

print(allowed, message)

预期结果：

True 降级继续，预期排除: {'BTCUSDT', 'EURUSD'}

集合的显示顺序可能不同，但不影响判断。

七、校验器也需要反例测试

仅测试正常数据不够。完整性校验器必须证明自己会在异常情况下关闭批次。

反例一：预期排除项意外返回

returned = [
    ("600519.SH", "stock"),
    ("700.HK", "stock"),
    ("AAPL.US", "stock"),
    ("BTCUSDT", "crypto"),
]

若 BTCUSDT 已被声明为预期排除项，它却出现在返回结果中，说明实际行为与批次配置不一致。

校验器应返回：

False 批次关闭: 预期排除项意外返回

反例二：返回记录缺少类型

returned = [
    ("600519.SH", ""),
    ("700.HK", "stock"),
    ("AAPL.US", "stock"),
]

校验器应返回：

False 返回记录字段为空

反例三：返回记录结构不一致

returned = [
    ("600519.SH", "stock"),
    "700.HK",
    ("AAPL.US", "stock"),
]

校验器应返回：

False 返回记录结构无效

反例四：资产类型错配

returned = [
    ("600519.SH", "index"),
    ("700.HK", "stock"),
    ("AAPL.US", "stock"),
]

校验器应返回：

False 批次关闭: 类型不一致

这些反例证明的是客户端校验逻辑，而不是 MCP 本身会产生所有这些异常。

MCP 调用负责提供结构化返回；是否允许这批数据进入下游，仍然要由你的程序决定。

八、量化工程师检查清单

每次跨市场任务进入下游前，可以逐项确认：

• [ ] 已定义请求 symbol 和预期资产类型；
• [ ] 已保存本次请求的原始 symbol 集合；
• [ ] 预期排除项已经针对当前工具验证；
• [ ] 已从返回结果中提取 symbol 和 type；
• [ ] 已检查请求与返回中的重复 symbol；
• [ ] 已检查意外缺失；
• [ ] 已检查意外新增；
• [ ] 已检查资产类型错配；
• [ ] 已检查预期排除项是否意外返回；
• [ ] 意外异常会关闭批次，而不是只打印警告；
• [ ] 降级批次会记录排除条件和缺失项；
• [ ] 下游只消费通过完整性检查的批次；
• [ ] 新工具、新端点或新资产类别接入时，会重新验证实际行为。

不能从本文推出什么

本文的证据边界需要明确：

• 三组结果仅来自 2026 年 6 月 13 日的 TickDB MCP get_ticker 调用；
• 本轮 code=0 行为不代表其他 MCP 工具具有相同行为；
• MCP 实测不能直接证明 REST、WebSocket、K 线或逐笔接口的行为；
• 五个代表品种成功返回，不表示所有品种在所有时间均可用；
• 重复项、意外新增和类型错配的关闭行为，由客户端校验器反例证明，不是 MCP 异常实测；
• 下游比率失真、权重偏移和监控盲区属于工程风险推演，不是 MCP 返回结论；
• 本文不涉及策略收益、回测绩效或买卖决策；
• 本文不构成投资建议。

跨市场数据任务真正需要判断的，不只是“请求有没有成功”，而是：

我要求的数据是否不多、不少、类型正确地回来了？

只看 code=0，你知道的是工具调用结束了。

核对请求集合、返回集合和资产类型之后，你才知道这批数据是否可以进入下游。

通过 TickDB API 获取实时行情数据

一个 API 接入外汇、加密货币、美股、港股、A股、贵金属和全球指数的实时行情。支持 WebSocket 低延迟推送，免费开始使用。