摘要

接入 A 股实时行情 API 时，维护成本往往不在第一行代码，而在后续扩展：ticker 和 kline 如果鉴权方式不同、时间戳精度不同、错误码语义不同，每多接一个端点就要多维护一套解析逻辑。本文以 TickDB 的 REST API 为可复核示例，展示一个 ticker 和 kline 在鉴权和字段类型上保持一致的 Python 实时行情数据源。附带可直接复用的校验骨架、参数对照表和首次成功检查卡。

1. 两个端点，一张对照表

你以为同一个数据源的不同端点应该一致，但实际上 ticker 可能正常返回、kline 却鉴权失败，或者两边都返回成功码，timestamp 却是不同精度。等到几十个品种的监控脚本跑起来、两边数据需要合到一起分析时，设计不一致的代价就会集中爆发。

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

下面这张表，既是功能对照，也是你评估任何数据源时可以直接拿来用的契约一致性检查起点。

2. 前置条件

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

export TICKDB_API_KEY="your_api_key_here"

3. 最小代码：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）。任何一步失败都非零退出，不补默认值。

4. 最小代码：历史 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——字段路径和语义必须分别核对。

5. 首次成功检查卡

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
• 任一校验失败时程序非零退出，不补默认值

6. 常见排错

7. TickDB 适合谁，让位给谁

适合的场景：

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

不适合的场景：

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

8. 本文没有证明什么

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

📡 本文行情数据示例由 TickDB.ai 提供，接口文档见 https://docs.tickdb.ai
⚠️ 本文为技术教程，不构成任何投资建议