Initialization

npm install centrifuge

pip install centrifuge-python aiohttp

import { Centrifuge } from "centrifuge";

const RELAYER_URL = "https://api.sx.bet"; // Mainnet — use https://api.toronto.sx.bet for testnet
const WS_URL = "wss://realtime.sx.bet/connection/websocket"; // Mainnet — use wss://realtime.toronto.sx.bet/connection/websocket for testnet

async function fetchToken(apiKey) {
  const res = await fetch(`${RELAYER_URL}/user/realtime-token-v3/api-key`, {
    headers: { "x-sx-api-key": apiKey },
  });
  if (!res.ok) {
    const body = await res.text();
    throw new Error(`Token endpoint returned ${res.status}: ${body}`);
  }
  const { token } = await res.json();
  return token;
}

const client = new Centrifuge(WS_URL, {
  getToken: () => fetchToken(YOUR_API_KEY),
});

client.connect();

import asyncio
import os
import aiohttp
from centrifuge import Client

RELAYER_URL = "https://api.sx.bet"  # Mainnet — use https://api.toronto.sx.bet for testnet
WS_URL = "wss://realtime.sx.bet/connection/websocket"  # Mainnet — use wss://realtime.toronto.sx.bet/connection/websocket for testnet

# Must be `async def`, and use an async HTTP client. A blocking call here stalls the
# event loop, and with it every subscription on the connection.
async def fetch_token(ctx=None):
    async with aiohttp.ClientSession() as session:
        async with session.get(
            f"{RELAYER_URL}/user/realtime-token-v3/api-key",
            headers={"x-sx-api-key": os.environ["SX_API_KEY"]},
        ) as resp:
            if not resp.ok:
                body = await resp.text()
                raise Exception(f"Token endpoint returned {resp.status}: {body}")
            data = await resp.json()
            return data["token"]

async def main():
    client = Client(WS_URL, get_token=fetch_token)
    await client.connect()
    await asyncio.Future()  # keep running

asyncio.run(main())

const sub = client.newSubscription("channel:name", { recoverable: true });

sub.on("publication", (ctx) => {
  const data = ctx.data;
  // handle incoming message
});

sub.subscribe();

import asyncio
from centrifuge import Client, PublicationContext, SubscriptionEventHandler, SubscriptionOptions

async def on_publication(ctx: PublicationContext) -> None:
    print(ctx.data)

async def main():
    client = Client(WS_URL, get_token=fetch_token)
    await client.connect()
    handler = SubscriptionEventHandler(on_publication=on_publication)
    options = SubscriptionOptions(recoverable=True)
    sub = client.new_subscription("channel:name", handler, options)
    await sub.subscribe()
    await asyncio.Future()  # keep running

asyncio.run(main())

const sub = client.newSubscription(`orderbook_v3:${marketHash}`);

// Every publication runs through the version rule; stale ones are discarded.
sub.on("publication", (ctx) => applyUpdate(ctx.data));

sub.on("subscribed", async (ctx) => {
  // recovered: true means the book was replayed — no REST seed needed.
  if (ctx.recovered) return;

  const { data } = await fetch(
    `https://api.sx.bet/orderbook-v3/snapshot?marketHash=${marketHash}`
  ).then((r) => r.json());

  // The snapshot goes through the same version rule, so a newer live update
  // that arrived during the fetch is not clobbered. `data` is flat — the same
  // shape a publication carries, with the level arrays beside `version`.
  applySnapshot(marketHash, data.version, data);
});

sub.subscribe();
client.connect();

import aiohttp
from centrifuge import (
    PublicationContext,
    SubscribedContext,
    SubscriptionEventHandler,
)

class BookHandler(SubscriptionEventHandler):
    def __init__(self, market_hash: str) -> None:
        self.market_hash = market_hash

    async def on_subscribed(self, ctx: SubscribedContext) -> None:
        # recovered: True means the book was replayed — no REST seed needed.
        if ctx.recovered:
            return
        async with aiohttp.ClientSession() as session:
            async with session.get(
                f"https://api.sx.bet/orderbook-v3/snapshot?marketHash={self.market_hash}"
            ) as res:
                body = await res.json()
        data = body["data"]
        # The snapshot runs through the same version rule as live updates, and
        # carries the same flat shape a publication does.
        apply_snapshot(self.market_hash, data["version"], data)

    async def on_publication(self, ctx: PublicationContext) -> None:
        apply_update(ctx.data)

async def subscribe_book(client, market_hash: str) -> None:
    sub = client.new_subscription(
        f"orderbook_v3:{market_hash}", BookHandler(market_hash)
    )
    await sub.subscribe()

sub.unsubscribe();
sub.removeAllListeners();
client.removeSubscription(sub);

await sub.unsubscribe()
client.remove_subscription(sub)