WebSocket 连接成功但行情不更新 应该先查哪几个状态

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

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

你的 WebSocket 连上了，也发了订阅，但行情看板一动不动。是网络问题？是服务端没推？还是你的代码根本没收到？大多数开发者会反复检查连接状态，但真正的问题往往藏在四个被忽略的状态转换里：

1. 真的 connected 了吗——通道打开，但还没有业务语义。
2. 真的 subscribed 了吗——订阅被接受，但可能还没收到第一条数据。
3. 真的收到了 第一条 ticker 消息 吗——链路跑通的唯一证据。
4. 数据是不是已经 stale 了——连接还在，但业务消息早就停了。

下面按“症状 → 可能原因 → 怎么查 → 日志字段”的顺序，把每个状态拆开，并配上一段可直接运行的 Python WebSocket 校验脚本。

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

症状：连接事件触发，但没有任何行情数据过来。

可能原因：只是完成了 WebSocket 握手，还没订阅任何频道。

connected 只说明客户端和服务端之间的 WebSocket 通道已经建立。在 TickDB 的 WebSocket 实测中，连接建立后会收到类似这样的控制消息：

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

怎么查：收到 cmd=connected 后，立即发送订阅命令。不要在这一步就认为链路正常——通道打开不等于有数据。

日志建议：记录 connected_at。

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

症状：订阅返回了 code=0，但 ticker 消息始终不来。

可能原因：订阅确认只代表服务端接受了请求，不代表该 symbol 正在产生行情（比如非交易时段），或者推送通道有其他限制。

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

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

服务端返回订阅成功时，会收到类似：

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

怎么查：等待第一条 cmd=ticker 消息。设置一个超时（比如 30 秒），如果超时未收到，标记为 stale 或 first_message_timeout。

日志建议：记录 subscribed_at，并启动首条消息超时计时器。

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

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

症状：订阅成功了，但依然没有数据——或者有数据但解析失败。

可能原因：字段结构不匹配、symbol 不一致、或者收到的根本不是业务消息（可能是控制帧）。

当收到第一条 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
  }
}

怎么查：收到消息后，必须逐项校验：

任何一项不通过，都应该把链路标记为异常，而不是静默吞掉。

日志建议：记录 first_message_at、message_count，并保存第一帧的原始快照。

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

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

症状：看板价格“一动不动”，但 WebSocket 没有断，心跳也正常。

可能原因：非交易时段、服务端推送异常、本地网络阻塞但未触发超时。最隐蔽的情况是——程序在运行，连接也在，但业务消息已经很久没更新了。

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

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

怎么查：记录 last_message_at 和 message_count。出现问题时，先问两件事：

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

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

日志建议：记录 last_message_at、stale_reason。

五、状态总览

六、生产日志字段建议

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

七、可直接运行的 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 业务消息。

图中可以看到：

• 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 的接入校验、日志留痕和数据新鲜度判断，不构成任何投资建议。