Live stream

import json
import os
import time

import requests

NEXUS = "https://app.skylit.ai"
KEY = os.environ["SKYLIT_API_KEY"]


def events(last_id=None):
    """Yield (id, event, data) from the stream until it ends."""
    headers = {"Authorization": f"Bearer {KEY}", "Accept": "text/event-stream"}
    if last_id:
        headers["Last-Event-ID"] = last_id
    # The read timeout is 45 s: three missed heartbeats means a dead stream.
    with requests.get(f"{NEXUS}/api/nexus/v1/trading/stream",
                      headers=headers, stream=True, timeout=(10, 45)) as r:
        if r.status_code != 200:
            raise RuntimeError(r.json()["error"])
        eid, event, data = None, None, []
        for line in r.iter_lines(decode_unicode=True):
            if line == "":
                if event:
                    yield eid, event, json.loads("".join(data)) if data else None
                eid, event, data = None, None, []
            elif line.startswith("id: "):
                eid = line[4:]
            elif line.startswith("event: "):
                event = line[7:]
            elif line.startswith("data: "):
                data.append(line[6:])


def resync():
    # Re-read what you hold: GET /api/nexus/v1/trades?status=open,
    # GET /api/nexus/v1/trading/wallets/main/orders/working, and so on.
    pass


last_id = None
while True:
    try:
        for eid, event, data in events(last_id):
            if eid:
                last_id = eid
            if event == "resync":
                resync()
            elif event == "trade":
                print(data["change"], data["trade"]["ticker"], data["fill"])
            elif event == "order":
                print(data["assetClass"], data["change"], data["order"]["id"])
            elif event == "closed":
                raise SystemExit(f"stream closed: {data['reason']}")
    except (requests.ConnectionError, requests.Timeout):
        pass  # dropped or went quiet: reconnect below
    time.sleep(2)

const NEXUS = "https://app.skylit.ai";
const KEY = process.env.SKYLIT_API_KEY;
let lastId = null;

async function resync() {
  // Re-read what you hold with the REST routes.
}

async function stream() {
  const headers = { Authorization: `Bearer ${KEY}`, Accept: "text/event-stream" };
  if (lastId) headers["Last-Event-ID"] = lastId;

  // Three missed heartbeats (45 s of nothing) means a dead stream.
  const quiet = new AbortController();
  let timer = setTimeout(() => quiet.abort(), 45_000);

  const res = await fetch(`${NEXUS}/api/nexus/v1/trading/stream`, { headers, signal: quiet.signal });
  if (!res.ok) throw new Error(JSON.stringify((await res.json()).error));

  const decoder = new TextDecoder();
  let buf = "";
  try {
    for await (const chunk of res.body) {
      clearTimeout(timer);
      timer = setTimeout(() => quiet.abort(), 45_000);
      buf += decoder.decode(chunk, { stream: true });
      let end;
      while ((end = buf.indexOf("\n\n")) >= 0) {
        const frame = buf.slice(0, end);
        buf = buf.slice(end + 2);
        let id = "", event = "", data = "";
        for (const line of frame.split("\n")) {
          if (line.startsWith("id: ")) id = line.slice(4);
          else if (line.startsWith("event: ")) event = line.slice(7);
          else if (line.startsWith("data: ")) data += line.slice(6);
        }
        if (id) lastId = id;
        const msg = data ? JSON.parse(data) : null;
        if (event === "resync") await resync();
        else if (event === "trade") console.log(msg.change, msg.trade.ticker, msg.fill);
        else if (event === "order") console.log(msg.assetClass, msg.change, msg.order.id);
        else if (event === "closed") {
          console.error(`stream closed: ${msg.reason}`);
          process.exit(1);
        }
      }
    }
  } finally {
    clearTimeout(timer);
  }
}

(async () => {
  for (;;) {
    try {
      await stream();
    } catch (err) {
      console.error(err.message); // dropped or went quiet: reconnect below
    }
    await new Promise((r) => setTimeout(r, 2000));
  }
})();