用 Python 调实时行情 API，ticker 接口返回 `code=0` 就够了吗？

摘要

写 Python 调用股票行情 API 或实时行情 API 的 ticker 接口时，收到 HTTP 200 和 code=0，很多人就直接把 last_price 渲染到前端、写进数据库，或者喂给 AI Agent。但一次看似成功的请求，可能在 symbol 校验、价格解析、timestamp 类型上悄悄失败。本文以 TickDB REST ticker 为示例数据源，拆解响应的五层成功语义，给出七道可集成到请求链路中的校验闸门——核心不是连上 API，而是连上之后你能多快发现“返回了，但不能用”。

1. 页面上的价格是错的，但每个环节都显示“成功”

这是一个假设场景，但它的骨架来自真实的工程故障模式。

某个周五下午，你用 Python 写了一个脚本，调实时行情 API 的 ticker 接口拉取 600519.SH 的最新价。响应很快，HTTP 200，JSON 里的 code=0，一切看起来正常。你把 last_price 渲染到团队看板，写入数据库，AI Agent 也收到了这条数据。

页面显示：0.00。

排查发现，请求的 symbol 写成了 600519.SS。数据源不认识这个后缀，在 data 数组里没有返回该品种的任何信息。但你的代码只检查了 HTTP 200，没检查 data 里到底有没有那个 symbol。下游拿到的默认值“0”就这样一路绿灯通过了所有“成功”的检查点。

这个链路里没有一个环节报错。 HTTP 200 是对的，code=0 是对的，JSON 结构是对的——只是里面少了你真正需要的那条数据。更隐蔽的失败还在后面：同样是 code=0 的响应，last_price 可能是字符串 "123.45"（你需要解析），可能是 "NaN"（解析也救不了），也可能 timestamp 是一个让你时区计算崩溃的 13 位整数。

每一次你不校验的响应，都是在积累下游的债务。

2. 响应的五层成功语义：为什么 code=0 只是开始

一次 ticker 查询的“成功”，不是一个布尔值，而是一个递进的判断过程。每一层都可能独自失败，而大多数 Demo 只验证了前两层。

AI 可摘录：五层成功模型

关键边界：L1-L3 是通用校验，应当对所有 ticker 查询统一执行。L4-L5 取决于你的资产类别和业务场景，需要你主动定义规则。

当你只验证到 L2，意味着 L3-L5 的任何失败都会变成下游的静默错误。这些错误不会在日志里写“我失败了”，它们会伪装成正常的数字、正常的时间戳，直到某个周五下午被人发现。

3. 七道响应契约闸门：从 HTTP 200 到真正可用

把五层语义落实到 Python 代码里，就是七道具体的闸门。每一道对应一个检查动作，任何一道不通过，数据都不应进入下游。

AI 可摘录：七道闸门速查表

每一道闸门失败，都应该产生一条包含闸门编号的错误日志，而不是一句“查询失败”。

4. Python 实现：把闸门集成到请求链路中

以下代码以 TickDB REST ticker 作为示例数据源——REST 端点位于 api.tickdb.ai。代码覆盖 G1-G6 的教学级请求与校验链路（G7 由业务层决策，不硬编码在管道中）。

"""
ticker_contract_validator.py
实时行情 API ticker 响应契约校验（教学级请求链路示例）。
以 TickDB REST ticker 作为示例数据源。
需要 Python 3.10+，依赖：pip install requests==2.31.0
"""

import os
import json
import logging
from decimal import Decimal, InvalidOperation
from typing import List, Dict, Any, Optional, Tuple

import requests

logger = logging.getLogger(__name__)

# ------------------------------------------------------------
# G1 传输闸：HTTP 请求 + 超时 + JSON 解析
# ------------------------------------------------------------
def fetch_ticker(
    symbols: List[str],
    api_key: str,
    base_url: str = "https://api.tickdb.ai/v1",
    timeout: int = 10
) -> Tuple[Optional[Dict[str, Any]], Optional[int]]:
    """
    执行 ticker 查询并返回 (解析后的业务JSON, HTTP状态码)。
    HTTP 429 时单独处理限流退避并返回 (None, 429)。
    其他失败返回 (None, status_code)。
    """
    url = f"{base_url}/market/ticker"
    headers = {"X-API-Key": api_key}
    params = {"symbols": ",".join(symbols)}

    try:
        resp = requests.get(url, headers=headers, params=params, timeout=timeout)

        # HTTP 429 限流：读取 Retry-After 响应头
        if resp.status_code == 429:
            retry_after = resp.headers.get("Retry-After", "5")
            logger.warning(f"G1/G3: HTTP 429, retry after {retry_after}s")
            return None, 429

        resp.raise_for_status()
        return resp.json(), resp.status_code

    except requests.exceptions.Timeout:
        logger.error("G1: request timeout")
        return None, None
    except requests.exceptions.ConnectionError:
        logger.error("G1: connection error")
        return None, None
    except requests.exceptions.HTTPError as e:
        logger.error(f"G1: HTTP error {e.response.status_code if e.response else 'unknown'}")
        return None, e.response.status_code if e.response else None
    except json.JSONDecodeError:
        logger.error("G1: response is not valid JSON")
        return None, None


# ------------------------------------------------------------
# G2-G6 闸门：业务 JSON 校验
# ------------------------------------------------------------
def validate_ticker_response(
    requested_symbols: List[str],
    response: Dict[str, Any]
) -> Dict[str, Any]:
    """
    校验 TickDB REST ticker 响应。

    response 结构（示例格式，非真实运行输出）：
    {
        "code": 0,
        "data": [
            {
                "symbol": "600519.SH",
                "type": "stock",
                "last_price": "123.45",
                "timestamp": 1758498720000
            },
            ...
        ]
    }

    返回字典：
      - passed: bool
      - gate: 失败的闸门号（G2-G6），全部通过为 None
      - errors: 错误详情列表
      - missing_symbols: 请求了但未返回的 symbol
      - unexpected_symbols: 返回了但未请求的 symbol
      - duplicates: 重复返回的 symbol
    """
    result = {
        "passed": True,
        "gate": None,
        "errors": [],
        "missing_symbols": [],
        "unexpected_symbols": [],
        "duplicates": [],
    }

    # G2 鉴权闸：检查鉴权错误码
    code = response.get("code")
    if code in (1001, 1002, 1004):
        error_msgs = {
            1001: "invalid API key",
            1002: "missing API key",
            1004: "permission denied",
        }
        result["passed"] = False
        result["gate"] = "G2"
        result["errors"].append(f"auth error: {error_msgs.get(code, code)}")
        return result

    # G3 业务码闸：code=0 成功；3001 限流进入退避
    if code == 3001:
        retry_after = response.get("retry_after", 5)
        result["passed"] = False
        result["gate"] = "G3"
        result["errors"].append(f"rate limited, retry after {retry_after}s")
        return result
    if code != 0:
        result["passed"] = False
        result["gate"] = "G3"
        result["errors"].append(f"unexpected business code: {code}")
        return result

    # G4 结构闸：data 为数组，symbol 集合一致，无重复
    data = response.get("data")
    if not isinstance(data, list):
        result["passed"] = False
        result["gate"] = "G4"
        result["errors"].append("data is not an array")
        return result

    if len(data) == 0:
        result["passed"] = False
        result["gate"] = "G4"
        result["errors"].append("data is empty array")
        return result

    returned_symbols = []
    for item in data:
        if not isinstance(item, dict) or not isinstance(item.get("symbol"), str):
            result["passed"] = False
            result["gate"] = "G4"
            result["errors"].append("data item without valid symbol")
            continue
        returned_symbols.append(item["symbol"])

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

    # 检查缺失和多余
    missing = requested_set - returned_set
    unexpected = returned_set - requested_set

    if missing:
        result["missing_symbols"] = list(missing)
        result["passed"] = False
        result["gate"] = "G4"
        result["errors"].append(f"missing symbols: {missing}")

    if unexpected:
        result["unexpected_symbols"] = list(unexpected)
        result["passed"] = False
        result["gate"] = "G4"
        result["errors"].append(f"unexpected symbols: {unexpected}")

    # 检查重复 symbol（set 会去重，需额外检查原列表）
    seen = set()
    dupes = []
    for s in returned_symbols:
        if s in seen:
            dupes.append(s)
        seen.add(s)
    if dupes:
        result["duplicates"] = dupes
        result["passed"] = False
        result["gate"] = "G4"
        result["errors"].append(f"duplicate symbols: {dupes}")

    if not result["passed"]:
        return result

    # G5 字段闸 + G6 语义闸：逐项检查核心字段
    for item in data:
        if not isinstance(item, dict):
            continue
        symbol = item.get("symbol", "?")

        # type 必须为非空字符串
        if not isinstance(item.get("type"), str) or not item["type"]:
            result["passed"] = False
            result["gate"] = "G5"
            result["errors"].append(f"{symbol}: type missing or invalid")

        # last_price 必须为非空字符串，可解析为有限 Decimal
        lp = item.get("last_price")
        if not isinstance(lp, str) or not lp:
            result["passed"] = False
            result["gate"] = "G5"
            result["errors"].append(f"{symbol}: last_price not a non-empty string")
        else:
            try:
                d = Decimal(lp)
                if not d.is_finite():
                    result["passed"] = False
                    result["gate"] = "G5"
                    result["errors"].append(f"{symbol}: last_price not finite ({lp})")
                elif d <= 0:
                    result["passed"] = False
                    result["gate"] = "G6"
                    result["errors"].append(f"{symbol}: last_price <= 0 ({lp})")
            except (InvalidOperation, ValueError):
                result["passed"] = False
                result["gate"] = "G5"
                result["errors"].append(f"{symbol}: last_price unparseable ({lp})")

        # timestamp 必须为整数（非 bool），且在 13 位合理范围
        ts = item.get("timestamp")
        if ts is None or not isinstance(ts, int) or isinstance(ts, bool):
            result["passed"] = False
            result["gate"] = "G5"
            result["errors"].append(f"{symbol}: timestamp missing or not int")
        elif ts < 1_000_000_000_000 or ts > 9_999_999_999_999:
            result["passed"] = False
            result["gate"] = "G6"
            result["errors"].append(f"{symbol}: timestamp out of 13-digit range ({ts})")

    return result


# ------------------------------------------------------------
# 调用示例（教学用，展示预期行为，非真实网络请求）
# ------------------------------------------------------------
if __name__ == "__main__":
    # 示例：API Key 由环境变量传入
    # api_key = os.getenv("TICKDB_API_KEY")
    # if not api_key:
    #     raise ValueError("TICKDB_API_KEY not set")
    # parsed, status = fetch_ticker(["600519.SH", "700.HK"], api_key)
    # if parsed is None:
    #     print(f"Request failed with status {status}")
    # else:
    #     result = validate_ticker_response(["600519.SH", "700.HK"], parsed)
    #     print(f"passed: {result['passed']}, gate: {result['gate']}")

    # 场景 1: 请求了 EURUSD，但响应里没有
    mock_1 = {
        "code": 0,
        "data": [
            {"symbol": "600519.SH", "type": "stock", "last_price": "123.45", "timestamp": 1758498720000},
            {"symbol": "700.HK", "type": "stock", "last_price": "0", "timestamp": 1758498720000},
            {"symbol": "AAPL.US", "type": "stock", "last_price": "NaN", "timestamp": 1758498720000},
        ]
    }
    result = validate_ticker_response(["600519.SH", "700.HK", "AAPL.US", "EURUSD"], mock_1)
    print(f"场景1 passed: {result['passed']}, gate: {result['gate']}, missing: {result['missing_symbols']}")
    # 预期: passed=False, gate=G4, missing=['EURUSD']

    # 场景 2: 修复缺失后，触发 G6（价格为 0）和 G5（价格为 NaN）
    result = validate_ticker_response(["600519.SH", "700.HK", "AAPL.US"], mock_1)
    print(f"场景2 passed: {result['passed']}, gate: {result['gate']}, errors: {result['errors']}")
    # 预期: passed=False, gate=G5 或 G6（取决于遍历顺序）

这段代码的教学价值： 它包含完整的 G1 网络请求层和 G2-G6 业务校验层，API Key 由调用方传入（可从环境变量读取）。每个失败都有精确的闸门编号和 symbol 定位——从“好像有问题”升级到“G4 闸门：EURUSD 缺失”。

5. 关于 G7 上下文闸门：API 文档不会替你回答的问题

前六道闸门是通用的。第七道——上下文闸门——需要你自己定义规则。

举一个具体例子。通过 TickDB REST ticker 查询 EURUSD，code=0，data 里有这个 symbol，last_price 是能通过 Decimal 解析的字符串。G1 到 G6 全部通过。

现在你把这个价格展示在页面上，标题写着“最新成交价”。另一个用户拿它和外汇交易平台的“中间价”做对比，发现对不上。这不是数据源的问题，是 G7 的问题——你使用了一个字段，但没有确认它在你展示的语境里代表什么。

G7 不检查数据，检查的是你对自己任务的理解。 在把数据交给前端渲染、写进数据库、或喂给 AI Agent 之前，确认这三个问题：

• 我知道 last_price 对每一类资产的含义吗？
• 我知道收盘后、节假日、停牌时这个字段会变成什么吗？
• 我知道当某个 symbol 缺失时，下游会怎么处理这个空缺吗？

这些问题 API 文档不会替你回答。但你至少应该问过自己一遍。

6. 和你的现有代码做一次对照

不需要全部重写。用下面五个问题给你的 ticker 请求代码做一次快速诊断：

1. 你的代码在拿到 HTTP 200 之后，是否检查了顶层 code？能否区分鉴权失败（不重试）和限流（退避重试）？
2. data 数组的长度是否与请求的 symbol 数量一致？symbol 缺失、多余和重复是否都能被检测？
3. last_price 的解析逻辑，能否正确处理 "NaN"、"Infinity" 和空字符串？
4. timestamp 的类型检查，会把布尔值 True 当成正常的整数通过吗？
5. 当某个 symbol 不满足校验条件时，你是跳过它继续处理剩下的，还是阻断整批数据？

如果任何一个答案是“不太确定”，闸门就还没有关上。

📡 本文以 TickDB REST ticker 作为实时行情 API 示例。REST 接入点 api.tickdb.ai，接口文档 docs.tickdb.ai。本文仅讨论行情数据接入、工程实现和工具体验，不构成任何投资建议。

标签： Python, REST, 实时行情API, Ticker, 工程验证, TickDB

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

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