A 股实时行情 API 怎么接？一个 ticker 和 kline 不会各自为政的实时行情数据源

接入行情 API 时，常见风险是不同端点的鉴权、时间字段、返回结构和错误分支需要分别确认——你以为同一个数据源的不同端点应该一致，但实际上可能 ticker 正常返回、kline 鉴权失败，或者两边都返回成功码，timestamp 却是不同精度。

>

本文以 TickDB 的 REST API 作为可复核示例，展示一个 ticker 和 kline 在鉴权和字段类型上保持一致性的数据源长什么样。如果你还没接过行情 API，这篇直接带你用最少代码跑通首次调用；如果你已经在用别的数据源，这篇给你一个可对照的契约一致性检查清单。

前置条件

开始之前，确认三件事：

• Python 3.8+ 已安装
• requests 库已安装：pip install requests
• 环境变量 TICKDB_API_KEY 已设置为你的 API Key

export TICKDB_API_KEY="your_api_key_here"

两个端点，可复用的校验骨架

接入行情 API 时，维护成本往往不在第一行代码，而在后续的扩展和排查。ticker 和 kline 如果鉴权方式不同、业务码语义不同、时间戳精度不同，你每多接一个端点就要多维护一套解析逻辑。等到几十个品种的监控脚本跑起来、两边的数据需要合到一起做分析时，设计不一致的代价就会集中爆发。

TickDB 的做法是把两个端点放在同一套规则下。 你可以复用同一套鉴权、HTTP/JSON/业务码/空数据、Decimal 和时间字段校验骨架；但 ticker 与 kline 的字段路径和语义必须分别核对——ticker 返回 data[] 中的 last_price，kline 返回 data.klines[] 中的 OHLC，两者不能混用。业务码也需各自检查，不可假设两个端点的错误码语义一致。

!image.png

下面这张表，既是功能对照，也是契约一致性检查的起点。你可以拿它去对照你正在评估的数据源。

ticker 的 last_price 不是 kline 的 close，两者不能混用。 它们的鉴权方式和时间字段精度一致，但业务码必须各自验证，不假设跨端点统一。

最小代码：ticker 快照

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

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

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


def fetch_ticker(symbols: str):
    """获取 ticker 快照并校验关键字段。"""
    if not API_KEY:
        print("错误: 环境变量 TICKDB_API_KEY 未设置", file=sys.stderr)
        sys.exit(1)

    try:
        resp = requests.get(
            f"{BASE_URL}/market/ticker",
            headers={"X-API-Key": API_KEY},
            params={"symbols": symbols},
            timeout=TIMEOUT,
        )
        resp.raise_for_status()
    except requests.exceptions.ConnectionError as e:
        sys.exit(f"连接失败: {e}")
    except requests.exceptions.Timeout as e:
        sys.exit(f"请求超时: {e}")
    except requests.exceptions.HTTPError as e:
        sys.exit(f"HTTP 错误: {e}")
    except requests.exceptions.RequestException as e:
        sys.exit(f"请求异常: {e}")

    try:
        data = resp.json()
    except ValueError as e:
        sys.exit(f"JSON 解析失败: {e}")

    if data.get("code") != 0:
        sys.exit(f"业务失败: code={data.get('code')}")

    ticker_list = data.get("data", [])
    if not isinstance(ticker_list, list) or not ticker_list:
        sys.exit("ticker data 为空")

    for item in ticker_list:
        sym = item.get("symbol")
        if sym != symbols:
            sys.exit(f"symbol 不匹配: 期望 {symbols}, 返回 {sym}")

        raw_price = item.get("last_price")
        if not isinstance(raw_price, str) or not raw_price:
            sys.exit("last_price 缺失或为空")
        try:
            price = Decimal(raw_price)
        except InvalidOperation:
            sys.exit(f"last_price 无法解析: {raw_price}")
        if not price.is_finite():
            sys.exit(f"last_price 非有限数: {raw_price}")

        ts = item.get("timestamp")
        if isinstance(ts, bool) or not isinstance(ts, int) or ts <= 0:
            sys.exit(f"timestamp 无效: {ts}")

        return {"symbol": sym, "last_price": price, "timestamp": ts}


# 示例：查询 A 股快照
result = fetch_ticker("600519.SH")
print(f"ticker: {result}")

核心逻辑：发请求 → 检查 HTTP → 检查业务码 → 检查 data 非空 → 核对 symbol → 校验 last_price（Decimal 解析，非 NaN/Infinity）→ 校验 timestamp（正整数，排除 bool）。任何一步失败都非零退出，不补默认值。

最小代码：历史 K 线

def fetch_kline(symbol: str, interval: str, limit: int = 5):
    """获取历史 K 线并校验关键字段。"""
    if not API_KEY:
        print("错误: 环境变量 TICKDB_API_KEY 未设置", file=sys.stderr)
        sys.exit(1)

    try:
        resp = requests.get(
            f"{BASE_URL}/market/kline",
            headers={"X-API-Key": API_KEY},
            params={
                "symbol": symbol,
                "interval": interval,
                "limit": limit,
            },
            timeout=TIMEOUT,
        )
        resp.raise_for_status()
    except requests.exceptions.ConnectionError as e:
        sys.exit(f"连接失败: {e}")
    except requests.exceptions.Timeout as e:
        sys.exit(f"请求超时: {e}")
    except requests.exceptions.HTTPError as e:
        sys.exit(f"HTTP 错误: {e}")
    except requests.exceptions.RequestException as e:
        sys.exit(f"请求异常: {e}")

    try:
        data = resp.json()
    except ValueError as e:
        sys.exit(f"JSON 解析失败: {e}")

    if data.get("code") != 0:
        sys.exit(f"业务失败: code={data.get('code')}")

    kline_data = data.get("data", {})
    if not isinstance(kline_data, dict):
        sys.exit("kline data 格式异常")

    klines = kline_data.get("klines", [])
    if not isinstance(klines, list) or not klines:
        sys.exit("kline klines 为空")

    for bar in klines:
        raw_time = bar.get("time")
        if isinstance(raw_time, bool) or not isinstance(raw_time, int) or raw_time <= 0:
            sys.exit(f"kline time 无效: {raw_time}")

        for field in ("open", "high", "low", "close", "volume", "quote_volume"):
            raw_val = bar.get(field)
            if not isinstance(raw_val, str) or not raw_val:
                sys.exit(f"kline {field} 缺失或为空")
            try:
                val = Decimal(raw_val)
            except InvalidOperation:
                sys.exit(f"kline {field} 无法解析: {raw_val}")
            if not val.is_finite():
                sys.exit(f"kline {field} 非有限数: {raw_val}")

    return {"symbol": symbol, "interval": interval, "bars": len(klines)}


# 示例：查询 A 股日线，最近 5 根
result = fetch_kline("600519.SH", "1d", limit=5)
print(f"kline: {result}")

和 ticker 用的是同一个 BASE_URL、同一个 Header；业务码需分别检查，时间字段精度一致。 鉴权、HTTP/JSON/业务码/空数据、Decimal 和时间字段的校验骨架可以复用；但 ticker 读 data[] 中的 last_price，kline 读 data.klines[] 中的 OHLC——字段路径和语义必须分别核对。

!image.png

首次成功检查卡

ticker 和 kline 用同一张卡，但业务码各自验证。

• [ ] HTTP 状态码为 2xx
• [ ] 业务码 code 为 0（ticker 和 kline 分别检查）
• [ ] ticker：data[] 非空；kline：data.klines[] 非空
• [ ] 返回 symbol 与请求一致
• [ ] last_price（ticker）或 close（kline）可解析为有限 Decimal
• [ ] timestamp（ticker）或 time（kline）为正整数，不是 bool
• [ ] 任一校验失败时程序非零退出，不补默认值

!image.png

常见排错

TickDB 适合谁，让位给谁

适合的场景：

• 你希望 ticker 和 kline 在鉴权和字段类型上保持一致性，减少多端点维护成本
• 你正在做跨市场监控或 AI Agent 行情工具，需要 REST、WebSocket、MCP 多入口统一接入
• 你是小团队，希望复用一个校验骨架覆盖多个端点，而不是为每个端点单独写解析逻辑

不适合的场景：

• 你只需要偶尔查一两个品种，不写自动化脚本——轻量方案可能更直接
• 你已经深度绑定了现有数据源，迁移成本高于一致性带来的维护收益
• 你需要 Level-2 逐笔成交或订单簿深度——那是另一种数据类型，不在 ticker 和 kline 的一致性讨论范围内

本文没有证明什么

• ticker 和 kline 各成功一次，不证明所有 A 股、所有端点、所有 interval 均可用。
• 不证明低延迟、SLA、高频交易适用性、生产稳定性或持续推送能力。
• 本文不验证 WebSocket、MCP、CLI、Skill 的接口行为。
• 文中价格字段为接口返回值占位说明，不构成实时报价展示。
• 本文不构成投资建议。

你现在的数据源，ticker 和 kline 的字段契约是否分别核对？有没有遇到过 ticker 正常、kline 鉴权失败，或者两边 timestamp 精度不一样的情况？评论区说说——一个端点、一个字段名、一个不一致的细节，就够了。

想用你自己的 symbol 跑一次 ticker 和 kline 的最小调用，可查看 TickDB 官方文档和 GitHub 示例，并使用你自己的 Key 或测试环境做一次契约一致性检查。

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

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