WebSocket quickstart

wss://api.omnibook.xyz/v1/ws

<timestamp>\nGET\n/v1/ws\n

{
  "id": 1,
  "cmd": "subscribe",
  "params": { "channels": ["orderbook_delta", "trades"], "market_ids": [66] }
}

{ "id": 1, "type": "subscribed", "sid": 3, "channel": "orderbook_delta" }

#!/usr/bin/env python3
"""Connect, subscribe to the active round's book, print frames."""
import base64, hashlib, hmac, json, time
import websocket  # pip install websocket-client
import requests

BASE = "https://api.omnibook.xyz"
WS_URL = "wss://api.omnibook.xyz/v1/ws"
API_KEY = "<your alphanumeric API key>"
SECRET = bytes.fromhex("<your 64-char hex secret>")


def sign(method: str, target: str, body: bytes = b"") -> dict:
    ts = str(int(time.time() * 1000))
    canonical = f"{ts}\n{method}\n{target}\n".encode() + body
    sig = base64.b64encode(
        hmac.new(SECRET, canonical, hashlib.sha256).digest()
    ).decode()
    return {
        "DX-ACCESS-KEY": API_KEY,
        "DX-ACCESS-TIMESTAMP": ts,
        "DX-ACCESS-SIGNATURE": sig,
    }


def active_market_id() -> int:
    target = "/v1/rounds?limit=1"
    r = requests.get(BASE + target, headers=sign("GET", target))
    r.raise_for_status()
    round_ = r.json()["rounds"][0]
    assert round_["status"] == "trading", round_
    return int(round_["market_id"])


def main() -> None:
    market_id = active_market_id()
    headers = sign("GET", "/v1/ws")
    ws = websocket.create_connection(
        WS_URL,
        header=[f"{k}: {v}" for k, v in headers.items()],
    )

    ws.send(json.dumps({
        "id": 1,
        "cmd": "subscribe",
        "params": {
            "channels": ["orderbook_delta", "trades", "rounds", "user"],
            "market_ids": [market_id],
        },
    }))

    last_seq: dict[int, int] = {}
    while True:
        frame = json.loads(ws.recv())
        print(frame)

        if frame.get("type") == "error":
            continue

        sid, seq = frame.get("sid"), frame.get("seq")
        if sid is None or seq is None:
            continue

        prev = last_seq.get(sid)
        if prev is not None and seq != prev + 1:
            # Public gap — re-snapshot without tearing down the subscription.
            ws.send(json.dumps({
                "id": 2,
                "cmd": "update_subscription",
                "params": {
                    "sid": sid,
                    "action": "get_snapshot",
                    "market_ids": [market_id],
                },
            }))
        last_seq[sid] = seq


if __name__ == "__main__":
    main()