Stream Robinhood Chain trades over WebSocket

// trades-stream.mjs: node trades-stream.mjs  (Node.js 22+, built-in WebSocket)
const KEY = process.env.CRIUS_API_KEY;
const API = "https://api.crius.sh/v1/robinhood";
const TOKEN = "0x322f0929c4625ed5bad873c95208d54e1c003b2d"; // TSLA, Tesla tokenized stock
const CHANNELS = [`trades:${TOKEN}`];
const MIN_USD = 100; // only print trades of $100 or more

const lastSeq = {}; // channel -> last seq seen
const seen = new Map(); // trade id -> trade (de-duplication and reorg undo)
let lastTs = 0; // chain time of the newest trade we hold
let lastMsgAt = 0; // wall-clock time of the last message
let backoff = 1000;

function onTrade(t, source) {
  if (seen.has(t.id)) return;
  seen.set(t.id, t);
  if (seen.size > 50_000) seen.delete(seen.keys().next().value);
  lastTs = Math.max(lastTs, t.ts);
  if (t.value_usd >= MIN_USD) {
    console.log(`${source} ${new Date(t.ts).toISOString()} ${t.type} $${t.value_usd} @ $${t.price_usd} by ${t.trader}`);
  }
}

async function backfill() {
  if (!lastTs) return;
  const res = await fetch(`${API}/token/${TOKEN}/trades?from_ts=${lastTs}&limit=500`, { headers: { "x-api-key": KEY } });
  const { data } = await res.json();
  for (const t of (data ?? []).reverse()) onTrade(t, "rest"); // the tape is newest first
}

function connect() {
  const ws = new WebSocket(`wss://api.crius.sh/v1/robinhood/stream?api_key=${KEY}`);
  let idle;
  const touch = () => {
    clearTimeout(idle);
    idle = setTimeout(() => ws.close(), 45_000); // heartbeats arrive every 15 s
  };

  ws.onopen = () => {
    backoff = 1000;
    touch();
    const offline = lastMsgAt ? Date.now() - lastMsgAt : 0;
    ws.send(JSON.stringify({ op: "subscribe", channels: CHANNELS }));
    if (Object.keys(lastSeq).length) ws.send(JSON.stringify({ op: "resume", seqs: lastSeq }));
    if (offline > 55_000) backfill().catch(console.error); // longer than the 60 s replay buffer
  };

  ws.onmessage = (e) => {
    touch();
    lastMsgAt = Date.now();
    const m = JSON.parse(e.data);
    if (m.seq) lastSeq[m.ch] = m.seq;
    if (m.ch === "error") {
      if (m.data.code === "lagged") ws.send(JSON.stringify({ op: "resume", seqs: lastSeq }));
      else console.error(m.data);
      return;
    }
    if (m.ch === "reorg") return console.warn(`reorg: blocks ${m.data.removed_blocks.join(", ")} replaced`);
    if (!m.ch.startsWith("trades:")) return; // hello, subscribed, heartbeat, finality, resumed
    if (m.data.removed) seen.delete(m.data.id); // undo whatever you did for this trade
    else onTrade(m.data, "live");
  };

  ws.onclose = () => {
    clearTimeout(idle);
    setTimeout(connect, backoff + Math.random() * 500);
    backoff = Math.min(backoff * 2, 30_000);
  };
  ws.onerror = () => ws.close();
}

connect();

# trades_stream.py: pip install websockets requests; python trades_stream.py
import asyncio, json, os, random, time

import requests
import websockets

KEY = os.environ["CRIUS_API_KEY"]
API = "https://api.crius.sh/v1/robinhood"
TOKEN = "0x322f0929c4625ed5bad873c95208d54e1c003b2d"  # TSLA, Tesla tokenized stock
CHANNELS = [f"trades:{TOKEN}"]

last_seq, seen = {}, {}
state = {"last_ts": 0, "last_msg": 0.0}


def on_trade(t, source):
    if t["id"] in seen:
        return
    seen[t["id"]] = t
    state["last_ts"] = max(state["last_ts"], t["ts"])
    print(source, t["type"], f"${t['value_usd']}", f"@ ${t['price_usd']}", t["trader"])


def backfill():
    if not state["last_ts"]:
        return
    r = requests.get(f"{API}/token/{TOKEN}/trades", params={"from_ts": state["last_ts"], "limit": 500},
                     headers={"x-api-key": KEY}, timeout=10)
    for t in reversed(r.json().get("data") or []):  # the tape is newest first
        on_trade(t, "rest")


async def run():
    backoff = 1
    while True:
        try:
            async with websockets.connect(f"wss://api.crius.sh/v1/robinhood/stream?api_key={KEY}") as ws:
                backoff = 1
                offline = time.time() - state["last_msg"] if state["last_msg"] else 0
                await ws.send(json.dumps({"op": "subscribe", "channels": CHANNELS}))
                if last_seq:
                    await ws.send(json.dumps({"op": "resume", "seqs": last_seq}))
                if offline > 55:  # longer than the 60 s replay buffer
                    await asyncio.to_thread(backfill)
                while True:
                    m = json.loads(await asyncio.wait_for(ws.recv(), timeout=45))
                    state["last_msg"] = time.time()
                    if m.get("seq"):
                        last_seq[m["ch"]] = m["seq"]
                    ch, data = m["ch"], m.get("data") or {}
                    if ch == "error":
                        if data.get("code") == "lagged":
                            await ws.send(json.dumps({"op": "resume", "seqs": last_seq}))
                        else:
                            print("error", data)
                    elif ch.startswith("trades:"):
                        if data.get("removed"):
                            seen.pop(data["id"], None)  # undo whatever you did for this trade
                        else:
                            on_trade(data, "live")
        except (OSError, asyncio.TimeoutError, websockets.WebSocketException) as e:
            print("reconnecting:", e)
        await asyncio.sleep(backoff + random.random() / 2)
        backoff = min(backoff * 2, 30)


asyncio.run(run())