Live stream

import json, time, urllib.error, urllib.request

URL = "https://api.routeur.app/v1/stream?topics=trades,leads&min_usd=100"
KEY = "..."  # your API key

def handle(name, data):
    if name == "reset":
        resync_from_rest()      # rebuild state from /v1/opportunities etc.
    elif name in ("trade", "lead"):
        print(name, data["market_id"] if name == "trade" else data["net_edge_usd"])

last_id = None
while True:
    headers = {"X-API-Key": KEY, "Accept": "text/event-stream"}
    if last_id:
        headers["Last-Event-ID"] = last_id
    req = urllib.request.Request(URL, headers=headers)
    try:
        with urllib.request.urlopen(req, timeout=45) as resp:   # 45 s of silence = dead
            event, data, event_id = None, [], None
            for raw in resp:
                line = raw.decode().rstrip("\n")
                if line == "":                                  # blank line ends an event
                    if event and data:
                        payload = json.loads("\n".join(data))
                        if event_id:
                            last_id = event_id
                        if event == "reconnect":
                            last_id = payload["cursor"]
                            break                               # reconnect with the cursor
                        if event == "error":
                            raise SystemExit(payload["message"])
                        handle(event, payload)
                    event, data, event_id = None, [], None
                elif line.startswith(":"):
                    continue                                    # heartbeat
                elif line.startswith("id:"):
                    event_id = line[3:].strip()
                elif line.startswith("event:"):
                    event = line[6:].strip()
                elif line.startswith("data:"):
                    data.append(line[5:].strip())
    except (urllib.error.URLError, TimeoutError, ConnectionError) as e:
        print("dropped:", e)
    time.sleep(3)                                               # the stream's retry interval

const URL = "https://api.routeur.app/v1/stream?topics=trades,leads&min_usd=100";
const KEY = "..."; // your API key

function handle(name, data) {
  if (name === "reset") resyncFromRest();            // rebuild state from /v1/opportunities etc.
  else if (name === "trade" || name === "lead") console.log(name, data);
}

let lastId = null;
for (;;) {
  const headers = { "X-API-Key": KEY, Accept: "text/event-stream" };
  if (lastId) headers["Last-Event-ID"] = lastId;
  try {
    const res = await fetch(URL, { headers });
    if (!res.ok) throw new Error(`${res.status} ${await res.text()}`);
    let buffer = "", event = null, data = [], id = null, done = false;
    let timer = null;
    const reader = res.body.getReader();
    const armSilenceTimer = () => {
      clearTimeout(timer);
      timer = setTimeout(() => reader.cancel("45 s of silence"), 45_000);
    };
    armSilenceTimer();
    while (!done) {
      const chunk = await reader.read();
      if (chunk.done) break;
      armSilenceTimer();
      buffer += new TextDecoder().decode(chunk.value);
      let nl;
      while ((nl = buffer.indexOf("\n")) >= 0) {
        const line = buffer.slice(0, nl); buffer = buffer.slice(nl + 1);
        if (line === "") {
          if (event && data.length) {
            const payload = JSON.parse(data.join("\n"));
            if (id) lastId = id;
            if (event === "reconnect") { lastId = payload.cursor; done = true; break; }
            if (event === "error") throw new Error(payload.message);
            handle(event, payload);
          }
          event = null; data = []; id = null;
        } else if (line.startsWith(":")) continue;             // heartbeat
        else if (line.startsWith("id:")) id = line.slice(3).trim();
        else if (line.startsWith("event:")) event = line.slice(6).trim();
        else if (line.startsWith("data:")) data.push(line.slice(5).trim());
      }
    }
    clearTimeout(timer);
    reader.cancel();
  } catch (e) {
    console.error("dropped:", e.message);
  }
  await new Promise((r) => setTimeout(r, 3000));               // the stream's retry interval
}