幣安WebSocket怎麼訂閱行情？單流多流生產級程式碼

幣安 WebSocket 實時行情的最優方案是：使用 wss://stream.binance.com:9443/stream?streams= 組合流端點，一個連線訂閱最多 1024 個流，訊息延遲小於 50ms，權重消耗幾乎為零。本文給出單流、組合流、動態訂閱、深度本地撮合、斷線重連的完整 Python 與 Node.js 程式碼，全部可直接執行。在開始之前，如果你還沒註冊幣安賬號，可以先到 幣安官網 檢視官方入口，或透過 免費註冊 完成開戶。

一、WebSocket 端點總覽

連線數上限：單 IP 最多 300 個 WebSocket 連線，單連線最多訂閱 1024 個流，每 24 小時自動斷開一次，客戶端必須實現重連。

二、Stream 型別清單

幣安提供 11 種主要行情流：

命名規則：小寫 + @ 符號分隔，例如 btcusdt@ticker 而不是 BTCUSDT@ticker。

三、Python asyncio 單流訂閱

import asyncio, json
import websockets

async def subscribe_ticker(symbol: str):
    url = f"wss://stream.binance.com:9443/ws/{symbol.lower()}@ticker"
    async with websockets.connect(url, ping_interval=180) as ws:
        async for message in ws:
            data = json.loads(message)
            print(f"{data['s']}: 最新價 {data['c']}, "
                  f"24h 漲跌 {data['P']}%, "
                  f"成交量 {data['v']}")

asyncio.run(subscribe_ticker("BTCUSDT"))

輸出示例：

BTCUSDT: 最新價 68420.12, 24h 漲跌 2.35%, 成交量 45231.2
BTCUSDT: 最新價 68421.00, 24h 漲跌 2.36%, 成交量 45231.8

四、組合流訂閱多個交易對

一個連線訂閱 10 個交易對的 ticker + 5 個交易對的 K 線：

import asyncio, json
import websockets

STREAMS = [
    "btcusdt@ticker", "ethusdt@ticker", "bnbusdt@ticker",
    "solusdt@ticker", "xrpusdt@ticker", "adausdt@ticker",
    "dogeusdt@ticker", "maticusdt@ticker", "dotusdt@ticker",
    "avaxusdt@ticker",
    "btcusdt@kline_1m", "ethusdt@kline_1m",
    "bnbusdt@kline_1m", "solusdt@kline_1m", "xrpusdt@kline_1m"
]

async def combined_stream():
    streams = "/".join(STREAMS)
    url = f"wss://stream.binance.com:9443/stream?streams={streams}"
    async with websockets.connect(url, ping_interval=180) as ws:
        async for message in ws:
            payload = json.loads(message)
            stream = payload["stream"]
            data = payload["data"]
            if "@ticker" in stream:
                print(f"[{stream}] {data['s']}: {data['c']}")
            elif "@kline" in stream:
                k = data["k"]
                if k["x"]:  # 只在 K 線閉合時列印
                    print(f"[{stream}] {k['s']} {k['i']} "
                          f"O:{k['o']} H:{k['h']} L:{k['l']} C:{k['c']}")

asyncio.run(combined_stream())

五、動態訂閱/取消（SUBSCRIBE 方法）

連線建立後，可以透過 JSON 訊息動態新增或移除訂閱：

import asyncio, json
import websockets

async def dynamic_sub():
    url = "wss://stream.binance.com:9443/ws"
    async with websockets.connect(url) as ws:
        # 訂閱 BTC 和 ETH 的 bookTicker
        await ws.send(json.dumps({
            "method": "SUBSCRIBE",
            "params": ["btcusdt@bookTicker", "ethusdt@bookTicker"],
            "id": 1
        }))

        count = 0
        async for msg in ws:
            data = json.loads(msg)
            if "result" in data:
                print(f"訂閱確認: {data}")
                continue
            count += 1
            print(f"{data['s']} 買一 {data['b']} 賣一 {data['a']}")

            # 收到 50 條後取消 ETH 訂閱，只保留 BTC
            if count == 50:
                await ws.send(json.dumps({
                    "method": "UNSUBSCRIBE",
                    "params": ["ethusdt@bookTicker"],
                    "id": 2
                }))
                print("已取消 ETH 訂閱")

asyncio.run(dynamic_sub())

六、深度增量合併（本地維護訂單簿）

高頻策略常需要本地撮合級別的訂單簿，流程是：

1. 訂閱 @depth@100ms 差分流，放入緩衝
2. REST 呼叫 GET /api/v3/depth?symbol=X&limit=1000 獲取快照
3. 丟棄緩衝中 u < lastUpdateId 的訊息
4. 逐個應用 U <= lastUpdateId+1 <= u 的訊息
5. 後續訊息直接應用，pu 必須等於上條的 u

import asyncio, json, aiohttp
import websockets
from sortedcontainers import SortedDict

class OrderBook:
    def __init__(self, symbol):
        self.symbol = symbol.lower()
        self.bids = SortedDict()  # 價格 -> 數量
        self.asks = SortedDict()
        self.last_update_id = 0
        self.buffer = []

    def apply_update(self, bids, asks):
        for price, qty in bids:
            price, qty = float(price), float(qty)
            if qty == 0:
                self.bids.pop(price, None)
            else:
                self.bids[price] = qty
        for price, qty in asks:
            price, qty = float(price), float(qty)
            if qty == 0:
                self.asks.pop(price, None)
            else:
                self.asks[price] = qty

    async def load_snapshot(self):
        async with aiohttp.ClientSession() as s:
            url = f"https://api.binance.com/api/v3/depth?symbol={self.symbol.upper()}&limit=1000"
            async with s.get(url) as r:
                snap = await r.json()
                self.last_update_id = snap["lastUpdateId"]
                self.apply_update(snap["bids"], snap["asks"])

    async def run(self):
        url = f"wss://stream.binance.com:9443/ws/{self.symbol}@depth@100ms"
        async with websockets.connect(url) as ws:
            # Step 1: 先接收一段時間差分
            asyncio.create_task(self._collect(ws))
            await asyncio.sleep(1)
            # Step 2: 拉快照
            await self.load_snapshot()
            # Step 3: 應用緩衝中有效差分
            for diff in self.buffer:
                if diff["u"] < self.last_update_id:
                    continue
                self.apply_update(diff["b"], diff["a"])
                self.last_update_id = diff["u"]
            self.buffer = None
            # Step 4: 後續直接應用
            async for msg in ws:
                diff = json.loads(msg)
                self.apply_update(diff["b"], diff["a"])
                self.last_update_id = diff["u"]
                best_bid = self.bids.keys()[-1] if self.bids else 0
                best_ask = self.asks.keys()[0] if self.asks else 0
                print(f"買一 {best_bid} 賣一 {best_ask} 差 {best_ask-best_bid:.2f}")

    async def _collect(self, ws):
        if self.buffer is None:
            return
        async for msg in ws:
            if self.buffer is None:
                break
            self.buffer.append(json.loads(msg))

ob = OrderBook("BTCUSDT")
asyncio.run(ob.run())

七、斷線重連與心跳

幣安 WebSocket 要求客戶端在 3 分鐘內響應 pong，且連線 24 小時強制斷開：

import asyncio, json, websockets, logging

async def resilient_ws(url, on_message, max_retries=1000):
    retry = 0
    while retry < max_retries:
        try:
            async with websockets.connect(
                url,
                ping_interval=180,  # 3 分鐘發一次 ping
                ping_timeout=10,
                close_timeout=5
            ) as ws:
                retry = 0  # 連線成功重置
                async for message in ws:
                    await on_message(json.loads(message))
        except (websockets.ConnectionClosed, asyncio.TimeoutError) as e:
            retry += 1
            wait = min(2 ** retry, 60)  # 指數退避，最多 60s
            logging.warning(f"WS 斷開 {e}，{wait}s 後重連（第 {retry} 次）")
            await asyncio.sleep(wait)

八、Node.js WebSocket 示例

const WebSocket = require('ws');

const streams = ['btcusdt@bookTicker', 'ethusdt@bookTicker'].join('/');
const url = `wss://stream.binance.com:9443/stream?streams=${streams}`;

function connect() {
  const ws = new WebSocket(url);
  ws.on('open', () => console.log('連線成功'));
  ws.on('message', (raw) => {
    const { stream, data } = JSON.parse(raw);
    console.log(`${data.s} 買一 ${data.b} 賣一 ${data.a}`);
  });
  ws.on('close', () => {
    console.log('斷線，3s 後重連');
    setTimeout(connect, 3000);
  });
  ws.on('error', (err) => console.error('錯誤', err.message));
}

connect();

九、常見問題 FAQ

Q1: WebSocket 也有權重消耗嗎？

A: 建立連線本身有 2 權重；後續推送訊息 完全不消耗權重。但連線數超過 300 / IP 會被拒絕。

Q2: 為什麼訂閱後沒收到任何訊息？

A: 三種常見原因：1) stream 名必須 全小寫（btcusdt 不是 BTCUSDT）；2) 交易對寫錯（例如用 BTC-USDT 而不是 BTCUSDT）；3) 訂閱訊息必須在 wss://.../ws 端點而不是 /stream 端點。

Q3: @depth 和 @depth@100ms 什麼區別？

A: @depth 推 1000ms 一次完整快照（20 檔）；@depth@100ms 推 100ms 一次差分（增量）。本地維護訂單簿必須用 @100ms，看盤展示用 @depth 即可。

Q4: WebSocket 可以下單嗎？

A: 可以。幣安提供 WebSocket API（wss://ws-api.binance.com:443/ws-api/v3）用於簽名下單，比 REST 延遲低 30-50ms。但生態不成熟，主流策略仍用 REST 下單 + WS 監聽。

Q5: 24 小時自動斷開如何無縫切換？

A: 雙連線滾動方案：到 23 小時時建立新連線 B，等 B 穩定收到訊息後再關閉舊連線 A，避免瞬間丟資料。生產級框架如 python-binance 已內建此邏輯。

看完 WebSocket 方案，回到 分類導航 檢視「API接入」分類其它 SDK 教程。