Python 获取 A 股实时行情，为什么 600519 和 600519.SH 不是一回事？

凌晨两点，你盯着终端，第八次重跑同一个 Python 脚本。API Key 是对的，网络是通的，requests.get 返回了 HTTP 200。但你传 600519 进去，返回的 data 字段是空的，或者报了一个品种不存在的业务码。

你查过 API 文档，确认端点路径没错。你用 curl 重放了完全相同的请求——依旧空。你甚至怀疑是 Key 的权限不够，去后台翻了一圈，权限正常。

后来你在某个 GitHub issue 的评论区看到一句话：你传的是用户输入，不是 symbol。顺着这个线索，你发现你用的那个行情接口根本没有提供“可用品种查询”的端点——你只能在黑暗中扔请求，靠返回是否为空来判断代码对不对。

股票代码、用户输入的字符串、API 要求的 symbol 参数，是三层完全不同的东西。

>

把用户输入直接当参数传，等于把“朝阳区”当成快递地址填上去——快递系统可能靠经验猜出“北京市朝阳区”，也可能直接退回。

>

工程上该做的，不是在请求发出后祈祷接口容错，而是在请求发出前完成身份翻译和校验。三层身份不对齐，是实时行情 API 接入中最高频的静默失败源——它不会报网络错误，只会安静地返回空数据，让你的下游计算在“一切正常”的假象下空转。

一、三层身份，只有一层能上路

用户在你的输入框里敲的东西，可能是以下任意一种：

• 600519
• 贵州茅台
• 600519.SH
• sh600519
• Kweichow Moutai
• 甚至一个带有全角字符的 ６００５１９

这些都可以称之为“用户意图”——人想表达的东西。但它不是股票代码，更不是 API symbol。

真正的股票代码，是交易所分配的一串标识符。上交所上市的公司股票通常是六位数字，深交所也是六位。但同一个六位数字理论上可以先后出现在不同交易所的不同证券上——000001 今天是平安银行，但“000001”这个字符串本身不天然属于任何一个市场。

而一个设计更细致的行情 API，会要求 symbol 是代码加市场后缀的规范化组合：

600519.SH    000001.SZ    920186.BJ

后缀不是可选的装饰，是 symbol 不可分割的一部分。没有后缀，API 就无法确定你指向的是哪个市场的哪个品种。此时可能返回空数据或报品种不存在类错误——至于具体返回什么，取决于该接口的当前实现。

工程上更稳的做法，不是去测试接口对未规范化输入的容错边界，而是在请求发出前就用一个可靠的工具把身份查清楚。

比如 TickDB 这类行情 API，提供了一个 /v1/symbols/available 品种查询端点。调用它，你会拿到一个当前可用品种的规范 symbol 列表，每个都是 代码.后缀 的格式。这意味着你不用自己维护一份“600519 等于 600519.SH”的静态映射表，也不用担心退市或代码变更让你的映射表过时——你每次跑脚本前都可以拉一次最新列表，把身份翻译这件事交给数据源本身。

这种设计，相当于快递分拣系统直接给你一本最新的标准地址簿，而不是让你拿着手写面单去猜。

二、三道闸门：在请求发出前解决问题

把用户意图翻译成 API 可消费的 symbol，靠“传进去试试看”是危险的。不是因为每次都失败——恰恰是因为有时候能成功，你会误以为这个做法是可靠的。

工程上更稳的方式，是在请求发出之前设三道闸门。每一道只做一件事，不混在一起。

第一道：输入规范化

把用户输入的字符串清洗掉首尾空格、全角字符、不可见字符，然后尝试匹配到一个候选代码。

• ６００５１９ → 600519
• sh600519 → 剥离前缀，得到 600519

这一步的输出是候选代码，不是最终 symbol。知道代码，不等于知道市场——这一步不做后缀补全。

第二道：可用品种校验

拿着候选代码去查行情 API 的品种列表端点。以 TickDB 为例，它的 /v1/symbols/available 返回的 data.products[].symbol 就是规范格式，你可以在返回的列表里直接搜索 600519.SH、000001.SZ 这类 symbol。

这步同时解决两个问题：

1. 确认代码对应的市场后缀到底是什么
2. 确认这个品种当前是否确实在可用列表中

品种的可用性可能因退市、停牌、代码变更而变化，不依赖于“本地存了一份静态映射表就永远有效”。这一步不通过，就不该发出 ticker 请求。

第三道：ticker 响应校验

行情数据返回后，不要假设返回结果数组的顺序和你发出的 symbol 列表顺序一致。要按 data[].symbol 字段做对齐校验：请求的是 600519.SH，返回的 symbol 字段就必须是 600519.SH。

同时校验两个关键字段：

• last_price：是否存在且可解析（用 Decimal，不用 float）
• timestamp：是否存在且类型为整数（ticker 端点返回的 timestamp 为毫秒 UTC 时间戳——但这描述的是字段精度，不代表数据延迟、新鲜度、采样频率或 SLA）

一个值得单独拎出来的工程细节：错误码分层

不是所有行情 API 都会把“Key 过期”和“品种不存在”分成两个不同的业务码，但分成两件事意味着你可以写出清晰的处理分支，而不是靠读日志里的文案去猜。

TickDB 的错误码体系就做了这种区分：

凭证类错误（1001/1002/1004）告诉你要检查 Key，品种类错误（2002）告诉你要检查 symbol，限流类错误（3001）告诉你要放慢节奏。一套结构化的错误码，让你在排错时不用猜谜，这是接入行情 API 时值得单独校验的工程细节。

三道闸门有一条不通过，就应该非零退出，而不是补一个默认值继续跑。一条被悄悄放过的错配数据，对下游的污染远大于一条缺失数据——因为你会以为它是对的。

三、代码实现：排错优先，不展示 happy path

下面这段 Python 代码演示了上述三道闸门的落地，完全基于 TickDB 的端点实现。它和你平时写的脚本有一个关键区别：它不是从“假设一切正常”开始的，而是从“每一行都可能失败”开始的。

import os
import sys
import requests
from decimal import Decimal, InvalidOperation

BASE_URL = "https://api.tickdb.ai/v1"
API_KEY = os.environ.get("TICKDB_API_KEY")
TIMEOUT = 10

# 教学示例，非生产级代码
# 示例中的价格字段值为接口返回值的占位说明，不构成实时报价展示


def normalize_input(user_input: str) -> str:
    """
    第一道闸门：输入规范化
    输出候选代码，不补后缀——后缀由第二道闸门通过可用品种列表确认
    """
    cleaned = user_input.strip().upper()

    # 去除常见的字母前缀
    for prefix in ("SH", "SZ", "BJ"):
        if cleaned.startswith(prefix) and len(cleaned) > len(prefix):
            cleaned = cleaned[len(prefix):]
            break

    # 若已是代码.后缀的格式，直接返回
    if "." in cleaned and len(cleaned.split(".")) == 2:
        return cleaned

    # 全角字符转半角（数字和字母）
    fullwidth_map = str.maketrans(
        "０１２３４５６７８９ＡＢＣＤＥＦＧＨＩＪＫＬＭＮＯＰＱＲＳＴＵＶＷＸＹＺ",
        "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ",
    )
    cleaned = cleaned.translate(fullwidth_map)

    if not cleaned:
        raise ValueError(f"输入为空或无法识别: {user_input}")
    return cleaned


def fetch_available_symbols() -> set:
    """
    第二道闸门：获取可用品种列表（TickDB /v1/symbols/available）
    失败不补缓存，直接退出——用一份空的可用列表往下跑，比不跑更危险
    """
    try:
        resp = requests.get(
            f"{BASE_URL}/symbols/available",
            headers={"X-API-Key": API_KEY},
            timeout=TIMEOUT,
        )
        resp.raise_for_status()
    except requests.RequestException as e:
        raise RuntimeError(f"可用品种列表请求失败: {e}") from e

    try:
        data = resp.json()
    except ValueError as e:
        raise RuntimeError(f"可用品种列表 JSON 解析失败: {e}") from e

    # TickDB 业务码校验：凭证类错误必须阻断并提示
    code = data.get("code")
    if code == 1001:
        raise RuntimeError("API Key 无效或已过期，请更换 Key")
    if code == 1002:
        raise RuntimeError("请求未提供 API Key")
    if code == 1004:
        raise RuntimeError("API Key 权限不足")
    if code != 0:
        raise RuntimeError(f"symbols/available 业务失败: code={code}")

    products = data.get("data", {}).get("products", [])
    if not products:
        raise RuntimeError("可用品种列表为空，中断请求——继续跑会把所有输入判为无效")

    symbols = set()
    for product in products:
        sym = product.get("symbol")
        if sym:
            symbols.add(sym)

    if not symbols:
        raise RuntimeError("可用品种列表中没有可解析的 symbol")
    return symbols


def fetch_ticker(symbol: str) -> dict:
    """
    第三道闸门：获取 ticker 快照（TickDB /v1/market/ticker）并逐项校验
    任何校验失败都抛异常——不补默认值，不跳过
    """
    try:
        resp = requests.get(
            f"{BASE_URL}/market/ticker",
            headers={"X-API-Key": API_KEY},
            params={"symbols": symbol},
            timeout=TIMEOUT,
        )
        resp.raise_for_status()
    except requests.RequestException as e:
        raise RuntimeError(f"ticker 请求失败: {symbol}, {e}") from e

    try:
        data = resp.json()
    except ValueError as e:
        raise RuntimeError(f"ticker JSON 解析失败: {symbol}, {e}") from e

    code = data.get("code")
    if code != 0:
        # TickDB 错误码分层：凭证类 → 换 Key；品种类 → 查可用列表；限流类 → 等 Retry-After
        if code == 1001:
            raise RuntimeError("API Key 无效或已过期，请更换 Key")
        if code == 1002:
            raise RuntimeError("请求未提供 API Key")
        if code == 1004:
            raise RuntimeError("API Key 权限不足")
        if code == 2002:
            raise ValueError(f"品种不存在——可用品种列表校验可能已过期: {symbol}")
        if code == 3001:
            retry_after = resp.headers.get("Retry-After", "未知")
            raise RuntimeError(f"频率限制: Retry-After={retry_after}")
        raise RuntimeError(f"ticker 未预期的业务码: code={code}")

    ticker_list = data.get("data", [])
    if not ticker_list:
        raise ValueError(f"ticker 返回空数据——symbol 可能在请求时已失效: {symbol}")

    for item in ticker_list:
        returned_symbol = item.get("symbol")
        if returned_symbol != symbol:
            raise ValueError(
                f"symbol 错配: 请求={symbol}, 返回={returned_symbol}"
            )

        raw_price = item.get("last_price")
        if raw_price is None:
            raise ValueError(f"last_price 缺失: {symbol}")
        try:
            price = Decimal(str(raw_price))
            if not price.is_finite():
                raise ValueError(
                    f"last_price 为非有限值: {symbol}, value={raw_price}"
                )
        except (InvalidOperation, ValueError) as e:
            raise ValueError(
                f"last_price 解析失败: {symbol}, value={raw_price}"
            ) from e

        raw_ts = item.get("timestamp")
        if raw_ts is None or isinstance(raw_ts, bool):
            raise ValueError(f"timestamp 缺失或为布尔: {symbol}")
        # timestamp 为毫秒 UTC 整数；此描述是字段精度，不表示延迟或新鲜度
        if not isinstance(raw_ts, int):
            raise ValueError(f"timestamp 类型异常: {symbol}, type={type(raw_ts)}")

    return data


def main():
    if not API_KEY:
        print("错误: 环境变量 TICKDB_API_KEY 未设置", file=sys.stderr)
        sys.exit(1)

    # 假设场景的用户输入——包含沪深北三个市场
    user_inputs = ["600519", "000001", "920186"]

    # 先拉可用品种列表，整个批次共用一份
    try:
        available_symbols = fetch_available_symbols()
    except Exception as e:
        print(f"致命错误: {e}", file=sys.stderr)
        sys.exit(1)

    for raw in user_inputs:
        try:
            base = normalize_input(raw)
        except ValueError as e:
            print(f"输入规范化失败: {e}", file=sys.stderr)
            sys.exit(1)

        # 在可用品种列表中匹配完整 symbol（TickDB 规范格式：代码.SH/.SZ/.BJ）
        candidates = [f"{base}.SH", f"{base}.SZ", f"{base}.BJ"]
        matched = None
        for c in candidates:
            if c in available_symbols:
                matched = c
                break

        if matched is None:
            print(
                f"错误: {raw}（规范化后: {base}）未在可用品种列表中找到匹配，"
                f"不发起 ticker 请求",
                file=sys.stderr,
            )
            sys.exit(1)

        try:
            fetch_ticker(matched)
        except Exception as e:
            print(f"ticker 校验失败: {e}", file=sys.stderr)
            sys.exit(1)

        print(f"{matched}: 三道闸门校验通过")


if __name__ == "__main__":
    main()

代码里每一个 sys.exit(1) 都是故意的。不是因为不想容错，而是因为在这个环节里，沉默比报错更危险。一个缺失的价格被默认成 0，或者一个错配的 symbol 被当成对的放过去，对下游的破坏远比一段脚本崩溃严重——脚本崩溃你至少会收到告警，数据污染你可能很久以后才发现。

四、请求前、请求中、请求后检查卡

这张清单可以截图保存，每次接新行情 API 或切换数据源时逐项打勾。

请求前

• [ ] 用户输入已去除空格、全角字符和不可见字符
• [ ] 用户输入已剥离字母前缀（如 sh600519 → 600519）
• [ ] 已通过可用品种列表端点（如 /v1/symbols/available）确认 symbol 存在
• [ ] 确认可用品种列表本身非空（空列表放行等于把所有输入判死）

请求中

• [ ] 请求设置了 timeout（单位秒，不是毫秒）
• [ ] HTTP 非 2xx 已处理，不会静默继续
• [ ] JSON 解析失败已捕获并退出

请求后

• [ ] 业务码已检查：1001（Key 无效）、1002（未提供 Key）、1004（权限不足）均阻断
• [ ] 非成功码按错误码分层处理，不混用处理逻辑
• [ ] 2002（品种不存在）已终止，不会回退到猜测的 symbol
• [ ] 3001（频率限制）已读取 Retry-After，不做死循环重试
• [ ] 返回 data[] 非空，空数组按失败处理
• [ ] 返回 symbol 与请求完全一致，不依赖数组下标对齐
• [ ] last_price 非 None，已用 Decimal 解析，且 is_finite() 为 True
• [ ] timestamp 非 None，类型为 int，不是 float 或 bool

五、不能从本文推出什么

• 本文的 symbol 校验方法以 TickDB 的端点设计为示例，但三层身份区分和三道闸门的思路是通用的工程方法。其他数据源可能使用纯数字、ISIN 或其他编码方案，symbol 规范应以各自文档为准。
• 代码中的价格字段是接口返回值的占位说明，不构成任何实时报价展示，也不是投资参考。
• 本文讨论的范围严格限定在 REST ticker 端点的 symbol 处理和响应校验，不可外推到 K 线、盘口、成交明细、WebSocket 推送、MCP 工具或 CLI 命令的行为——不同端点的 symbol 解析、错误码语义和返回结构可能不同，需各自独立验证。

你接行情 API 时，排查最久的问题出在 symbol 格式、timestamp 含义，还是字段类型不一致？评论区讲讲——格式随意，一个坑或一个你觉得“文档早该写清楚”的细节都行。

想复现本文的 symbol 校验流程，可查看 TickDB 文档中 /v1/symbols/available 和 /v1/market/ticker 端点的完整说明，或使用测试 Key 复现。

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

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