Python 获取 A 股实时行情：股票代码怎么传、symbol 怎么校验、ticker 怎么查

Tangkengzai168

194人浏览 · 2026-06-11 10:04:11

Tangkengzai168 · 2026-06-11 10:04:11 发布

摘要

用 Python 获取 A 股实时行情 API 数据时，股票代码怎么传、symbol 参数怎么校验、ticker 查询返回后还要检查什么——这三个问题比选什么库更重要。本文以 TickDB 的 REST 接口为实操对象，给出一套可复现的接入流程：输入规范化（生成候选后缀，用可用品种列表确认）→ 可用品种列表校验 → ticker 查询 → 响应字段校验。代码覆盖超时、连接错误、业务码 1001/1002/1004/2002/3001，以及 last_price 的 Decimal 解析和 timestamp 的 13 位毫秒校验。缺失字段不会默认成 0，任何校验失败都会以非零码退出。

1. 开篇：股票代码传错了，后面的代码再漂亮也没用

如果你正在搜索"Python 获取 A 股实时行情"，大概率已经遇到过这种情况：

import requests
resp = requests.get(
    "https://api.tickdb.ai/v1/market/ticker",
    params={"symbols": "600519"},
    headers={"X-API-Key": "sk-xxx"},
)
print(resp.status_code)  # 200
print(resp.json())       # 可能返回空 data 或品种不存在类错误

HTTP 200，JSON 也解析成功了，但没有数据。网络没问题，Key 也没过期。问题出在一开始：你把 600519 传给了要求 600519.SH 的接口。 具体返回的错误码以当前接口实测为准——重点是认识到未规范化的输入不应该直接传入接口。

本文直接回答三个问题：股票代码怎么传、symbol 怎么校验、ticker 查询返回后还要检查什么。下面以 TickDB 的 REST ticker 和 symbols 查询接口为实操示例，给出完整可运行的代码。

2. 三层概念：用户输入、股票代码、API symbol

在工程中，一个 A 股品种标识会以三种形态出现。理解这个分层是写好校验逻辑的前提：

层次	示例	来源	接口能不能用
用户输入	`600519`、`贵州茅台`、`sh600519`	用户键入或语音输入	不能直接用
股票代码	`600519`	交易所或数据商定义	缺少交易所标识，接口不识别
API symbol	`600519.SH`	接口规范	这是接口真正接受的格式

核心认知：用户输入不能直接扔给接口。工程上需要一条转换链——先把用户输入规范化为候选 CODE.EXCHANGE 格式，再用可用品种列表确认哪个后缀是正确的，最后才发起 ticker 请求。

注意：本文示例代码只处理数字代码（如 600519）和已带后缀的 symbol（如 920186.BJ）。对于中文名称（如 贵州茅台）或拼音缩写等输入，需另接名称映射或搜索服务转换为标准 symbol，不在本示例覆盖范围内。

TickDB 为这条链提供了两个关键端点：GET /v1/symbols/available 用于请求前校验，GET /v1/market/ticker 用于获取快照及请求后校验。下面用完整代码串联这两个端点。

3. 完整代码：从输入规范化到响应字段校验

环境：Python 3.9+

pip install requests==2.31.0

设置环境变量（.env 或直接在终端 export，不要提交到版本控制）：

export TICKDB_API_KEY=your_api_key_here

完整代码：

import os
import sys
from decimal import Decimal, InvalidOperation
from http import HTTPStatus

import requests

# ---------------------------
# 配置
# ---------------------------
API_KEY = os.environ.get("TICKDB_API_KEY")
if not API_KEY:
    sys.exit("未设置环境变量 TICKDB_API_KEY，请先 export 或在 .env 中配置。")

BASE_URL = "https://api.tickdb.ai/v1"
HEADERS = {"X-API-Key": API_KEY}

# 用户输入的原始字符串（真实项目中来自用户键入或配置文件）
RAW_INPUTS = ["600519", "000001", "920186.BJ"]

# A 股可能的交易所后缀
A_SHARE_SUFFIXES = [".SH", ".SZ", ".BJ"]


# ---------------------------
# 步骤 1：输入规范化（生成候选 symbol）
# ---------------------------
def normalize_to_candidates(raw: str) -> list:
    """将用户输入规范化为候选 CODE.EXCHANGE 列表。
    若已有合法后缀则直接返回；若为裸代码则生成 .SH/.SZ/.BJ 三个候选。
    实际后缀以 /v1/symbols/available 的命中结果为准。"""
    raw = raw.strip().upper()
    if "." in raw:
        return [raw]
    return [f"{raw}{suffix}" for suffix in A_SHARE_SUFFIXES]


CANDIDATES = []
for r in RAW_INPUTS:
    CANDIDATES.extend(normalize_to_candidates(r))
print(f"候选 symbol: {CANDIDATES}")


# ---------------------------
# 步骤 2：获取可用品种列表
# ---------------------------
def fetch_available_symbols():
    """请求 /v1/symbols/available，返回可用 symbol 集合。
    这一步的作用是在请求 ticker 之前排除无效 symbol，
    避免浪费请求配额并得到更明确的错误信息。"""
    try:
        resp = requests.get(
            f"{BASE_URL}/symbols/available",
            headers=HEADERS,
            timeout=10,
        )
        if resp.status_code != HTTPStatus.OK:
            sys.exit(f"获取可用品种列表失败: HTTP {resp.status_code}")
        body = resp.json()
    except requests.Timeout:
        sys.exit("获取可用品种列表超时，请检查网络或增大 timeout。")
    except requests.ConnectionError:
        sys.exit("获取可用品种列表失败: 网络连接错误，请检查网络和端点地址。")
    except requests.RequestException as e:
        sys.exit(f"获取可用品种列表失败: {e}")
    except ValueError:
        sys.exit("获取可用品种列表失败: 响应非 JSON，可能返回了 HTML 错误页。")

    code = body.get("code")
    if code != 0:
        if code == 1001:
            sys.exit("可用品种列表接口错误: API Key 无效或已过期 (1001)。")
        if code == 1002:
            sys.exit("可用品种列表接口错误: 未提供 API Key (1002)。")
        if code == 1004:
            sys.exit("可用品种列表接口错误: 权限或访问范围不足 (1004)。")
        sys.exit(f"可用品种列表接口业务错误: code={code}")

    products = body.get("data", {}).get("products", [])
    if not isinstance(products, list) or len(products) == 0:
        sys.exit("可用品种列表为空，请检查 API Key 权限或接口返回结构。")

    available = set()
    for item in products:
        if not isinstance(item, dict):
            continue
        symbol = item.get("symbol")
        if isinstance(symbol, str) and symbol.strip():
            available.add(symbol.strip())
    if not available:
        sys.exit("可用品种列表解析后为空，请检查接口返回结构。")
    print(f"可用品种总数: {len(available)}")
    return available


AVAILABLE_SYMBOLS = fetch_available_symbols()

# ---------------------------
# 步骤 3：用可用品种列表校验候选 symbol
# ---------------------------
VALID_SYMBOLS = []
for sym in CANDIDATES:
    if sym in AVAILABLE_SYMBOLS:
        VALID_SYMBOLS.append(sym)
    else:
        print(f"信息: {sym} 不在可用品种列表中，已跳过。")

if not VALID_SYMBOLS:
    sys.exit("没有有效的 symbol，流程终止。请检查输入或 API Key 的权限范围。")

print(f"校验通过的 symbol: {VALID_SYMBOLS}")


# ---------------------------
# 步骤 4：请求 ticker 快照
# ---------------------------
def fetch_ticker(symbols: list):
    """请求 /v1/market/ticker，返回 data 列表。
    所有异常路径均以 sys.exit 终止，不返回 None 或空列表——
    避免调用方拿到假数据继续计算。"""
    try:
        resp = requests.get(
            f"{BASE_URL}/market/ticker",
            params={"symbols": ",".join(symbols)},
            headers=HEADERS,
            timeout=10,
        )
        if resp.status_code != HTTPStatus.OK:
            sys.exit(f"Ticker 请求失败: HTTP {resp.status_code}")
        body = resp.json()
    except requests.Timeout:
        sys.exit("Ticker 请求超时，请检查网络或增大 timeout。")
    except requests.ConnectionError:
        sys.exit("Ticker 请求失败: 网络连接错误，请检查网络和端点地址。")
    except requests.RequestException as e:
        sys.exit(f"Ticker 请求失败: {e}")
    except ValueError:
        sys.exit("Ticker 响应非 JSON，可能返回了 HTML 错误页。")

    code = body.get("code")
    if code == 0:
        return body.get("data", [])
    # 以下错误码来自 TickDB REST ticker 接口文档
    if code == 1001:
        sys.exit("业务错误: API Key 无效或已过期 (1001)，请检查 Key。")
    if code == 1002:
        sys.exit("业务错误: 未提供 API Key (1002)，请检查 Header。")
    if code == 1004:
        sys.exit("业务错误: 权限或访问范围不足 (1004)，请确认 Key 的授权范围。")
    if code == 2002:
        sys.exit(
            "业务错误: 交易品种不存在 (2002)。"
            "建议调用 /v1/symbols/available 查询可用品种列表。"
        )
    if code == 3001:
        retry_after = resp.headers.get("Retry-After", "未知")
        sys.exit(
            f"业务错误: 请求频率超限 (3001)。"
            f"Retry-After: {retry_after}。请等待后重试，不要死循环。"
        )
    sys.exit(f"业务错误: 未知错误码 {code}")


data = fetch_ticker(VALID_SYMBOLS)

# ---------------------------
# 步骤 5：响应字段校验
# ---------------------------
def validate_records(data, expected_symbols):
    """对实际使用的字段逐条校验。校验失败立即退出——
    在行情数据管线中，一条脏数据的排查成本远高于在入口处拦截。"""
    expected_upper = {s.upper() for s in expected_symbols}

    if not isinstance(data, list) or len(data) == 0:
        sys.exit("校验失败: data 为空，请结合 symbol、权限、接口当前行为排查。")

    if len(data) != len(expected_symbols):
        sys.exit(
            f"校验失败: 返回 {len(data)} 条，期望 {len(expected_symbols)} 条。"
        )

    returned = []
    for i, item in enumerate(data):
        if not isinstance(item, dict):
            sys.exit(f"校验失败: data[{i}] 不是字典，类型: {type(item)}。")

        # symbol：必须是非空字符串，且属于目标品种
        symbol = item.get("symbol")
        if not isinstance(symbol, str) or not symbol.strip():
            sys.exit(f"校验失败: data[{i}] symbol 缺失或非字符串。")
        if symbol.upper() not in expected_upper:
            sys.exit(f"校验失败: 返回了非预期的 symbol {symbol}。")
        returned.append(symbol.upper())

        # last_price：必须是非空字符串，可解析为 Decimal，且非 NaN/Infinity
        # 缺失字段不能默认成 0——默认值会掩盖数据缺失
        last_price = item.get("last_price")
        if not isinstance(last_price, str) or not last_price.strip():
            sys.exit(
                f"校验失败: {symbol} last_price 缺失或非字符串，不能默认成 0。"
            )
        try:
            d = Decimal(last_price)
            if not d.is_finite():
                sys.exit(f"校验失败: {symbol} last_price 为 NaN/Infinity。")
        except (InvalidOperation, ValueError):
            sys.exit(
                f"校验失败: {symbol} last_price 无法解析为 Decimal: {last_price}"
            )

        # timestamp：必须是 int 且不是 bool；必须是 13 位毫秒
        # 毫秒 UTC 是字段精度，不等于延迟、新鲜度、采样频率或 SLA
        # 不是 13 位毫秒不能当作正常结果继续处理——时间单位混用会导致策略信号错位
        ts = item.get("timestamp")
        if not isinstance(ts, int) or isinstance(ts, bool):
            sys.exit(
                f"校验失败: {symbol} timestamp 不是整数，"
                f"类型: {type(ts).__name__}。"
            )
        if ts < 1000000000000 or ts > 9999999999999:
            sys.exit(
                f"校验失败: {symbol} timestamp 不是 13 位毫秒: {ts}。"
                f"不是 13 位毫秒不能当作正常结果继续处理。"
            )

    # 重复检查：防止接口返回了两次同一品种
    if len(set(returned)) != len(returned):
        sys.exit("校验失败: data 中存在重复 symbol。")

    # 缺失检查：请求了但没返回
    missing = expected_upper - set(returned)
    if missing:
        sys.exit(f"校验失败: 缺失品种: {', '.join(sorted(missing))}。")


validate_records(data, VALID_SYMBOLS)

# ---------------------------
# 步骤 6：校验全部通过后使用数据
# ---------------------------
print("全部校验通过，以下是实时行情：")
for item in data:
    price = Decimal(item["last_price"])
    print(f"  {item['symbol']}: {price} (时间戳: {item['timestamp']})")

4. 排错表：请求前、请求中、请求后分别检查什么

这张表覆盖了从输入到最终使用的完整链路。建议保存为独立参考卡片。

阶段	检查项	常见现象	排查方向
请求前	输入规范化	裸代码直接传入接口，可能返回空 data 或品种不存在类错误	生成 `.SH/.SZ/.BJ` 候选，用可用品种列表确认真实后缀
请求前	可用品种校验	所有候选 symbol 都不在列表中	检查 symbol 格式、确认 API Key 权限覆盖范围
请求中	HTTP 状态码	超时、连接拒绝、非 200	检查网络连通性、防火墙规则、端点地址是否正确
请求中	业务码 2002	返回 `code: 2002`	品种不存在，回到可用品种列表重新校验 symbol
请求中	业务码 3001	返回 `code: 3001`	频率超限，读 `Retry-After` 响应头，等待后重试，不要死循环
请求中	业务码 1001/1002/1004	返回对应错误码	分别检查 Key 有效性、Header 携带、权限范围
请求后	data 为空	`"data": []`	结合 symbol、权限、接口当前行为排查
请求后	symbol 缺失/多余	返回品种与请求不一致	可能是接口行为变更或 Key 权限变化，记录日志并退出
请求后	last_price 解析失败	值为 `None`、空字符串、`"NaN"`	数据异常，不能默认成 0，应退出并告警
请求后	timestamp 非 13 位毫秒	值为 10 位秒级或字符串	时间单位混用会导致策略信号错位，必须拒绝

5. 常见错误码速查

以下错误码来自 TickDB REST ticker 接口文档。每个 code 对应的处理方向是固定的，但实际返回以接口当前行为为准：

code	含义	处理方向
`1001`	API Key 无效或已过期	检查 Key 是否正确，是否已续期
`1002`	未提供 API Key	检查 Header 中是否携带 `X-API-Key`
`1004`	权限或访问范围不足	确认当前 Key 是否有该接口或品种的访问权限
`2002`	交易品种不存在	调用 `/v1/symbols/available` 查询可用品种列表，确认 symbol 格式
`3001`	请求频率超限	读取 `Retry-After` 响应头，等待指定秒数后重试，不得死循环

6. 为什么本文用 TickDB 做实操示例

选 TickDB 是因为它具备三个对本文教学流程有直接帮助的特性——这三个特性恰好对应本文的三个核心步骤：

Symbol Query 接口（/v1/symbols/available）→ 对应步骤 2-3：在请求 ticker 之前校验品种是否存在，避免无效请求消耗配额。
REST ticker 接口（/v1/market/ticker）→ 对应步骤 4-5：返回 data[]、symbol、last_price、timestamp，错误码分层清晰，便于做字段级校验。
公开文档和 GitHub 示例 → 对应复现实验：文档见 https://docs.tickdb.ai，GitHub 示例见 https://github.com/TickDB/tickdb-unified-realtime-marketdata-api。