import asyncio, os, requests
from centrifuge import (
Client, SubscriptionEventHandler, SubscriptionOptions,
PublicationContext, SubscribedContext,
)
API = "https://api.sx.bet" # REST — no API key needed
WS = "wss://realtime.sx.bet/connection/websocket" # WS token fetched below
# The WebSocket requires a realtime token from /user/realtime-token/api-key
# (this is the only place an API key is used — every REST call below is public).
def fetch_token():
res = requests.get(
f"{API}/user/realtime-token/api-key",
headers={"x-api-key": os.environ["SX_API_KEY"]},
)
res.raise_for_status()
return res.json()["token"]
# Decode one maker order → taker price + taker/maker liquidity.
# percentageOdds = maker_implied * 10^20; taker_implied = 1 - maker_implied.
# A maker betting outcome ONE is quoting a TAKER price on outcome TWO.
def decode(o):
maker_implied = int(o["percentageOdds"]) / 1e20
taker_implied = 1 - maker_implied
if not (0 < taker_implied < 1):
return None
decimal_odds = 1 / taker_implied
maker_usdc = int(o["totalBetSize"]) / 1e6
taker_usdc = maker_usdc / (decimal_odds - 1)
taker_side = "two" if o["isMakerBettingOutcomeOne"] else "one"
return decimal_odds, maker_usdc, taker_usdc, taker_side
def print_book(orders):
book = {"one": [], "two": []}
for o in orders.values():
if o.get("status", "ACTIVE") != "ACTIVE":
continue
row = decode(o)
if row: book[row[3]].append(row)
for side in ("one", "two"):
rows = sorted(book[side], key=lambda r: -r[0])
if not rows:
print(f" {side}: —"); continue
taker = sum(r[2] for r in rows); maker = sum(r[1] for r in rows)
print(f" {side}: best {rows[0][0]:.2f}"
f" · taker_liq={taker:,.0f} · maker_liq={maker:,.0f} USDC")
async def stream(market_hash):
orders, seen, buffer = {}, set(), []
ready = False
# Full REST snapshot — public endpoint, no API key needed.
def reseed():
nonlocal orders
data = requests.get(f"{API}/orders",
params={"marketHashes": market_hash}).json()["data"]
orders = {o["orderHash"]: {**o, "status": "ACTIVE"}
for o in data if o["marketHash"] == market_hash}
def apply_update(batch):
for u in batch:
prev = orders.get(u["orderHash"])
# Compare updateTime as int — drop stale messages.
if prev and "updateTime" in prev and "updateTime" in u:
if int(u["updateTime"]) < int(prev["updateTime"]): continue
orders[u["orderHash"]] = u
print_book(orders)
async def on_subscribed(ctx: SubscribedContext):
nonlocal ready, buffer
# If Centrifugo replayed the full gap, keep local state as-is.
if ctx.was_recovering and ctx.recovered:
ready = True; return
ready = False; buffer = []
reseed() # fresh connect OR history pruned
for b in buffer: apply_update(b) # drain anything that came in meanwhile
buffer = []; ready = True
print_book(orders)
async def on_publication(ctx: PublicationContext):
msg_id = (ctx.tags or {}).get("messageId")
if msg_id: # dedupe — at-least-once delivery
if msg_id in seen: return
seen.add(msg_id)
if not ready: buffer.append(ctx.data)
else: apply_update(ctx.data)
client = Client(WS, get_token=fetch_token)
handler = SubscriptionEventHandler(on_subscribed=on_subscribed,
on_publication=on_publication)
opts = SubscriptionOptions(positioned=True, recoverable=True)
sub = client.new_subscription(f"order_book:market_{market_hash}",
handler, opts)
await client.connect()
await sub.subscribe()
await asyncio.Future() # run forever
# Pick a market (public REST — no API key needed) and stream it live.
markets = requests.get(f"{API}/markets/active",
params={"onlyMainLine": "true", "pageSize": 5}
).json()["data"]["markets"]
asyncio.run(stream(markets[0]["marketHash"]))
