> ## Documentation Index > Fetch the complete documentation index at: https://docs.routeur.app/llms.txt > Use this file to discover all available pages before exploring further. > ## Agent Instructions > Prices are US dollars per contract that pays $1: 0.46 means 46 cents. Spread is the Yes ask plus the No ask minus $1. Liquidity grades are deep, tradeable and thin; lists hide thin markets unless include_thin=true. Relations are proven from contract terms, never inferred from wording; a lead marked informational is never executable. Every figure carries a graph_run; read /agents/reading-order first, then /concepts/relations and /concepts/liquidity. # A resumable stream consumer > Consume trades and leads from /v1/stream with Last-Event-ID resume, heartbeat detection, the reconnect event, and backoff, in Python and Node. ## What you'll get A consumer for `GET /v1/stream` that survives everything the stream does on purpose and by accident: it saves the last event id and resumes from it, treats 45 seconds of silence as a dead connection, reopens when the server sends `reconnect`, backs off on errors, and waits out `429` and `503`. Stopped and restarted, it picks up where it left off with nothing missed or repeated. On 2026-09-17 the `trades` topic delivered 6,575 fills in 12 seconds. ## Prerequisites * An API key in `ROUTEUR_API_KEY` * Python 3 with `requests`, or Node 18+ (native `fetch` and `TextDecoderStream`) * A server to run it on. A browser `EventSource` cannot send the key header. ```bash theme={null} curl -N "https://api.routeur.app/v1/stream?topics=trades,leads" -H "X-API-Key: $ROUTEUR_API_KEY" ``` The first lines observed 2026-09-17 12:41 UTC: ```text theme={null} retry: 3000 event: ready data: {"heartbeat_seconds":15,"reconnect_after_seconds":3284,"replay_window_seconds":3600,"topics":["trades","leads"]} id: 1532226 event: trade data: {"venue":"kalshi","id":"072212e8-5c09-abaf-b81b-4971f7c28b79","market_id":"KXCS2GAME-26SEP170700GLM80-M80","outcome":"yes","taker_action":"sell","price_usd":"0.4400","quantity":"44.00","notional_usd":"19.36","executed_at":"2026-09-17T12:41:06.077649Z","block":false,"recorded_at":"2026-09-17T12:41:21.946445Z"} id: 1532227 event: trade data: {"venue":"kalshi","id":"072212e8-ada9-a038-996b-3f806ca68c99","market_id":"KXCS2GAME-26SEP170700GLM80-M80","outcome":"yes","taker_action":"sell","price_usd":"0.4400","quantity":"150.00","notional_usd":"66.00","executed_at":"2026-09-17T12:41:06.077649Z","block":false,"recorded_at":"2026-09-17T12:41:21.946445Z"} : heartbeat ``` Reconnecting with the last id replays what was missed, then continues live. `ready` then names the cursor it resumed from: ```bash theme={null} curl -N "https://api.routeur.app/v1/stream?topics=trades,leads" -H "X-API-Key: $ROUTEUR_API_KEY" -H "Last-Event-ID: 1538923" ``` ```text theme={null} event: ready data: {"heartbeat_seconds":15,"reconnect_after_seconds":2843,"replay_window_seconds":3600,"resumed_from":"1538923","topics":["trades","leads"]} ``` A cursor older than the replay window, or unknown, gets `reset` and the stream continues from now: ```text theme={null} event: reset data: {"message":"The cursor is older than the replay window or unknown; events since it may be missing. Resync from the REST endpoints, then continue from here.","reason":"cursor_expired"} ``` Both versions do the same things: parse SSE by hand (`id:`, `event:`, `data:`, and `:` comment lines as heartbeats), write every event id to a file so a restart resumes, cut the connection after 45 seconds without bytes, reopen on `reconnect` with the cursor it carries, sleep out `Retry-After` on `429` and `503`, and double the wait on connection errors up to a minute. ```python Python theme={null} 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() ``` ```javascript Node theme={null} 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(); ``` Run it, stop it after a few seconds, and run it again. The second `ready` names the cursor from the file, and the first events are the ones recorded while it was down. The Python consumer, run for 12 seconds and then for 8, observed 2026-09-17 13:11 UTC. The first run caught a trade batch and the 32 leads of graph run 75, 8,314 events in all; the second resumed from the last id and found nothing new in its 8 seconds: ```text theme={null} $ python3 -u consumer.py ready: topics ['trades', 'leads'], resumed from now, stream ends in 2916s trade polymarket will-mrbeast-gamings-next-video-get-between-45-and-50-million-views-in-week-1-20260910 buy Yes 55 @ 59¢ = $32.44 trade polymarket btc-updown-15m-1789650000 buy Down 62 @ 37¢ = $23.00 ... lead run 75 superset net -0.01: yes kalshi KXCOPPERW-26SEP1817-T6.60 @ 0.36 + no kalshi KXCOPPERW-26SEP1817-T6.63 @ 0.61 ^C $ cat routeur-cursor.txt 1886375-32 $ python3 -u consumer.py ready: topics ['trades', 'leads'], resumed from 1886375-32, stream ends in 3064s ``` The Node consumer, run for 90 seconds and then for 10, observed 2026-09-17 13:14 UTC. The second run replayed the 7,600 fills recorded during the three seconds it was down and the batch that landed while it ran: ```text theme={null} $ node consumer.mjs ready: topics trades,leads, resumed from now, stream ends in 2891s trade kalshi KXWTACHALLENGERMATCH-26SEP17NOHRAP-NOH buy no 401 @ 32¢ = $128.17 trade kalshi KXBTC15M-26SEP170930-30 sell yes 39 @ 50¢ = $19.32 ... (4,934 trades in 90 s) ^C $ cat routeur-cursor.txt 1929443 $ node consumer.mjs ready: topics trades,leads, resumed from 1929443, stream ends in 2756s trade polymarket eth-updown-5m-1789650600 buy Up 5 @ 56¢ = $2.80 ... (7,600 trades in 10 s) ``` When the key is out of stream slots, the consumer waits rather than crashing: ```text theme={null} 429 too_many_streams: waiting 60s ``` ## Read the result | Field | Meaning | | -------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `id` | The event's position in the stream: an opaque, ordered string such as `1532226` or `90467-0`. Send the last one back as `Last-Event-ID` (or `cursor`) to resume after it. Only data events carry one. | | `ready.replay_window_seconds` | How old a cursor may be and still resume: 3,600 seconds. Older or unknown cursors get `reset`. | | `ready.reconnect_after_seconds` | When this stream will end itself with `reconnect`. Between 45 and 55 minutes, jittered. | | `ready.resumed_from` | Present when the stream replayed after a cursor. Absent means it started from now. | | `: heartbeat` | A comment line every 15 seconds. It has no id and is not an event; it only proves the connection is alive. | | `reconnect.reason`, `reconnect.cursor` | Why the server is closing (`lifetime`, `shutdown`, `behind`, `unavailable`) and where to resume from. Reopen after the `retry` interval, 3 seconds. | | `reset.reason` | `cursor_expired`: events since the cursor may be missing. Resync from the REST endpoints, then continue. | | `error.code` | `daily_quota_exceeded`: the key's quota is used up. The stream ends and will not reopen until the quota resets at midnight UTC. | | `trade.recorded_at` | When the fill was recorded, as opposed to `executed_at`. Recording runs every minute, so fills arrive in bursts, and a backlog can deliver older ones. | | `lead.net_edge_usd`, `lead.depth` | The lead as `/v1/opportunities` reports it. Leads arrive once per graph run, hourly, up to 1,000 per run. | ## Pitfalls * **Stream slots outlive the client.** A key holds one stream per 60 requests a minute of its limit (10 on the pro tier). On 2026-09-17, ten streams closed from the client side (with `curl --max-time`, or a killed process) between 12:41 and 12:53 UTC were all still counted at 13:09, every new connection got `429 too_many_streams` with `Retry-After: 60`, and the first slot came back at 13:11, about 30 minutes after the stream that held it was abandoned. Do not open a fresh stream per retry; keep one consumer, let it reconnect, and expect a wait after a crash loop. * **Fills come in bursts, with quiet in between.** Trades are recorded in batches: 8,741 fills arrived in eight seconds at 13:14:14 UTC on 2026-09-17, and nothing for the two and a half minutes before. A consumer that sees only `ready` for a minute is not broken. Write the cursor after each event and keep `handle` fast, or process in batches off a queue. * **Same filters on resume.** Resuming with different `topics`, `venue`, `markets` or `min_usd` from the ones the cursor came from replays the wrong set. Store the filters with the cursor. * **`cursor` wins over `Last-Event-ID`** when both are sent, and a cursor that is not `^[0-9]+(-[0-9]+)?$` is a `400 invalid_parameter`. Never do arithmetic on ids. * **Python buffers stdout when piped.** Run with `python3 -u`, or nothing shows until the buffer fills. This is not the stream being silent. * **Booleans and lists in the query string are plain text**: `topics=trades,leads`, not JSON. * **Silence is not the same as no events.** Heartbeats prove the connection; a stream with no trades for a minute is normal on a quiet venue. Only the 45-second timeout should reconnect. ## Related * [Rule-change watch](/cookbook/rule-change-watch): the same consumer on the `rules` topic. * [Watch for unusual flow](/cookbook/unusual-flow): the five-minute scored version of the trade feed. * [Rate limits](/rate-limits): how streams count against the minute and the day. * [Webhooks](/webhooks/overview): for executable leads only, delivered to you instead of pulled. * `GET /v1/stream` in the API reference: every topic and event schema.