A股 WebSocket 行情订阅成功后，如何判断实时数据真的可用？

摘要

WebSocket 行情接入里，connected 不是终点，subscribed 也不是终点。真正能进入行情面板、监控告警或数据管道之前，至少要确认四个状态：连接已建立、订阅已确认、首条业务消息已收到、后续数据没有进入 stale。本文用 TickDB WebSocket 真实订阅 600519.SH 的 ticker 频道，把这四个状态拆开，并给出一份可直接运行的 Python 状态校验脚本、真实终端截图和日志字段设计。

!image.png

WebSocket 行情接入至少要分清 connected、subscribed、first_message_received 和 stale 四个状态。*

这篇文章解决什么问题

如果你正在做 A股实时行情面板、监控告警、量化数据管道，或者准备把实时行情接入 AI Agent，最容易误判的一件事是：

WebSocket 连接成功，不等于实时行情已经可用。

一条 WebSocket 行情链路至少有四个不同状态：

1. connected：连接建立，通道打开；
2. subscribed：订阅命令被服务端接受；
3. first_message_received：首条业务行情消息真正到达；
4. stale：连接仍在，但业务消息长时间没有更新。

!image.png

四状态校验：每一步回答的问题不同。不要把连接成功、订阅成功、业务消息到达和数据新鲜度混成一个“成功”。

很多系统只检查前两个状态。只要连接还在、订阅返回成功，就把行情面板标成正常。但如果没有收到第一条 ticker 消息，或者最近一条消息已经过期，面板上的价格其实只是“看起来还在运行”。

本文不讨论交易信号，不讨论收益，不讨论高频交易，也不承诺延迟或 SLA。只讨论一个工程问题：

用 WebSocket 接 A股实时行情时，如何判断订阅链路真的已经可用？

一、connected：连接建立了，但还没有业务语义

connected 只说明客户端和服务端之间的 WebSocket 通道已经建立。

在 TickDB 的 WebSocket 实测中，连接建立后会收到类似这样的控制消息：

{
  "cmd": "connected",
  "code": 0,
  "message": "Connected successfully",
  "data": {
    "connection_id": "739c7ebb-97b9-436f-a09a-37f17101f741",
    "server_time": 1783480593712
  }
}

这一步能证明“通道打开了”，但不能证明行情数据已经进来。

你还没有告诉服务端要订阅哪个频道、哪个 symbol。此时不能把面板标记为可用，也不能开始做价格判断。正确动作是发送订阅命令。

二、subscribed：订阅被接受了，但还没有收到业务数据

订阅 ticker 频道时，命令结构是：

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

服务端返回订阅成功时，会收到类似这样的消息：

{
  "cmd": "subscribe",
  "code": 0,
  "message": "Subscribed successfully",
  "data": {
    "channel": "ticker",
    "symbols": ["600519.SH"]
  }
}

这一步能证明“订阅关系建立了”，但仍然不能证明业务数据已经开始流动。

很多接入问题就卡在这里：订阅成功，但迟迟没有 ticker 消息。如果代码只看 cmd=subscribe, code=0，就会误以为行情链路正常。

所以，订阅确认之后，还必须等第一条业务消息。

!image.png

实测：终端依次记录 connecting -> connected、connected -> subscribed、subscribed -> first_message_received。API Key 已脱敏。

三、first_message_received：第一条业务消息才证明链路跑通

当收到第一条 cmd=ticker 消息时，才说明 WebSocket 从连接、订阅到业务推送的最小链路跑通。

本次实测中，600519.SH 的 ticker 消息包含这些字段：

{
  "cmd": "ticker",
  "data": {
    "symbol": "600519.SH",
    "type": "stock",
    "last_price": "1186.46",
    "volume_24h": "11486",
    "high_24h": "1195",
    "low_24h": "1177",
    "price_change_24h": "-2.34",
    "price_change_percent_24h": "-0.20",
    "timestamp": 1783480596000
  }
}

注意，这里只能说“本次实测收到这些字段”。不同频道、不同资产类别、后续文档更新，都可能影响字段结构。生产系统不应该只靠文章里的样例写死字段，而应该保留 raw snapshot，并按官方文档和当日实测做字段校验。

最低限度，ticker 消息进入系统前要检查：

!image.png

实测：收到 cmd=ticker 业务消息后，至少要核对 symbol、last_price、timestamp，并保存 raw snapshot。

四、stale：连接还在，但行情可能已经停了

最隐蔽的问题不是连接断开，而是连接看起来还在，业务消息却不再更新。

这时系统不能继续显示“正常”。应该根据交易时段和业务容忍度设定阈值。例如：

• 首条消息等待超过 30 秒：first_message_timeout
• 交易时段内 120 秒没有新 ticker：no_message_in_trading_hours
• 非交易时段或午间休市：不能直接判异常，需要结合交易日历或市场状态

stale 的本质不是“WebSocket 断了”，而是：

业务数据已经不能证明自己还新鲜。

所以生产系统需要同时记录 last_message_at 和 message_count。出现问题时，先问两件事：

1. 最后一条业务消息是什么时候到的？
2. 在进入 stale 前总共收到了多少条业务消息？

没有这两个字段，排查只能靠猜。

五、状态总览

六、生产日志字段建议

每一次状态变化都应该写日志。最小字段建议如下：

七、可直接运行的 Python WebSocket 状态校验脚本

下面这段是可直接运行的 Python 脚本，用于验证 TickDB WebSocket ticker 订阅的四状态链路。

运行前安装依赖：

python3 -m pip install websockets certifi

设置环境变量：

export TICKDB_API_KEY="你的 API Key"
export TICKDB_SYMBOL="600519.SH"
python3 tickdb_ws_state_guard.py

代码如下：

import asyncio
import contextlib
import json
import os
import ssl
import time
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from pathlib import Path
from urllib.parse import quote

import certifi
import websockets


WS_URL = "wss://api.tickdb.ai/v1/realtime"
CHANNEL = os.environ.get("TICKDB_CHANNEL", "ticker")
SYMBOL = os.environ.get("TICKDB_SYMBOL", "600519.SH")
API_KEY = os.environ["TICKDB_API_KEY"]

FIRST_MESSAGE_TIMEOUT = int(os.environ.get("FIRST_MESSAGE_TIMEOUT", "30"))
STALE_THRESHOLD = int(os.environ.get("STALE_THRESHOLD", "120"))
RUN_SECONDS = int(os.environ.get("RUN_SECONDS", "45"))

SNAPSHOT_DIR = Path(os.environ.get("SNAPSHOT_DIR", "./ws_snapshots"))
SNAPSHOT_DIR.mkdir(parents=True, exist_ok=True)


def utc_now():
    return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")


@dataclass
class StreamLog:
    symbol: str
    channel: str
    state: str = "connecting"
    connected_at: str | None = None
    subscribed_at: str | None = None
    first_message_at: str | None = None
    last_message_at: str | None = None
    message_count: int = 0
    stale_reason: str | None = None
    raw_snapshot_path: str | None = None


def save_snapshot(payload, symbol: str, label: str) -> str:
    safe_symbol = symbol.replace(".", "_")
    path = SNAPSHOT_DIR / f"{safe_symbol}_{label}_{int(time.time() * 1000)}.json"
    if isinstance(payload, str):
        path.write_text(payload, encoding="utf-8")
    else:
        path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
    return str(path)


def validate_ticker_message(message: dict, symbol: str) -> tuple[bool, str]:
    if message.get("cmd") != "ticker":
        return False, "not_ticker_message"

    data = message.get("data")
    if not isinstance(data, dict):
        return False, "missing_data_object"

    if data.get("symbol") != symbol:
        return False, "symbol_mismatch"

    if "last_price" not in data:
        return False, "missing_last_price"

    try:
        Decimal(str(data["last_price"]))
    except (InvalidOperation, ValueError):
        return False, "invalid_last_price"

    if not isinstance(data.get("timestamp"), int):
        return False, "invalid_timestamp"

    return True, "ok"


async def ping_loop(ws, stop_event: asyncio.Event):
    while not stop_event.is_set():
        await ws.send(json.dumps({"cmd": "ping"}))
        await asyncio.sleep(1)


async def run():
    log = StreamLog(symbol=SYMBOL, channel=CHANNEL)
    subscribe = {
        "cmd": "subscribe",
        "data": {
            "channel": CHANNEL,
            "symbols": [SYMBOL],
        },
    }

    url = f"{WS_URL}?api_key={quote(API_KEY)}"
    ssl_context = ssl.create_default_context(cafile=certifi.where())
    stop_ping = asyncio.Event()

    print("WS_OPEN wss://api.tickdb.ai/v1/realtime?api_key=********")
    print("SEND", json.dumps(subscribe, ensure_ascii=False))

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

        started_at = time.monotonic()
        first_wait_started_at = time.monotonic()
        last_seen_monotonic = None

        try:
            while time.monotonic() - started_at < RUN_SECONDS:
                try:
                    raw = await asyncio.wait_for(ws.recv(), timeout=5)
                except asyncio.TimeoutError:
                    if log.first_message_at is None and time.monotonic() - first_wait_started_at > FIRST_MESSAGE_TIMEOUT:
                        log.state = "stale"
                        log.stale_reason = "first_message_timeout"
                        print("STATE subscribed -> stale")
                        break

                    if last_seen_monotonic is not None and time.monotonic() - last_seen_monotonic > STALE_THRESHOLD:
                        log.state = "stale"
                        log.stale_reason = "no_message_after_first_message"
                        print("STATE first_message_received -> stale")
                        break
                    continue

                try:
                    message = json.loads(raw)
                except json.JSONDecodeError:
                    log.state = "error"
                    log.stale_reason = "invalid_json"
                    log.raw_snapshot_path = save_snapshot(raw, SYMBOL, "invalid_json")
                    print("STATE error")
                    break

                cmd = message.get("cmd")

                if cmd == "connected":
                    log.state = "connected"
                    log.connected_at = utc_now()
                    print("STATE connecting -> connected")
                    print("RECV", json.dumps(message, ensure_ascii=False, indent=2))
                    continue

                if cmd == "subscribe":
                    if message.get("code") == 0:
                        log.state = "subscribed"
                        log.subscribed_at = utc_now()
                        print("STATE connected -> subscribed")
                    else:
                        log.state = "error"
                        log.stale_reason = "subscribe_rejected"
                        log.raw_snapshot_path = save_snapshot(message, SYMBOL, "subscribe_rejected")
                    print("RECV", json.dumps(message, ensure_ascii=False, indent=2))
                    continue

                if cmd == "pong":
                    continue

                ok, reason = validate_ticker_message(message, SYMBOL)
                if not ok:
                    log.state = "error"
                    log.stale_reason = reason
                    log.raw_snapshot_path = save_snapshot(message, SYMBOL, "invalid")
                    print("STATE error")
                    print(json.dumps(asdict(log), ensure_ascii=False, indent=2))
                    break

                now = utc_now()
                last_seen_monotonic = time.monotonic()
                log.last_message_at = now
                log.message_count += 1
                log.raw_snapshot_path = save_snapshot(message, SYMBOL, "ticker")

                if log.first_message_at is None:
                    log.state = "first_message_received"
                    log.first_message_at = now
                    print("STATE subscribed -> first_message_received")

                print("RECV", json.dumps(message, ensure_ascii=False, indent=2))

        finally:
            stop_ping.set()
            ping_task.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await ping_task

    print("FINAL_STATE")
    print(json.dumps(asdict(log), ensure_ascii=False, indent=2))


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

这段代码做了几件事：

• WebSocket 鉴权使用 URL query api_key；
• 订阅命令使用 cmd=subscribe 和 data.channel / data.symbols；
• 每秒发送一次 ping，但不把心跳当成业务数据；
• 只把 cmd=ticker 当成业务消息；
• 校验 symbol、last_price、timestamp；
• 保存 ticker 原始快照；
• 第一条业务消息超时则进入 stale；
• 持续阶段消息沉寂超阈值也进入 stale。

八、本次真实实测截图

本次实测使用 TickDB WebSocket 订阅 600519.SH 的 ticker 频道，终端记录了 connected、subscribed、first_message_received 三个状态，并收到 ticker 业务消息。

!image.png

图中可以看到：

• WebSocket 连接地址为 wss://api.tickdb.ai/v1/realtime?api_key=**；
• API Key 已脱敏；
• 订阅频道为 ticker；
• 订阅 symbol 为 600519.SH；
• 收到 connected、subscribe、ticker 消息；
• ticker 消息里包含 symbol、last_price、timestamp 等字段。

这张图只证明采样时刻的最小订阅链路跑通，不代表延迟、SLA、稳定性、全市场覆盖或交易信号。

九、TickDB 在这条链路里的位置

TickDB 是一个统一行情数据 API，提供 REST、WebSocket、MCP 等接入方式。放到这篇文章的问题里，可以这样理解：

• WebSocket：适合实时订阅和持续推送，本文实测的就是这一层；
• REST：适合初始化快照、查询历史数据、在 stale 恢复后做补数据；
• MCP：适合让 AI 工具按需查询行情，不替代 WebSocket 的持续推送。

三种方式的鉴权不能混写：

如果你的系统要做实时行情面板或监控告警，不能只看 WebSocket 是否连接成功。应用侧仍然要记录状态、保存 raw snapshot、设置 stale 阈值、处理重连和回补。

TickDB 提供的是可验证的行情接入入口；状态判断、日志留痕和异常处理，仍然应该由你的系统自己完成。

十、上线前检查清单

接入任何 WebSocket 行情 API，都建议先过这张表：

十一、FAQ

Q1：WebSocket connected 后为什么不能直接认为行情可用？

因为 connected 只说明通道建立。你还没有确认订阅是否成功，也没有收到业务行情消息。

Q2：subscribe 成功后为什么还要等 first message？

因为 subscribe 成功只说明订阅请求被接受。第一条 ticker 消息才证明业务数据真的进入客户端。

Q3：什么情况下应该标记 stale？

交易时段内，超过阈值没有收到新业务消息时。首条消息迟迟不到，也应该标记为 stale 或 first message timeout。

Q4：心跳正常能不能证明行情正常？

不能。ping/pong 只能证明连接还活着，不证明 ticker 业务消息还在持续推送。

Q5：本次 TickDB WebSocket 实测能证明什么？

能证明 2026-07-08 采样时刻，600519.SH 的 ticker 频道完成了 connected、subscribed、first_message_received 的最小链路。不能证明延迟、SLA、全市场覆盖、全时段可用、生产稳定性或交易信号。

最后

WebSocket 行情接入，不要只问“连上了吗”。

更应该问：

• 订阅确认了吗？
• 第一条业务消息到了吗？
• 最后一条业务消息是什么时候？
• 现在是不是已经 stale？

如果这四个问题都能回答，你的实时行情系统才不是靠感觉在运行，而是靠状态、日志和证据在运行。

本文只讨论 A股实时行情 WebSocket 的接入校验、日志留痕和数据新鲜度判断，不构成任何投资建议。

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

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