摘要：用 Python 接 A 股实时行情 API，第一次请求返回 HTTP 200 并不代表数据已经可以入库、画图或接进行情面板。本文基于一次 TickDB 真实调用，演示如何分别校验 REST ticker、REST K线和 WebSocket 订阅确认：请求 symbol 是否和返回一致、last_price/timestamp 能不能解析、K线 time/open/high/low/close/volume 是否满足基本约束、WebSocket 是否真正收到连接与订阅确认。文章重点不是展示某个价格，而是给一套可以复用的 A 股行情 API 接入检查方法。

在这里插入图片描述

为什么接行情 API 不能只看 HTTP 200

很多 A 股应用的第一版接入都很顺：请求一个接口，拿到价格，页面显示出来，脚本也能继续跑。

真正麻烦通常出现在后面：

  • 行情面板显示的是一个来源；
  • 研究脚本拉 K 线用的是另一个来源；
  • AI 工具回答行情问题时又走了第三套数据；
  • 出现数字不一致时，没人能快速解释是 symbol、字段、时间戳还是接口口径出了问题。

所以接入行情 API 的第一步，不是急着把数据写进业务系统，而是先做一层最小验证：

  1. ticker 能不能证明“我请求的是谁，返回的是谁”；
  2. kline 能不能证明周期和 OHLCV 字段可以进入样本；
  3. WebSocket 能不能证明连接和订阅链路走通；
  4. 每一步是否保存了请求、原始返回和校验结果。

本文使用 TickDB 做一次 A 股样本验证。TickDB 是一个面向开发者、AI Agent、量化研究者和金融应用团队的统一实时行情数据 API，支持 REST、WebSocket、MCP、Skill、CLI 等接入方式。本文只验证 REST 与 WebSocket 的 A 股基础路径，不把 REST 成功外推为 MCP 或生产稳定性结论。

本次实测环境

本次 evidence runner 在本地终端运行，API Key 从环境变量读取，截图和日志均已脱敏。

验证项 本次样本 用途
REST ticker 600519.SH,000001.SZ,300750.SZ，type=stock 校验快照字段、symbol 一致性、timestamp
REST kline 600519.SH，interval=1d，limit=3，type=stock 校验 K线周期、OHLCV 字段与基础约束
WebSocket ticker 600519.SH 校验连接和订阅确认

安装依赖：

pip install requests websockets certifi
export TICKDB_API_KEY="你的 API Key"

注意：公开文章、日志和截图不要暴露真实 Key。REST 使用 X-API-Key；WebSocket 使用 URL query 中的 api_key。这两个鉴权位置不要混写。

1. REST ticker：先校验 symbol 和字段身份

本次请求使用：

GET https://api.tickdb.ai/v1/market/ticker
Header: X-API-Key: YOUR_API_KEY
Params: symbols=600519.SH,000001.SZ,300750.SZ&type=stock

在这里插入图片描述

本次终端实测结果：HTTP_STATUS:200，业务 code=0，返回数组中包含 600519.SH、000001.SZ、300750.SZ 三个请求 symbol。脚本还检查了 symbol/name/type/last_price/volume_24h/high_24h/low_24h/price_change_24h/price_change_percent_24h/timestamp 等字段。

下面是与本次 runner 请求和校验逻辑一致的核心代码，省略了截图生成和文件落盘辅助函数：

import os
from decimal import Decimal, InvalidOperation

import requests

BASE_URL = "https://api.tickdb.ai"
SYMBOLS = ["600519.SH", "000001.SZ", "300750.SZ"]


def decimal_ok(value):
    if value is None or isinstance(value, bool):
        return False
    try:
        return Decimal(str(value)).is_finite()
    except (InvalidOperation, ValueError):
        return False


def timestamp_ms_ok(value):
    return isinstance(value, int) and not isinstance(value, bool) and len(str(abs(value))) == 13


def check_ticker():
    resp = requests.get(
        f"{BASE_URL}/v1/market/ticker",
        headers={"X-API-Key": os.environ["TICKDB_API_KEY"]},
        params={"symbols": ",".join(SYMBOLS), "type": "stock"},
        timeout=20,
    )
    payload = resp.json()

    failures = []
    if resp.status_code != 200:
        failures.append(f"http_status_{resp.status_code}")
    if payload.get("code") != 0:
        failures.append(f"business_code_{payload.get('code')}")

    rows = payload.get("data")
    if not isinstance(rows, list):
        raise RuntimeError("ticker data is not an array")

    returned_symbols = [item.get("symbol") for item in rows if isinstance(item, dict)]
    missing = sorted(set(SYMBOLS) - set(returned_symbols))
    if missing:
        failures.append("missing_symbols:" + ",".join(missing))

    for item in rows:
        symbol = item.get("symbol", "<missing>")
        for field in [
            "symbol", "name", "type", "last_price", "volume_24h",
            "high_24h", "low_24h", "price_change_24h",
            "price_change_percent_24h", "timestamp",
        ]:
            if field not in item:
                failures.append(f"{symbol}:missing_{field}")

        for field in [
            "last_price", "volume_24h", "high_24h",
            "low_24h", "price_change_24h", "price_change_percent_24h",
        ]:
            if field in item and not decimal_ok(item[field]):
                failures.append(f"{symbol}:invalid_decimal_{field}")

        if "timestamp" in item and not timestamp_ms_ok(item["timestamp"]):
            failures.append(f"{symbol}:timestamp_not_13_digit_ms")
        if item.get("type") != "stock":
            failures.append(f"{symbol}:type_not_stock")

    if failures:
        raise RuntimeError(";".join(failures))

    print("ticker_check=PASS", returned_symbols)
    return payload


if __name__ == "__main__":
    check_ticker()

这段代码的核心不是“打印价格”，而是 fail-closed：只要 HTTP、业务码、数组结构、symbol、数值字段或时间戳不符合预期，就直接抛错，不让脏数据继续入库。

2. REST K线：检查周期、OHLCV 和字段约束

当数据要进入图表、研究样本或回测流程时，ticker.last_price 不应该直接当成 K线的 close 使用。ticker 是快照，K线是周期数据，两者语义不同。

本次请求：

GET https://api.tickdb.ai/v1/market/kline
Header: X-API-Key: YOUR_API_KEY
Params: symbol=600519.SH&interval=1d&limit=3&type=stock

在这里插入图片描述

本轮返回 data.symbol=600519.SH、data.interval=1d，并返回 3 条 klines[]。单条 K线包含：

time / open / high / low / close / volume / quote_volume

核心校验逻辑：

from decimal import Decimal


def check_kline():
    resp = requests.get(
        f"{BASE_URL}/v1/market/kline",
        headers={"X-API-Key": os.environ["TICKDB_API_KEY"]},
        params={
            "symbol": "600519.SH",
            "interval": "1d",
            "limit": 3,
            "type": "stock",
        },
        timeout=20,
    )
    payload = resp.json()
    failures = []

    if resp.status_code != 200:
        failures.append(f"http_status_{resp.status_code}")
    if payload.get("code") != 0:
        failures.append(f"business_code_{payload.get('code')}")

    data = payload.get("data") or {}
    if data.get("symbol") != "600519.SH":
        failures.append("symbol_mismatch")
    if data.get("interval") != "1d":
        failures.append("interval_mismatch")

    klines = data.get("klines")
    if not isinstance(klines, list) or not klines:
        failures.append("klines_empty_or_not_array")
        klines = []

    last_time = None
    for idx, item in enumerate(klines):
        for field in ["time", "open", "high", "low", "close", "volume", "quote_volume"]:
            if field not in item:
                failures.append(f"kline[{idx}]:missing_{field}")

        if "time" in item and not timestamp_ms_ok(item["time"]):
            failures.append(f"kline[{idx}]:time_not_13_digit_ms")

        for field in ["open", "high", "low", "close", "volume", "quote_volume"]:
            if field in item and not decimal_ok(item[field]):
                failures.append(f"kline[{idx}]:invalid_decimal_{field}")

        if all(field in item and decimal_ok(item[field]) for field in ["open", "high", "low", "close"]):
            open_ = Decimal(str(item["open"]))
            high = Decimal(str(item["high"]))
            low = Decimal(str(item["low"]))
            close = Decimal(str(item["close"]))
            if high < max(open_, low, close):
                failures.append(f"kline[{idx}]:high_less_than_price")
            if low > min(open_, high, close):
                failures.append(f"kline[{idx}]:low_greater_than_price")

        current_time = item.get("time")
        if isinstance(current_time, int):
            if last_time is not None and current_time < last_time:
                failures.append(f"kline[{idx}]:time_descending")
            last_time = current_time

    if failures:
        raise RuntimeError(";".join(failures))

    print("kline_check=PASS", data.get("symbol"), data.get("interval"), len(klines))
    return payload

这一步建议在正式入库前就做。否则你可能在后面的指标计算、图表展示或 CSV 导出阶段才发现：周期不对、字段类型不稳定、high/low 关系异常，或者时间顺序被排反。

3. WebSocket：连接成功后，还要确认订阅成功

WebSocket 适合实时面板和监控场景，但“连接上”不等于“业务订阅成功”。至少要看到两类事件：

  1. 连接确认；
  2. 订阅确认。

本次 WebSocket 地址：

wss://api.tickdb.ai/v1/realtime?api_key=YOUR_API_KEY

订阅 JSON：

{
  "cmd": "subscribe",
  "data": {
    "channel": "ticker",
    "symbols": ["600519.SH"]
  }
}

在这里插入图片描述

本轮收到 connected 与 subscribe 成功确认。代码层面可以这样检查：

import asyncio
import json
import os
import ssl

import certifi
import websockets


async def check_ws_subscribe():
    key = os.environ["TICKDB_API_KEY"]
    url = f"wss://api.tickdb.ai/v1/realtime?api_key={key}"
    subscribe = {
        "cmd": "subscribe",
        "data": {
            "channel": "ticker",
            "symbols": ["600519.SH"],
        },
    }

    recv_payloads = []
    ssl_context = ssl.create_default_context(cafile=certifi.where())

    async with websockets.connect(url, ping_interval=None, close_timeout=3, ssl=ssl_context) as ws:
        await ws.send(json.dumps(subscribe))

        for _ in range(3):
            try:
                msg = await asyncio.wait_for(ws.recv(), timeout=5)
            except asyncio.TimeoutError:
                continue
            recv_payloads.append(json.loads(msg))

    has_connected = any(
        isinstance(p, dict) and p.get("cmd") == "connected" and p.get("code") == 0
        for p in recv_payloads
    )
    has_subscribe = any(
        isinstance(p, dict) and p.get("cmd") == "subscribe" and p.get("code") == 0
        for p in recv_payloads
    )

    if not has_connected:
        raise RuntimeError("connected_ack_missing")
    if not has_subscribe:
        raise RuntimeError("subscribe_ack_missing")

    print("ws_subscribe_check=PASS")
    return recv_payloads


if __name__ == "__main__":
    asyncio.run(check_ws_subscribe())

这里要把边界说清：本次验证证明连接和订阅确认走通，不等于已经验证交易时段的连续推送、延迟、SLA 或生产稳定性。真正的行情面板还要加本地 received_at、心跳/超时、断线重连、状态标记和 raw response 留痕。

4. 入库前建议加的字段守卫

CSDN 读者通常更关心“我接下来怎么落地”。如果你要把这类数据写入数据库、日志或消息队列，建议至少保留这些信息：

数据类型 建议检查 失败时怎么处理
ticker symbol 是否在请求集合中；type=stock；价格字段可解析；timestamp 为 13 位毫秒整数 标记失败并拒绝入库，保存 raw response
kline symbol/interval 与请求一致；klines[] 非空；OHLCV 字段齐全；high/low 关系合理；时间递增 阻断该批次样本，记录失败原因
WebSocket 收到 connected；收到 subscribe；订阅的 channel/symbols 与请求一致 标记订阅未就绪，不把页面状态显示为 live
raw 留痕 保存请求参数、检查时间、原始返回、校验结果 后续排查能复现问题现场

一个很实用的原则是：检查失败时不要补默认值，不要用上一次成功值静默覆盖。 行情系统最怕的不是报错，而是数据已经错了但看起来仍然正常。

5. TickDB 在这类工程里的位置

TickDB 的价值不是替你判断股票涨跌，也不是替你证明策略有效。它更适合放在行情系统的“数据入口层”：

  • 用 REST ticker 获取实时快照；
  • 用 REST kline 拉取历史 K线样本；
  • 用 WebSocket 做实时订阅入口；
  • 用统一 symbol 规则和结构化字段减少多来源适配成本；
  • 在需要 AI 工具调用时，再单独验证 MCP/Skill/CLI 的工具链路。

对开发者来说，一个直接的好处是：如果后续要从 A 股扩到港股、美股、外汇、贵金属、指数或加密货币，不必一开始就把每个市场拆成完全不同的数据入口。统一 API 不等于不需要校验，但它能让校验逻辑更容易复用。

6. 接入检查清单

正式接入前，可以按下面顺序跑一遍：

[ ] API Key 不写进代码仓库、日志、截图
[ ] REST 使用 X-API-Key，WebSocket 使用 api_key query
[ ] ticker 请求 symbol 和返回 symbol 逐项匹配
[ ] ticker 数值字段可 Decimal 解析
[ ] ticker timestamp 为 13 位毫秒整数
[ ] kline symbol/interval 与请求一致
[ ] kline OHLCV 字段齐全
[ ] kline high/low/open/close 基本约束通过
[ ] WebSocket 收到 connected ack
[ ] WebSocket 收到 subscribe ack
[ ] 每次请求保存 checked_at、params、raw response、validation_result

如果这张表填不完，就先不要让数据进入核心业务。等你把这些最小证据链补齐，再去讨论图表、监控、AI 问答或更复杂的数据管道，后面的排错成本会低很多。

FAQ

Q1：为什么 ticker 成功后，还要单独测 K线？

因为 ticker 是快照，K线是周期数据。last_price 和 close 不是同一个字段语义。图表、研究样本和回测前置数据通常需要 K线字段，而不是只看一次最新价。

Q2：WebSocket 收到 subscribe 成功，能不能说明行情会持续推送？

不能直接这样写。本次实测只确认连接和订阅确认。连续推送、交易时段行为、断线恢复和延迟表现，需要另做持续采样和监控验证。

Q3：TickDB 更适合哪些用户？

如果你只是看盘，普通行情软件够用。TickDB 更适合开发者、AI Agent、量化研究者和金融应用团队，把行情数据接进脚本、面板、监控、AI 工具或后端服务。

Q4：TickDB 和“只查 A 股价格”的接口有什么区别？

本文只演示 A 股样本，但 TickDB 的产品定位是统一行情数据 API：同一套入口可覆盖外汇、贵金属、指数、美股、港股、A 股、加密货币等市场。对多市场应用来说，统一 symbol 规则、结构化字段和统一鉴权方式，比单次查到一个价格更重要。

Q5：官方资料从哪里看？

可以从 TickDB 官方 GitHub 开始：
https://github.com/TickDB/tickdb-unified-realtime-marketdata-api

小结

行情 API 接入不是“返回价格就结束”。更稳妥的顺序是：

  1. 用 ticker 校验 symbol 和字段身份；
  2. 用 kline 校验周期和 OHLCV 契约；
  3. 用 WebSocket 校验连接和订阅确认；
  4. 用 raw response 和 validation result 留下排错证据。

这样做不会让系统一次性变成生产级，但能避免最常见的接入误区：接口看起来成功，数据却无法解释、无法复查、无法定位问题。

本文只讨论 A 股行情 API 的工程接入、字段校验和证据留痕，不构成任何投资建议。具体字段、端点和可用范围以官方文档及你的当日实测为准。

# python # websocket # 人工智能
Logo
https://edu.csdn.net/learn/39067/627173?utm_source=2019755004

汇聚全球AI编程工具，助力开发者即刻编程。

更多推荐

cover

每天 1000 万 Token，免费云端 AI 编程平台 MonkeyCode 上手指南

avatar AI编程社区
cover

AI Agent 的真相：水面之下，90% 是工程

avatar AI编程社区
cover

ChatGPT、Codex实战：为什么AI写代码以后，真正变慢的是测试和验证？

avatar AI编程社区