SBE BBO 接入指南

總覽

Bybit 以 SBE (Simple Binary Encoding) 格式提供即時 Level 1 訂單簿資料, 並遵循 FIX Protocol SBE 1.0 規範。

• 目前版本僅支援 SBE 二進制 格式。
• 對於 linear / inverse / spot 的 Level 1 資料: 如果 3 秒內訂單簿沒有變化, 會再次推送 snapshot 信息, 欄位 u 會與前一則信息相同。
• 在極端的市場情況下, 產生端與推送端都會採用合併與丟包策略, 因此 u 的連續性不被保證。

---

連接

測試網 Testnet

• wss://stream-testnet.bybits.org/v5/public-sbe/spot
• wss://stream-testnet.bybits.org/v5/public-sbe/linear
• wss://stream-testnet.bybits.org/v5/public-sbe/inverse

主網 Mainnet

• xxxx.bybit-aws.com/v5/public-sbe/spot
• xxxx.bybit-aws.com/v5/public-sbe/linear
• xxxx.bybit-aws.com/v5/public-sbe/inverse

請在主網 SBE 連接時使用您的 專屬 MMWS host。

數據框架類型

SBE 信息會以 二進制 WebSocket frame 傳送 (opcode = 2)。

控制信息

控制信息 (subscription, unsubscription, ping/pong 等) 遵循 Bybit V5 WebSocket JSON 標準。

---

訂閱流程

送出訂閱請求

{
  "op": "subscribe",
  "args": ["ob.rpi.1.sbe.BTCUSDT"]
}

• Topic 格式: ob.rpi.1.sbe.<symbol>
• 範例 symbol: BTCUSDT, ETHUSDT 等。

訂閱確認

{
  "success": true,
  "ret_msg": "",
  "conn_id": "d30fdpbboasp1pjbe7r0",
  "req_id": "xxx",
  "op": "subscribe"
}

接收數據

b"R\x00 N\x01\x00\x00\x00\xdb\x84\xd0k\x00\x00\x00\x00f\xb7\x003\x99\x01\x00\x00\x02\x06\xa1\xcb\xa1\x00\x00\x00\x00\x00\xe7\xda\x0b\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x04\xc8\xa1\x00\x00\x00\x00\x00 N\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x008\x01\x00\x00\x00\x00\x00\x00v\xba\x003\x99\x01\x00\x00\x07BTCUSDT"

解碼示例

{
  "header": {
    "block_length": 82,
    "template_id": 20000,
    "schema_id": 1,
    "version": 0
  },
  "seq": 1808827611,
  "cts": 1757497309030,
  "price_exponent": 2,
  "size_exponent": 6,
  "ask_price": 1060342500,
  "ask_normal_size": 776935000000,
  "ask_rpi_size": 0,
  "bid_price": 1060250000,
  "bid_normal_size": 20000000000,
  "bid_rpi_size": 0,
  "u": 312,
  "ts": 1757497309814,
  "symbol": "BTCUSDT",
  "parsed_length": 98
}

---

SBE 信息結構

SBE XML Schema

• templateId = 20000 用來識別信息型別。
• 驗證 templateId = 20000 以確認這是一則 Level 1 訂單簿事件。

<?xml version="1.0" encoding="UTF-8"?>
<sbe:messageSchema xmlns:sbe="http://fixprotocol.io/2016/sbe"
                   xmlns:mbx="https://bybit-exchange.github.io/docs/v5/intro"
                   package="quote.sbe"
                   id="1"
                   version="0"
                   semanticVersion="1.0.0"
                   description="Bybit market data streams SBE message schema"
                   byteOrder="littleEndian"
                   headerType="messageHeader">

  <types>
    <composite name="messageHeader" description="Template ID and length of message root">
      <type name="blockLength" primitiveType="uint16"/>
      <type name="templateId" primitiveType="uint16"/>
      <type name="schemaId" primitiveType="uint16"/>
      <type name="version" primitiveType="uint16"/>
    </composite>

    <composite name="varString8" description="Variable length UTF-8 string.">
      <type name="length" primitiveType="uint8"/>
      <type name="varData"
            length="0"
            primitiveType="uint8"
            semanticType="String"
            characterEncoding="UTF-8"/>
    </composite>
  </types>

  <!-- Stream event for "ob.rpi.1.sbe.<symbol>" channel -->
  <sbe:message name="BestOBRpiEvent" id="20000">
    <field id="1"  name="ts"             type="int64" description="The timestamp in microseconds that the system generates the data"/>
    <field id="2"  name="seq"            type="int64" description="Cross sequence ID"/>
    <field id="3"  name="cts"            type="int64" description="The timestamp in microseconds from the matching engine when this orderbook data is produced."/>
    <field id="4"  name="u"              type="int64" description="Update Id"/>
    <field id="5"  name="askNormalPrice" type="int64" mbx:exponent="priceExponent" description="Mantissa for the best ask normal price"/>
    <field id="6"  name="askNormalSize"  type="int64" mbx:exponent="sizeExponent" description="Mantissa for the best ask normal size"/>
    <field id="7"  name="askRpiPrice"    type="int64" mbx:exponent="priceExponent" description="Mantissa for the best ask rpi price"/>
    <field id="8"  name="askRpiSize"     type="int64" mbx:exponent="sizeExponent" description="Mantissa for the best ask rpi size"/>
    <field id="9"  name="bidNormalPrice" type="int64" mbx:exponent="priceExponent" description="Mantissa for the best bid normal price"/>
    <field id="10" name="bidNormalSize"  type="int64" mbx:exponent="sizeExponent" description="Mantissa for the best bid normal size"/>
    <field id="11" name="bidRpiPrice"    type="int64" mbx:exponent="priceExponent" description="Mantissa for the best bid rpi price"/>
    <field id="12" name="bidRpiSize"     type="int64" mbx:exponent="sizeExponent" description="Mantissa for the best bid rpi size"/>
    <field id="13" name="priceExponent"  type="int8"  description="Price exponent for decimal point positioning"/>
    <field id="14" name="sizeExponent"   type="int8"  description="Size exponent for decimal point positioning"/>
    <data  id="55" name="symbol"         type="varString8"/>
  </sbe:message>
</sbe:messageSchema>

信息結構細節

信息標頭 (8 bytes)

Field	Type	Size (bytes)	Description
blockLength	uint16	2	信息主體長度 Message body length
templateId	uint16	2	固定值 = 20000 Fixed = 20000
schemaId	uint16	2	固定值 = 1 Fixed = 1
version	uint16	2	固定值 = 0 Fixed = 0

信息主體 (BestOBRpiEvent)

ID	Field	Type	Description
1	ts	int64	Snapshot 時間戳 (µs) Snapshot timestamp (µs)
2	seq	int64	唯一信息序號 Unique message sequence number
3	cts	int64	交易時間戳 (µs) Trade timestamp (µs)
4	u	int64	更新 ID Update ID
5	askNormalPrice	int64	最佳賣價 mantissa Best ask price mantissa
6	askNormalSize	int64	最佳賣量 (normal) mantissa Best ask size (normal) mantissa
7	askRpiPrice	int64	最佳 RPI 賣價 mantissa Best RPI ask price mantissa
8	askRpiSize	int64	最佳 RPI 賣量 mantissa Best RPI ask size mantissa
9	bidNormalPrice	int64	最佳買價 mantissa Best bid price mantissa
10	bidNormalSize	int64	最佳買量 (normal) mantissa Best bid size (normal) mantissa
11	bidRpiPrice	int64	最佳買價 (RPI) mantissa Best bid price (RPI) mantissa
12	bidRpiSize	int64	最佳買量 (RPI) mantissa Best bid size (RPI) mantissa
13	priceExponent	int8	價格 exponent Price exponent
14	sizeExponent	int8	數量 exponent Size exponent
55	symbol	varStr	交易對 (例如 BTCUSDT) Trading pair (e.g., BTCUSDT)

可變長度字符串 (varString8)

• uint8 length (1 byte) – 後續字元的數量 number of following characters
• UTF-8 字串 – 交易標的 trading symbol

範例 Example: "BTCUSDT" 的編碼為:

0x07 0x42 0x54 0x43 0x55 0x53 0x44 0x54
^    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
len  "BTCUSDT"

---

相關優化

優化 1

ts 與 cts 在 BBO SBE 中會以微秒 (µs) 顯示。

優化 2

調整欄位順序。

優化 3

新的欄位定義 (以賣方為例, 買方邏輯相同):

Field	Definition
askNormalPrice	無 RPI 訂單的最佳賣價 No RPI order best ask price
askNormalSize	無 RPI 訂單的最佳賣量 No RPI order best ask size
askRpiPrice	RPI 訂單的最佳賣價 RPI order best ask price
askRpiSize	RPI 訂單的最佳賣量 RPI order best ask size

目前的邏輯可能出現以下狀況:

Price	normalQty	rpiQty
1000	0	100

這表示沒有 RPI 權限的使用者無法得知實際可成交的價格, 導致這則信息對他們沒有實際意義。為了解決此問題, 我們對信息內容做了上述調整。

情況 1: askNormalSize != 0 && askRpiSize != 0

這是基於實際市場情況的正常回應, 與原先信息表達的含義相同。

示例:

Field	Definition	Example
askNormalPrice	無 RPI 訂單的最佳賣價 No RPI order best ask price	1000
askNormalSize	無 RPI 訂單的最佳賣量 No RPI order best ask size	200
askRpiPrice	RPI 訂單的最佳賣價 RPI order best ask price	1000
askRpiSize	RPI 訂單的最佳賣量 RPI order best ask size	300

情況 2: askNormalSize != 0 && askRpiSize == 0

askRpiPrice 會被指定為 askNormalPrice, 並且 askRpiSize = 0。
在此情況下, 不會再向更深的價位搜尋 askRpiPrice。

示例:

Field	Definition	Example	Note
askNormalPrice	無 RPI 訂單的最佳賣價 No RPI order best ask price	1000
askNormalSize	無 RPI 訂單的最佳賣量 No RPI order best ask size	200
askRpiPrice	RPI 訂單的最佳賣價 RPI order best ask price	1000	This itself has no meaning. The price field is assigned a non-RPI sell price.
askRpiSize	RPI 訂單的最佳賣量 RPI order best ask size	0

情況 3: askNormalSize == 0 && askRpiSize != 0

此時 askNormalPrice = 真正的非 RPI 最佳賣價。
在這種情況下, 會回傳並使用 askNormalPrice。

示例:

Field	Definition	Example
askNormalPrice	無 RPI 訂單的最佳賣價 No RPI order best ask price	1200
askNormalSize	無 RPI 訂單的最佳賣量 No RPI order best ask size	100
askRpiPrice	RPI 訂單的最佳賣價 RPI order best ask price	1000
askRpiSize	RPI 訂單的最佳賣量 RPI order best ask size	20

情況 4

當市場極度冷清, 完全沒有流動性時, 不會推送任何 BBO 信息。

---

接入示例

import json
import logging
import struct
import threading
import time
from datetime import datetime
from typing import Dict, Any

import websocket

logging.basicConfig(
    filename="logfile_wrapper.log",
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)

# Change symbol/topic as you wish
TOPIC = "ob.rpi.1.sbe.BTCUSDT"
WS_URL = "wss://stream-testnet.bybits.org/v5/public-sbe/spot"


class SBEBestOBRpiParser:
    """
    Parser for BestOBRpiEvent (template_id = 20000) per XML schema:

    ts(int64), seq(int64), cts(int64), u(int64),
    askNormalPrice(int64), askNormalSize(int64),
    askRpiPrice(int64), askRpiSize(int64),
    bidNormalPrice(int64), bidNormalSize(int64),
    bidRpiPrice(int64), bidRpiSize(int64),
    priceExponent(int8), sizeExponent(int8),
    symbol(varString8)

    All values are little-endian.
    """

    def __init__(self) -> None:
        # Header: blockLength, templateId, schemaId, version
        self.header_fmt = "<HHHH"
        self.header_sz = struct.calcsize(self.header_fmt)

        # 12 x int64 + 2 x int8:
        # ts, seq, cts, u,
        # askNormalPrice, askNormalSize, askRpiPrice, askRpiSize,
        # bidNormalPrice, bidNormalSize, bidRpiPrice, bidRpiSize,
        # priceExponent, sizeExponent
        self.body_fmt = "<" + ("q" * 12) + "bb"
        self.body_sz = struct.calcsize(self.body_fmt)

        self.target_template_id = 20000

    def _parse_header(self, data: bytes) -> Dict[str, Any]:
        if len(data) < self.header_sz:
            raise ValueError("insufficient data for SBE header")

        block_length, template_id, schema_id, version = struct.unpack_from(
            self.header_fmt, data, 0
        )

        return {
            "block_length": block_length,
            "template_id": template_id,
            "schema_id": schema_id,
            "version": version,
        }

    @staticmethod
    def _parse_varstring8(data: bytes, offset: int) -> tuple[str, int]:
        if offset + 1 > len(data):
            raise ValueError("insufficient data for varString8 length")

        (length,) = struct.unpack_from("<B", data, offset)
        offset += 1

        if offset + length > len(data):
            raise ValueError("insufficient data for varString8 bytes")

        s = data[offset : offset + length].decode("utf-8")
        offset += length
        return s, offset

    @staticmethod
    def _apply_exponent(value: int, exponent: int) -> float:
        # Exponent is for decimal point positioning.
        # If exponent = 2 and value=1060342500 -> 10603425.00
        return value / (10 ** exponent) if exponent >= 0 else value * (
            10 ** (-exponent)
        )

    def parse(self, data: bytes) -> Dict[str, Any]:
        hdr = self._parse_header(data)
        if hdr["template_id"] != self.target_template_id:
            raise NotImplementedError(
                f"unsupported template_id={hdr['template_id']}"
            )

        if len(data) < self.header_sz + self.body_sz:
            raise ValueError("insufficient data for BestOBRpiEvent body")

        fields = struct.unpack_from(self.body_fmt, data, self.header_sz)
        (
            ts,
            seq,
            cts,
            u,
            ask_np_m,
            ask_ns_m,
            ask_rp_m,
            ask_rs_m,
            bid_np_m,
            bid_ns_m,
            bid_rp_m,
            bid_rs_m,
            price_exp,
            size_exp,
        ) = fields

        offset = self.header_sz + self.body_sz
        symbol, offset = self._parse_varstring8(data, offset)

        # Apply exponents
        ask_np = self._apply_exponent(ask_np_m, price_exp)
        ask_ns = self._apply_exponent(ask_ns_m, size_exp)
        ask_rp = self._apply_exponent(ask_rp_m, price_exp)
        ask_rs = self._apply_exponent(ask_rs_m, size_exp)
        bid_np = self._apply_exponent(bid_np_m, price_exp)
        bid_ns = self._apply_exponent(bid_ns_m, size_exp)
        bid_rp = self._apply_exponent(bid_rp_m, price_exp)
        bid_rs = self._apply_exponent(bid_rs_m, size_exp)

        return {
            "header": hdr,
            "ts": ts,
            "seq": seq,
            "cts": cts,
            "u": u,
            "price_exponent": price_exp,
            "size_exponent": size_exp,
            "symbol": symbol,
            # Normal book (best)
            "ask_normal_price": ask_np,
            "ask_normal_size": ask_ns,
            "bid_normal_price": bid_np,
            "bid_normal_size": bid_ns,
            # RPI book (best)
            "ask_rpi_price": ask_rp,
            "ask_rpi_size": ask_rs,
            "bid_rpi_price": bid_rp,
            "bid_rpi_size": bid_rs,
            "parsed_length": offset,
        }


parser = SBEBestOBRpiParser()


# --------------------------- WebSocket handlers ---------------------------

def on_message(ws, message):
    try:
        # Binary SBE frames; text frames for control/acks/errors
        if isinstance(message, (bytes, bytearray)):
            decoded = parser.parse(message)
            logging.info(
                "SBE %s seq=%s u=%s "
                "NORM bid=%.8f@%.8f ask=%.8f@%.8f "
                "RPI bid=%.8f@%.8f ask=%.8f@%.8f ts=%s",
                decoded["symbol"],
                decoded["seq"],
                decoded["u"],
                decoded["bid_normal_price"],
                decoded["bid_normal_size"],
                decoded["ask_normal_price"],
                decoded["ask_normal_size"],
                decoded["bid_rpi_price"],
                decoded["bid_rpi_size"],
                decoded["ask_rpi_price"],
                decoded["ask_rpi_size"],
                decoded["ts"],
            )
            print(
                f"{decoded['symbol']} u={decoded['u']} "
                f"NORM: {decoded['bid_normal_price']:.8f} x {decoded['bid_normal_size']:.8f} "
                f"| {decoded['ask_normal_price']:.8f} x {decoded['ask_normal_size']:.8f} "
                f"RPI: {decoded['bid_rpi_price']:.8f} x {decoded['bid_rpi_size']:.8f} "
                f"| {decoded['ask_rpi_price']:.8f} x {decoded['ask_rpi_size']:.8f} "
                f"(seq={decoded['seq']} ts={decoded['ts']})"
            )
        else:
            try:
                obj = json.loads(message)
                logging.info("TEXT %s", obj)
                print(obj)
            except json.JSONDecodeError:
                logging.warning("non-JSON text frame: %r", message)
    except Exception as e:
        logging.exception("decode error: %s", e)
        print("decode error:", e)


def on_error(ws, error):
    print("WS error:", error)
    logging.error("WS error: %s", error)


def on_close(ws, *_):
    print("### connection closed ###")
    logging.info("connection closed")


def on_open(ws):
    print("opened")
    sub = {"op": "subscribe", "args": [TOPIC]}
    ws.send(json.dumps(sub))
    print("subscribed:", TOPIC)

    threading.Thread(target=ping_per, args=(ws,), daemon=True).start()
    threading.Thread(target=manage_subscription, args=(ws,), daemon=True).start()


def manage_subscription(ws):
    # demo: unsubscribe/resubscribe once
    time.sleep(20)
    ws.send(json.dumps({"op": "unsubscribe", "args": [TOPIC]}))
    print("unsubscribed:", TOPIC)
    time.sleep(5)
    ws.send(json.dumps({"op": "subscribe", "args": [TOPIC]}))
    print("resubscribed:", TOPIC)


def ping_per(ws):
    while True:
        try:
            ws.send(json.dumps({"op": "ping"}))
        except Exception:
            return
        time.sleep(10)


def on_pong(ws, *_):
    print("pong received")


def on_ping(ws, *_):
    print("ping received @", datetime.now().strftime("%Y-%m-%d %H:%M:%S"))


def connWS():
    ws = websocket.WebSocketApp(
        WS_URL,
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close,
        on_ping=on_ping,
        on_pong=on_pong,
    )
    ws.run_forever(ping_interval=20, ping_timeout=10)


if __name__ == "__main__":
    websocket.enableTrace(False)
    connWS()