A resumable stream consumer

import json
import os
import time
import requests

API = "https://api.routeur.app"
TOPICS = "trades,leads"
CURSOR_FILE = "routeur-cursor.txt"       # survives restarts, so nothing is missed or repeated
SILENCE_LIMIT = 45                        # heartbeats come every 15 s; three missed means the connection is dead


def load_cursor():
    try:
        return open(CURSOR_FILE).read().strip() or None
    except FileNotFoundError:
        return None


def save_cursor(cursor):
    with open(CURSOR_FILE, "w") as f:
        f.write(cursor)


def events(response):
    """Yield (id, event, data) from an SSE body; comment lines are heartbeats."""
    event_id, name, data = None, None, []
    for raw in response.iter_lines(decode_unicode=True):
        if raw is None:
            continue
        if raw == "":
            if name or data:
                yield event_id, name, "\n".join(data)
            event_id, name, data = None, None, []
        elif raw.startswith(":"):
            yield None, "heartbeat", ""
        elif raw.startswith("id:"):
            event_id = raw[3:].strip()
        elif raw.startswith("event:"):
            name = raw[6:].strip()
        elif raw.startswith("data:"):
            data.append(raw[5:].lstrip())
        elif raw.startswith("retry:"):
            pass                                        # the server asks for 3 s; we back off from there


def handle(name, payload):
    if name == "trade":
        print(f"trade  {payload['venue']} {payload['market_id']} {payload['taker_action']} {payload['outcome']} "
              f"{float(payload['quantity']):.0f} @ {float(payload['price_usd']) * 100:.0f}¢ = ${float(payload['notional_usd']):,.2f}")
    elif name == "lead":
        legs = " + ".join(f"{l['side']} {l['venue']} {l['market_id']} @ {l['ask_usd']}" for l in payload["legs"])
        print(f"lead   run {payload['graph_run']} {payload['relation']} net {payload['net_edge_usd']}: {legs}")


def run():
    backoff = 3
    while True:
        cursor = load_cursor()
        headers = {"X-API-Key": os.environ["ROUTEUR_API_KEY"], "Accept": "text/event-stream"}
        if cursor:
            headers["Last-Event-ID"] = cursor
        try:
            with requests.get(f"{API}/v1/stream", params={"topics": TOPICS}, headers=headers,
                              stream=True, timeout=(10, SILENCE_LIMIT)) as res:
                if res.status_code in (429, 503):
                    wait = int(res.headers.get("Retry-After", backoff))
                    print(f"{res.status_code} {res.json()['error']['code']}: waiting {wait}s")
                    time.sleep(wait)
                    continue
                res.raise_for_status()
                backoff = 3                              # connected: reset the backoff
                for event_id, name, data in events(res):
                    if name == "heartbeat":
                        continue                         # the read timeout is the liveness check
                    payload = json.loads(data) if data else {}
                    if name == "ready":
                        print(f"ready: topics {payload['topics']}, resumed from {payload.get('resumed_from', 'now')}, "
                              f"stream ends in {payload['reconnect_after_seconds']}s")
                    elif name == "reset":
                        print(f"reset: {payload['message']}")     # resync from REST here, then carry on
                    elif name == "reconnect":
                        print(f"reconnect ({payload['reason']}): resuming from {payload['cursor']}")
                        save_cursor(payload["cursor"])
                        break                            # the server is closing; reopen at once
                    elif name == "error":
                        print(f"error {payload['code']}: {payload['message']}")
                        return
                    else:
                        handle(name, payload)
                    if event_id:
                        save_cursor(event_id)
        except (requests.ConnectionError, requests.Timeout) as err:
            print(f"connection lost ({type(err).__name__}): retrying in {backoff}s")
            time.sleep(backoff)
            backoff = min(backoff * 2, 60)
            continue
        time.sleep(3)                                    # the server's retry interval


if __name__ == "__main__":
    run()

import { readFile, writeFile } from 'node:fs/promises';

const API = 'https://api.routeur.app';
const TOPICS = 'trades,leads';
const CURSOR_FILE = 'routeur-cursor.txt';
const SILENCE_LIMIT_MS = 45_000; // heartbeats arrive every 15 s

const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
const loadCursor = () => readFile(CURSOR_FILE, 'utf8').then((s) => s.trim() || null).catch(() => null);
const saveCursor = (c) => writeFile(CURSOR_FILE, c);

// Parse an SSE byte stream into { id, event, data } objects; comment lines are heartbeats.
async function* events(body, onActivity) {
  let buffer = '';
  let id, event, data = [];
  for await (const chunk of body.pipeThrough(new TextDecoderStream())) {
    onActivity();
    buffer += chunk;
    let nl;
    while ((nl = buffer.indexOf('\n')) >= 0) {
      const line = buffer.slice(0, nl).replace(/\r$/, '');
      buffer = buffer.slice(nl + 1);
      if (line === '') {
        if (event || data.length) yield { id, event, data: data.join('\n') };
        id = event = undefined; data = [];
      } else if (line.startsWith(':')) {
        yield { event: 'heartbeat', data: '' };
      } 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).replace(/^ /, ''));
    }
  }
}

function handle(event, p) {
  if (event === 'trade') {
    console.log(`trade  ${p.venue} ${p.market_id} ${p.taker_action} ${p.outcome} ${Math.round(p.quantity)} @ ${Math.round(p.price_usd * 100)}¢ = $${p.notional_usd}`);
  } else if (event === 'lead') {
    const legs = p.legs.map((l) => `${l.side} ${l.venue} ${l.market_id} @ ${l.ask_usd}`).join(' + ');
    console.log(`lead   run ${p.graph_run} ${p.relation} net ${p.net_edge_usd}: ${legs}`);
  }
}

async function run() {
  let backoff = 3_000;
  for (;;) {
    const cursor = await loadCursor();
    const headers = { 'X-API-Key': process.env.ROUTEUR_API_KEY, Accept: 'text/event-stream' };
    if (cursor) headers['Last-Event-ID'] = cursor;
    const controller = new AbortController();
    let silence = setTimeout(() => controller.abort('silent'), SILENCE_LIMIT_MS);
    const touch = () => { clearTimeout(silence); silence = setTimeout(() => controller.abort('silent'), SILENCE_LIMIT_MS); };
    try {
      const res = await fetch(`${API}/v1/stream?topics=${TOPICS}`, { headers, signal: controller.signal });
      if (res.status === 429 || res.status === 503) {
        const wait = Number(res.headers.get('retry-after') ?? backoff / 1000) * 1000;
        console.log(`${res.status} ${(await res.json()).error.code}: waiting ${wait / 1000}s`);
        await sleep(wait);
        continue;
      }
      if (!res.ok) throw new Error(`stream: ${res.status}`);
      backoff = 3_000;
      for await (const { id, event, data } of events(res.body, touch)) {
        if (event === 'heartbeat') continue;
        const payload = data ? JSON.parse(data) : {};
        if (event === 'ready') {
          console.log(`ready: topics ${payload.topics}, resumed from ${payload.resumed_from ?? 'now'}, stream ends in ${payload.reconnect_after_seconds}s`);
        } else if (event === 'reset') {
          console.log(`reset: ${payload.message}`); // resync from REST here, then carry on
        } else if (event === 'reconnect') {
          console.log(`reconnect (${payload.reason}): resuming from ${payload.cursor}`);
          await saveCursor(payload.cursor);
          break;
        } else if (event === 'error') {
          console.log(`error ${payload.code}: ${payload.message}`);
          return;
        } else {
          handle(event, payload);
        }
        if (id) await saveCursor(id);
      }
    } catch (err) {
      console.log(`connection lost (${err.name ?? err}): retrying in ${backoff / 1000}s`);
      await sleep(backoff);
      backoff = Math.min(backoff * 2, 60_000);
      continue;
    } finally {
      clearTimeout(silence);
    }
    await sleep(3_000); // the server's retry interval
  }
}

run();