A 股实时行情 API 2026:从参数到错误处理,一次跑通 ticker 查询

TickDB Research ·

调通了 API ≠ 拿到了正确的价格。

你以为"查一下贵州茅台现价"只是一行 HTTP GET。第一次请求返回 data: []——参数名写错了。修好后,价格字段是 null——取了 close 而不是 last_price。再修好,触发了限流——没有退避逻辑。最后终于通了,但你用 timestamp 判断"数据是否新鲜",在周末看到的价格其实是周五下午三点的收盘价。

四次失败,没有一次是网络不通。全是参数、字段和错误处理的细节。

下面用一张误写自查表、一段可复现的 Python 代码和一个决策框架,把这些坑一次性排干净。读完你能跑通一次正确的 A 股 ticker 查询,并清楚这次跑通能证明什么、不能证明什么。

一、6 个最容易写错的点(自查表)

常见误写正确写法后果排查优先级
symbol=600519.SHsymbols=600519.SH参数名错误,API 忽略并返回空数组🔴 第一优先
data["close"]data["last_price"]ticker 用 last_price,close 是 K 线字段🔴 第一优先
价格用 float 计算用 Decimal(str(value))浮点误差累积,回测结果失真🟡 第二优先
timestamp 当"延迟指标"timestamp 仅表示行情生成时刻(毫秒 UTC)休市时误判数据"不新鲜"🟡 第二优先
收到错误码后无限重试鉴权错误阻断,限流读 Retry-After 退避触发封禁或死循环🔴 第一优先
不设 timeoutrequests.get(..., timeout=10)网络抖动时线程永久阻塞🟢 第三优先

二、核心概念区分:接口延迟 ≠ 数据年龄

"A 股实时行情 API 能返回实时数据吗?"——这个问题本身把两个概念混淆了。

概念定义本文能证明什么本文不能证明什么
接口延迟从行情生成到数据到达你服务器的时间差本次请求成功拿到了一个响应不能证明这个延迟是多少,也没有 SLA 承诺
数据年龄当前时间与 timestamp 字段的差值你能算出这个差值,并据此判断数据"有多旧"不能把这个差值当作接口延迟;休市时年龄持续增长是正常的

类比:快递单上的"发货时间"告诉你包裹哪天寄出的,不告诉你快递在路上的速度。timestamp 就是行情数据的"发货时间"。数据年龄 = 你收到的时间 - 发货时间。接口延迟 = 快递实际的运输耗时——需要专门的测量工具和统计方法才能估算。

这个区分直接决定了你如何设计数据可用性判断:用数据年龄判断行情是否"过旧"是可操作的;用数据年龄反推接口延迟是不可靠的。

三、困境与破局:字段名一混,排查成本翻倍

开发者最常见的痛点不是接口不可用,而是字段语义在不同的端点间不一致。close 在 K 线里有值,在 ticker 快照里根本没有——代码不报错,价格永远为空。如果你的系统同时接入 A 股、港股、美股,每个数据源的 ticker 字段名可能不同,维护成本随接口数量线性增长。

此时,你需要的不是一个"更快的 API",而是一个消除了字段歧义的统一数据层——让 ticker 永远用 last_price、kline 永远用 close,无论查询哪个市场。

本文痛点TickDB 解决方式工程收益
ticker/kline 字段名混淆,取值总是 null按端点严格隔离字段语义:ticker 用 last_price,kline 用 close从源头杜绝字段名混淆,不需要记忆"哪个端点用哪个字段"
不同市场 symbol 格式不同,校验正则写一堆A 股 600519.SH、港股 700.HK、美股 AAPL.US 统一后缀规范减少沪深北/港股/美股后缀的分支判断,降低 symbol 校验代码复杂度
限流信号不一致客户端应同时兼容业务码 3001 和 HTTP 429,优先读取 Retry-After 并退避错误处理只需写一次,不因返回形式不同而漏掉限流信号
价格字符串直接转 float,精度丢失价格和成交量统一以字符串返回强制使用 Decimal,从接口设计层面避免精度陷阱

以上能力来自 TickDB 的统一行情数据层。接下来,用一段 Python 代码验证 A 股 ticker 查询的完整通路——这段代码已内置了上述所有正确写法。

image.png

四、最小教学示例:A 股 ticker 查询(Python)

安装依赖

pip install requests python-dotenv

代码:a_share_ticker.py

import os
import time
import json
import requests
from decimal import Decimal, InvalidOperation
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("TICKDB_API_KEY")
if not API_KEY:
    raise RuntimeError("环境变量 TICKDB_API_KEY 未设置")

BASE_URL = "https://api.tickdb.ai"
TIMEOUT = 10
MAX_RETRIES = 3

def get_ticker(symbols: list[str]) -> list[dict]:
    """
    查询 A 股实时行情快照。
    
    正确用法:
    - 参数 symbols(复数),逗号分隔,最多 50 个
    - 读取 last_price(不是 close)
    - timestamp 为毫秒 UTC 行情生成时间,不直接等于数据新鲜度
    - 价格使用 Decimal 避免浮点误差
    """
    url = f"{BASE_URL}/v1/market/ticker"
    headers = {"X-API-Key": API_KEY}
    params = {"symbols": ",".join(symbols)}

    for attempt in range(MAX_RETRIES + 1):
        try:
            resp = requests.get(url, headers=headers, params=params, timeout=TIMEOUT)
        except requests.exceptions.Timeout:
            if attempt == MAX_RETRIES:
                raise RuntimeError("请求超时,已达最大重试次数")
            time.sleep(2 ** attempt)
            continue
        except requests.exceptions.ConnectionError:
            raise RuntimeError("网络连接失败,请检查网络或 API 地址")

        # HTTP 级错误(非 2xx 且非 429)直接抛出
        if resp.status_code != 200 and resp.status_code != 429:
            raise RuntimeError(f"HTTP 错误: {resp.status_code}")

        # JSON 解析防御:非 JSON 错误响应不应导致程序中断
        try:
            data = resp.json()
        except json.JSONDecodeError:
            raise RuntimeError(f"API 返回非 JSON 响应: {resp.text[:200]}")

        code = data.get("code")

        if code == 0:
            # 成功:映射字段
            results = []
            for item in data.get("data", []):
                try:
                    price = Decimal(str(item["last_price"]))
                    volume = Decimal(str(item.get("volume_24h", "0")))
                except (KeyError, InvalidOperation, ValueError):
                    continue
                results.append({
                    "symbol": item["symbol"],
                    "last_price": str(price),
                    "volume_24h": str(volume),
                    "timestamp_ms": item["timestamp"]
                })
            if not results:
                raise RuntimeError("API 返回成功但 data 为空,请检查 symbol 是否正确")
            return results

        elif code == 3001 or resp.status_code == 429:
            # 限流:优先读 Retry-After 头,否则指数退避
            if attempt == MAX_RETRIES:
                raise RuntimeError("触发限流,已达最大重试次数")
            retry_after = resp.headers.get("Retry-After")
            wait = int(retry_after) if retry_after and retry_after.isdigit() else 2 ** attempt
            time.sleep(wait)
            continue

        elif code == 1001:
            # API Key 无效或过期:不重试,立即阻断
            raise PermissionError("API Key 无效或已过期(code=1001)")
        elif code == 1004:
            # 权限不足:套餐或访问范围不足,不重试
            raise PermissionError("API Key 权限不足,请检查套餐或访问范围(code=1004)")
        else:
            raise RuntimeError(f"API 错误: code={code}, message={data.get('message')}")

    raise RuntimeError("未知错误:重试耗尽")


if __name__ == "__main__":
    # 示例:查询贵州茅台、宁德时代和一只北交所股票
    # 注:北交所符号以 available 接口实时查询为准,此处 920832.BJ 为实测可返回样例
    tickers = get_ticker(["600519.SH", "300750.SZ", "920832.BJ"])
    for t in tickers:
        print(f"{t['symbol']}: last_price={t['last_price']}, timestamp_ms={t['timestamp_ms']}")

审核实测返回结构示例(字段结构示意,非实时价格)

{
  "code": 0,
  "message": "success",
  "data": [
    {
      "symbol": "600519.SH",
      "type": "stock",
      "last_price": "1682.50",
      "volume_24h": "3214567",
      "timestamp": 1717000000000
    }
  ]
}

五、这段代码的决策框架

代码不是"写对了就行",每个分支背后都有一个工程决策。理解这些决策,比记住代码更有价值。

错误场景处理策略工程原则
code == 1001(Key 无效/过期)立即阻断,不重试鉴权失败是永久性错误——重试不会修复 Key,只会浪费配额并可能触发安全告警
code == 1004(权限不足)立即阻断,不重试套餐或访问范围不足同样是永久性错误,需人工升级套餐而非代码重试
code == 3001 或 HTTP 429指数退避重试,优先读 Retry-After 头限流是临时性错误——服务端在告诉你"等 X 秒后再来"。遵守这个协议是生产环境的基本素养
请求超时有限重试,退避间隔随重试次数增加超时可能是临时网络抖动,也可能是服务端过载——盲目重试和永远不重试一样危险
data 数组为空抛出异常,不返回空列表静默返回空数据会让上游逻辑误判"一切正常",比抛异常更难排查
价格/成交量字段缺失或非法跳过该项,不整体失败部分数据损坏不应阻断其他 symbol 的正常返回——但也不应静默填充默认值

核心是区分"临时性错误"和"永久性错误",不是写一堆 if-else。 限流(3001)是临时的——服务端在告诉你"稍后再来"。鉴权失败(1001/1004)是永久的——不换 Key 或不升级套餐,永远过不去。把这两种错误混用同一套重试策略,是生产环境限流问题演变为封禁事故的根因。同理,1001(Key 无效)和 1004(权限不足)虽然都是阻断,但排查方向不同——一个是检查 Key 是否正确,一个是检查套餐是否覆盖当前接口。

六、本文能证明什么、不能证明什么

一次成功的 REST ticker 查询能证明三件事:Key 有效、symbol 存在、当前时刻能拿到结构化行情。但它不能证明以下任何一点:

  • ❌ 不能证明接口延迟:timestamp 是行情生成时间,不是端到端延迟指标。本次请求的成功不能外推为延迟承诺或 SLA。
  • ❌ 不能证明高频可用:REST 快照是离散采样,不是逐笔推送。如果你的场景需要连续行情更新,需要另行评估推送接口或实时通道。
  • ❌ 不能替代交易系统:本文只讨论行情数据获取,不涉及下单、撤单、仓位管理或风控逻辑。
  • ❌ 不能证明任何投资结论:所有价格均为假设案例,仅用于说明计算方法,不构成买卖建议或收益暗示。

一个你需要在架构层面想清楚的问题:你在周六上午用 ticker 查到了 600519.SH 的 last_price,timestamp 是周五 15:00:00。你的系统如何向用户呈现这个价格——是"最新价"、"收盘价"、还是"上一交易日收盘价"?这三种标注对交易决策的影响完全不同。欢迎在评论区写下你的标注逻辑。

更多接口细节和字段说明,见 docs.tickdb.ai。


查看相关行情工具

相关文章

适合开发者的实时行情 API 怎么选:REST、WebSocket、MCP 和七个工程边界

直接答案 选择实时行情 API 没有统一标准答案。正确的做法是走三步:先核对市场、粒度、数据时效和授权,再选择传输协议,最后用真实请求验证字段与时间戳。具体来说,需要逐条检查七项工程边界: 1. 目标市场与资产类别。 2. 快照、K线、逐笔、报价、订单簿等数据粒度。 3. REST、WebSocket、MCP、SDK、批量文件等接入方式。 4. symbol、字段名、timestamp 和时区

TickDB Research ·

行情 WebSocket 从握手到断线重连:symbol 格式不一致才是真正让你对不齐数据的元凶

> 凌晨三点,你的行情数据断了。重连代码在跑,心跳也在发,但系统空转了两个小时——心跳只发了 ping,没检查 pong。更隐蔽的是:品种目录返回的 symbol 是 ,REST 行情接口要的却是 。WebSocket 推送用的是哪种?你的同一套业务代码在处理同一只股票时,格式对不齐,数据关联静默失败——不是报错,而是你的系统里同一个股票在两条链路上用的是两个不同的代码。 一、那个被 Demo 藏

TickDB Research ·

Python 行情 WebSocket 从握手到断线重连:symbol 格式不一致才是真正让你对不齐数据的元凶

> 凌晨三点,你的行情数据断了。重连代码在跑,心跳也在发,但系统空转了两个小时——心跳只发了 ping,没检查 pong。更隐蔽的是:品种目录返回的 symbol 是 ,REST 行情接口要的却是 。WebSocket 推送用的是哪种?你的同一套业务代码在处理同一只股票时,格式对不齐,数据关联静默失败——不是报错,而是你的系统里同一个股票在两条链路上用的是两个不同的代码。 一、那个被 Demo 藏

TickDB Research ·

你跑的回测可信吗?——Tick级行情双源交叉验证的工程实践

> 同一只股票,同一时刻,两个主流数据源的价格差了0.01元,时间戳差了3毫秒,成交量差了200股。不是谁错了——是“精确到Tick”这件事本身,就没有统一标准。 一、一个你必须面对的事实 你用Tick级数据跑回测,夏普比率漂亮得让自己都有点不好意思。但切换到实盘,曲线开始变得陌生。你查过程序,没问题;查过网络,没问题;最后怀疑数据——但你无从查起,因为你手里只有这一份数据。 多数量化开发者默认“

TickDB Research ·

量化系统冷启动与实盘无缝接续:一套架构同时搞定 REST 历史回放与 WebSocket 实时推送

> 凌晨 3:12,量化系统从历史回放模式切换到实时模式。REST 最后一条 K 线的时间戳是 09:31:00.000,WebSocket 推送的第一条实时 Tick 是 09:31:00.287。这 287 毫秒的缝隙里,成交了 47 笔订单——你的策略一无所知,它还在等“下一条 K 线”。这 47 笔里有一笔是当天最高点。 > > 独立量化开发者的经典困境:回测阶段用 REST 批量拉历史

TickDB Research ·