从零搭建：数据看板实时行情接口全流程示范

import json, websocket  # pip install websocket-client

class RealtimeFeed:
    def __init__(self, token: str):
        self.url = f"wss://quote.tradeswitcher.com/quote-stock-b-ws-api?token={token}"
        self.ws = None

    def on_open(self, ws):
        subscribe = {
            "cmd_id": 22002,
            "seq_id": 123,
            "trace": "3baaa938-uuid-timestamp",
            "data": {
                "symbol_list": [
                    {"code": "700.HK",   "depth_level": 5},
                    {"code": "UNH.US",   "depth_level": 5}
                ]
            }
        }
        ws.send(json.dumps(subscribe))
        print(">>> 已发送订阅请求")

    def on_message(self, ws, message):
        tick = json.loads(message)
        print(">>> 收到实时数据：", tick)
        # 此处可回传 DataBoard 进行绘图
        self.push_to_dashboard(tick)

    def on_error(self, ws, error):
        print(">>> 连接错误：", error)

    def on_close(self, ws, code, msg):
        print(">>> 连接关闭：", code, msg)

    def run(self):
        self.ws = websocket.WebSocketApp(
            self.url,
            on_open=self.on_open,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close
        )
        self.ws.run_forever()

    @staticmethod
    def push_to_dashboard(tick):
        # 仅打印，稍后接入可视化
        pass

if __name__ == "__main__":
    RealtimeFeed(token="你的TOKEN").run()

class RealtimeFeed:
    ...
    @staticmethod
    def push_to_dashboard(tick):
        symbol = next(iter(tick["data"]))
        snap   = tick["data"][symbol]["snapshot"]
        price  = snap["last"]
        volume = snap["volume"]
        Board.update(symbol, price, volume)

import matplotlib.pyplot as plt
from datetime import datetime
import threading

class DataBoard:
    lock = threading.Lock()
    cache = {}  # symbol -> (prices[], timestamps[])

    @classmethod
    def update(cls, symbol: str, price: float, _volume):
        with cls.lock:
            x_series, y_series = cls.cache.setdefault(symbol, ([], []))
            x_series.append(datetime.now())
            y_series.append(price)
            cls.drawing(symbol)

    @classmethod
    def drawing(cls, symbol):
        plt.clf()
        x, y = cls.cache[symbol]
        plt.plot(x[-100:], y[-100:], label=symbol)  # 仅画最近 100 点
        plt.pause(0.01)

plt.ion()

if __name__ == "__main__":
    threading.Thread(target=RealtimeFeed(token="你的TOKEN").run, daemon=True).start()
    plt.show(block=True)