实时行情 WebSocket 推送接口文档

务必加入群接收更新通知，QQ 群：781775236，客服 QQ：657688959（备注 id）。

概述

本文档描述如何通过 WebSocket 协议连接行情推送服务，订阅并接收实时行情数据（市场行情、涨停预警等）。

传输协议：WSS（WebSocket over TLS）
推送时间：交易日 09:15 ~ 15:00 + 15:00 ~ 15:30（收盘集合竞价）
连接地址：wss://zhunData.cn/ws/?key={AUTH_KEY}&version={CLIENT_VERSION}
鉴权方式：URL 查询参数 key；version 为客户端版本
版本提示：版本不是最新时，连接后提示升级；每天 09:00 再提示一次
文本消息：JSON，用于订阅确认、错误、Key 提醒、客户端更新通知
行情消息：二进制 QB1/QH1 帧；type/key 明文，data 使用 zstandard 压缩 + msgpack 序列化
字典：v20261029+ 的行情 data 使用共享 zstd 字典压缩，订阅前必须先同步字典（自动下载）
客户端顺序：连接 → 同步/下载字典 → 订阅 → 接收行情

客户端示例：client_demo.py（v20261029）

bash

pip install websockets msgpack zstandard psutil

推送字段

market_quote 外层结构

字段	类型	说明
`type`	string	消息类型，值为 `"market_quote"`
`key`	string	标的符号，如 `"tick.stock.sz.000001"`
`data`	object	行情明细，字段如下

`data` 行情明细字段

字段	类型	说明
`time`	long	行情时间戳（毫秒）
`lastPrice`	float	最新成交价
`open`	float	今日开盘价
`high`	float	今日最高价
`low`	float	今日最低价
`lastClose`	float	昨日收盘价
`amount`	float	累计成交金额
`volume`	float	累计成交数量
`pvolume`	float	原始成交数量（未复权）
`stockStatus`	int	证券状态
`openInt`	int	持仓量（期货适用）
`transactionNum`	int	成交笔数
`lastSettlementPrice`	float	昨日结算价（期货适用）
`settlementPrice`	float	结算价（期货适用）
`pe`	float	市盈率
`askPrice`	float[]	卖一到五委托价，5 个元素的数组
`bidPrice`	float[]	买一到五委托价，5 个元素的数组
`askVol`	int[]	卖一到五委托量，5 个元素的数组
`bidVol`	int[]	买一到五委托量，5 个元素的数组
`volRatio`	float	量比
`speed1Min`	float	1 分钟涨速
`speed5Min`	float	5 分钟涨速
`code`	string	证券代码

示例（解码后的逻辑结构）：

json

{
  "type": "market_quote",
  "key": "tick.stock.sh.600023",
  "data": {
    "time": 1790648864000,
    "lastPrice": 5.19,
    "open": 5.18,
    "high": 5.24,
    "low": 5.14,
    "lastClose": 5.19,
    "amount": 73898400.0,
    "volume": 142042,
    "pvolume": 14204200,
    "stockStatus": 3,
    "openInt": 13,
    "transactionNum": 0,
    "lastSettlementPrice": 5.19,
    "settlementPrice": 0,
    "pe": 0.0,
    "askPrice": [5.19, 5.20, 5.21, 5.22, 5.23],
    "bidPrice": [5.18, 5.17, 5.16, 5.15, 5.14],
    "askVol": [1348, 4596, 6523, 5281, 7574],
    "bidVol": [3735, 3061, 1424, 3849, 2094],
    "volRatio": 0.0,
    "speed1Min": 0.0,
    "speed5Min": 0.0,
    "code": "600023"
  }
}

这是解码后的结构；实际网络上是 QB1/QH1 二进制帧，data 部分是压缩的。

连接

wss://zhunData.cn/ws/?key={AUTH_KEY}&version={CLIENT_VERSION}

鉴权 Key 通过 URL 查询参数 key 传递；version 传客户端当前版本号，便于服务端判断是否需要提示升级。

连接关闭码

关闭码	说明
`4001`	缺少鉴权 Key
`4003`	Key 无效或已过期
`4009`	Key 已被其他连接占用（同一 Key 只允许一个连接）
`1013`	服务端背压保护，稍后重连即可
`4011`	Key 过期超过宽限期，服务端主动断开

1013 是服务端保护：单连接待发消息/内存或写缓冲超限时会断开。频繁出现时减少订阅数量、降低日志输出。

数据压缩字典同步

行情 data 使用共享 zstd 字典压缩。

流程概要：

发送 {"action": "dict_info", "version": 本地版本, "dict_id": 本地ID} 查询服务端字典版本（首次运行时 version=0, dict_id=0）
服务端返回版本信息，match=true 表示本地字典可用，直接加载
match=false 或本地缺失时，发送 {"action": "dict_download"} 下载字典，服务端先返回 JSON 元数据，再发送二进制字典帧，客户端校验后写入本地并初始化解压器
字典就绪后再发送 subscribe

完整的字典同步逻辑已内置在 client_demo.py 中，无需手动实现。

订阅行情

json

{
  "action": "subscribe",
  "channel": "market_quote",
  "symbols": ["tick.stock.sz.000001", "tick.stock.sh.600744"]
}

字段	类型	必填	说明
`action`	string	是	`"subscribe"` 或 `"unsubscribe"`
`channel`	string	是	频道名称
`symbols`	string[]	是	标的符号列表

频道

频道	说明
`market_quote`	市场行情数据
`limit_alert`	涨停预警数据

标的符号格式

tick.{asset_type}.{suffix}.{code}

资产类型：stock、index、fund、bond
交易所后缀：sz、sh、bj、*（通配）

示例：

符号	说明
`tick.stock.sz.000001`	深市平安银行
`tick.stock.sh.600744`	沪市华银电力
`tick.fund.sh.516100`	沪市基金
`tick.stock.*`	所有股票（通配）
`tick.index.sz.*`	深市所有指数（通配）

订阅权限

通配符订阅（如 tick.stock.*）：需要 {asset}_all 权限
单标的订阅：需要 {asset}_all 或 {asset}_100 权限
{asset}_100 类型下，单标的最多 100 个

订阅确认

json

{"type": "subscribe_ok", "channel": "market_quote", "symbols": ["tick.stock.sz.000001"]}
{"type": "subscribe_rejected", "channel": "market_quote", "symbols": ["..."], "msg": "..."}

取消订阅是频道级的：

json

{"action": "unsubscribe", "channels": ["market_quote"]}

json

{"type": "unsubscribe_ok", "channels": ["market_quote"]}

推送数据格式

QH1（单条消息）

b"QH1" + type(1B) + key_len(2B, 大端) + key(UTF-8) + zstd(msgpack(data), dict_data)

部分	说明
`QH1`	固定魔数
`type`	`1` = `market_quote`，`2` = `limit_alert`
`key_len` / `key`	标的符号，如 `tick.stock.sz.000001`
剩余部分	只对 `data` 做 zstandard 压缩 + msgpack 序列化；v20261029+ 使用共享字典

QB1（批量帧）

b"QB1" + count(2B, 大端) + 重复[ len(4B, 大端) + QH1 ]

服务端可能把多条行情打成一个 QB1 帧，客户端应遍历帧内每条消息。

type/key 是明文，只有 data 压缩。v20261029+ 使用共享字典；低于 v20261029 的客户端无法解压新行情，请升级。

文本通知

type	说明
`subscribe_ok` / `subscribe_rejected`	订阅成功 / 被拒绝
`unsubscribe_ok`	取消订阅成功
`error`	服务端错误
`key_expiring_soon`	Key 距过期不足 7 天
`key_expired_warning`	Key 已过期，仍在 8 小时宽限期
`key_expired_disconnect`	Key 超过宽限期，即将断开（4011）
`notice`	通用通知；`category=client_update` 表示客户端更新

notice 的 category 用于区分通知类型；目前 client_update 会在以下时机发送：

客户端连接时，若 version 与最新版本（v20261029）不一致
每天上午 09:00，对版本不是最新的在线客户端再提示一次

完整示例

覆盖连接、订阅、QB1/QH1 解码、文本通知和断线重连：

python

import asyncio
import hashlib
import json
import os
import random
import re
import ssl
import struct
import websockets
import msgpack
import zstandard as zstd
from datetime import datetime

WS_URI = "wss://zhunData.cn/ws/"
AUTH_KEY = "your_api_key"
CLIENT_VERSION = "v20261029"
RECONNECT_MIN = 2
RECONNECT_MAX = 5

CLIENT_DATA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "ws_client_data")
DICT_MAGIC = b"QD1"
dctx = None

_MAGIC_MSG = b"QH1"
_MAGIC_BATCH = b"QB1"
_TYPE_NAMES = {1: "market_quote", 2: "limit_alert"}


def _latest_local_dict(path_dir):
    if not os.path.isdir(path_dir):
        return None
    pattern = re.compile(r"^tick_dict_v(\d+)\.zdict$")
    best, best_version = None, -1
    for name in os.listdir(path_dir):
        m = pattern.match(name)
        if m and int(m.group(1)) > best_version:
            best_version = int(m.group(1))
            best = os.path.join(path_dir, name)
    return best


async def ensure_dictionary(ws):
    """连接后、订阅前同步字典。完整实现见 client_demo.py。"""
    global dctx
    os.makedirs(CLIENT_DATA_DIR, exist_ok=True)
    local_path = _latest_local_dict(CLIENT_DATA_DIR)
    local_version, local_id, local_data = 0, 0, None
    if local_path:
        try:
            raw = open(local_path, "rb").read()
            local_data = zstd.ZstdCompressionDict(raw)
            local_id = local_data.dict_id()
            local_version = int(os.path.basename(local_path)[len("tick_dict_v"):-len(".zdict")])
        except Exception:
            local_path, local_version, local_id = None, 0, 0

    await ws.send(json.dumps({"action": "dict_info",
                              "version": local_version, "dict_id": local_id}))
    while True:
        msg = await ws.recv()
        if isinstance(msg, str):
            info = json.loads(msg)
            if info.get("type") == "dict_info":
                break
            handle_text(info)
    if not info.get("available"):
        raise RuntimeError(info.get("msg", "server has no dictionary"))
    if info.get("match") and local_path:
        dctx = zstd.ZstdDecompressor(dict_data=local_data)
        return

    await ws.send(json.dumps({"action": "dict_download"}))
    while True:
        msg = await ws.recv()
        if isinstance(msg, str):
            meta = json.loads(msg)
            if meta.get("type") == "dict_download" and meta.get("ok") is False:
                raise RuntimeError(meta.get("msg", "dict download failed"))
            if meta.get("type") != "dict_download":
                handle_text(meta)
            continue
        if msg[:3] != DICT_MAGIC:
            continue
        version = struct.unpack(">I", msg[3:7])[0]
        raw = bytes(msg[7:])
        if version != int(info["version"]) or len(raw) != int(info["size"]):
            raise RuntimeError("dictionary version/size mismatch")
        if hashlib.sha256(raw).hexdigest() != info["sha256"]:
            raise RuntimeError("dictionary sha256 mismatch")
        break

    target = os.path.join(CLIENT_DATA_DIR, "tick_dict_v%d.zdict" % version)
    with open(target, "wb") as fh:
        fh.write(raw)
    loaded = zstd.ZstdCompressionDict(raw)
    if loaded.dict_id() != int(info["dict_id"]):
        raise RuntimeError("dictionary dict_id mismatch")
    for name in os.listdir(CLIENT_DATA_DIR):
        if re.match(r"^tick_dict_v\d+\.zdict$", name):
            old_path = os.path.join(CLIENT_DATA_DIR, name)
            if os.path.abspath(old_path) != os.path.abspath(target):
                os.remove(old_path)
    dctx = zstd.ZstdDecompressor(dict_data=loaded)


def decode_ws_frame(frame, dctx):
    if isinstance(frame, str):
        return []
    if frame[:3] == _MAGIC_BATCH:
        count = struct.unpack(">H", frame[3:5])[0]
        pos = 5
        envelopes = []
        for _ in range(count):
            length = struct.unpack(">I", frame[pos:pos + 4])[0]
            pos += 4
            envelopes.append(frame[pos:pos + length])
            pos += length
    elif frame[:3] == _MAGIC_MSG:
        envelopes = [frame]
    else:
        raise ValueError("unknown frame")

    payloads = []
    for env in envelopes:
        key_len = struct.unpack(">H", env[4:6])[0]
        key = env[6:6 + key_len].decode("utf-8")
        data = msgpack.unpackb(dctx.decompress(env[6 + key_len:]), raw=False)
        payloads.append({"type": _TYPE_NAMES.get(env[3], "market_quote"),
                         "key": key, "data": data})
    return payloads


def handle_text(text_data):
    msg_type = text_data.get("type")
    if msg_type == "subscribe_ok":
        print("订阅成功:", text_data.get("channel"))
    elif msg_type == "subscribe_rejected":
        print("订阅拒绝:", text_data.get("msg"))
    elif msg_type == "key_expiring_soon":
        print("Key 即将过期:", text_data.get("days_remaining"), "天")
    elif msg_type == "key_expired_warning":
        print("Key 已过期, 宽限剩余:", text_data.get("grace_hours_remaining"), "小时")
    elif msg_type == "key_expired_disconnect":
        print("Key 过期断开:", text_data.get("msg"))
    elif msg_type == "notice":
        print("通知:", text_data.get("category"), text_data.get("msg"))


def handle_tick(payload):
    msg_type = payload.get("type")
    key = payload.get("key")
    data = payload.get("data", {})
    if msg_type == "market_quote":
        ts = data.get("time")
        dt = datetime.fromtimestamp(ts / 1000) if ts else None
        print("行情:", key, dt, data.get("lastPrice"))
    elif msg_type == "limit_alert":
        print("涨停预警:", key)


async def subscribe_client(ws):
    await ws.send(json.dumps({
        "action": "subscribe",
        "channel": "market_quote",
        "symbols": ["tick.stock.sz.000001"],
    }))
    while True:
        response = await ws.recv()
        if not isinstance(response, str):
            print("服务端确认:", response)
            break
        data = json.loads(response)
        if data.get("type") == "subscribe_ok":
            print("订阅成功:", response)
            break
        handle_text(data)


async def receive_data(ws):
    async for message in ws:
        if isinstance(message, str):
            handle_text(json.loads(message))
            continue
        for payload in decode_ws_frame(message, dctx):
            handle_tick(payload)


async def run_client():
    while True:
        try:
            ws_uri = f"{WS_URI}?key={AUTH_KEY}&version={CLIENT_VERSION}"
            ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
            ssl_ctx.check_hostname = False
            ssl_ctx.verify_mode = ssl.CERT_NONE
            async with websockets.connect(ws_uri, ssl=ssl_ctx) as ws:
                print("连接成功")
                await ensure_dictionary(ws)
                await subscribe_client(ws)
                await receive_data(ws)
        except (websockets.ConnectionClosed, ConnectionRefusedError,
                OSError, asyncio.TimeoutError) as e:
            if getattr(e, "code", None) in (4001, 4003, 4009):
                print("鉴权失败，停止重连:", e)
                return
            wait_time = random.randint(RECONNECT_MIN, RECONNECT_MAX)
            print(f"{wait_time} 秒后重连...")
            await asyncio.sleep(wait_time)


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

实时行情 WebSocket 推送接口文档 ​

概述 ​

推送字段 ​

market_quote 外层结构 ​

data 行情明细字段 ​

连接 ​

连接关闭码 ​

数据压缩字典同步 ​

订阅行情 ​

频道 ​

标的符号格式 ​

订阅权限 ​

订阅确认 ​

推送数据格式 ​

QH1（单条消息） ​

QB1（批量帧） ​

文本通知 ​

完整示例 ​

实时行情 WebSocket 推送接口文档

概述

推送字段

market_quote 外层结构

`data` 行情明细字段

连接

连接关闭码

数据压缩字典同步

订阅行情

频道

标的符号格式

订阅权限

订阅确认

推送数据格式

QH1（单条消息）

QB1（批量帧）

文本通知

完整示例