WebSocket quickstart

// Browser client for tradesByCoin BTC, tradesByCoin ETH — generated by the Coinversa developer portal.
//
// A browser WebSocket cannot set headers, so the key goes in the subprotocol
// list: the literal 'bearer' first, the key second. It is still a request
// header (Sec-WebSocket-Protocol), never part of the URL. A key shipped to
// other people's browsers is a key you have published — use this for your own
// dashboards and tools; for anything served to others, hold the socket on a
// server (see the Node, Python and Go clients).
const URL = "wss://ws.coinversa.ai";
const apiKey = 'cvsa_...'; // paste your key here; never commit it

const SUBSCRIPTIONS = [
  { type: "tradesByCoin", coin: "BTC" },
  { type: "tradesByCoin", coin: "ETH" },
];

// Reconnect with backoff: 1 s doubling to 30 s, plus jitter so a fleet of
// clients does not reconnect in lockstep. Reset once a connection is live.
const BACKOFF = { baseMs: 1000, maxMs: 30000, jitterMs: 500 };
const PING_MS = 30000;
let attempt = 0;
let stopped = false;

// seq is the SOURCE BLOCK HEIGHT, the same on every feed. A filtered feed
// skips numbers whenever a block matches nothing, so a jump is normal and
// only a repeat or a rewind is a fault. A data frame carries no subscription
// id, so key the check on the stream you can identify from the frame itself:
// channel plus the coin or wallet of its first fill.
const streamKey = (msg) => {
  const [, fill] = msg.data?.[0] || [];
  if (msg.channel === 'tradesByCoin' && fill?.coin) return `tradesByCoin:${fill.coin}`;
  if (msg.channel === 'tradesByUser' && msg.data?.[0]?.[0]) return `tradesByUser:${msg.data[0][0]}`;
  return msg.channel;
};

function connect() {
  if (stopped) return;
  const lastSeq = new Map(); // per connection; seq is a block height, so it only moves forward
  const ws = new WebSocket(URL, ['bearer', apiKey]);
  let heartbeat = null;

  ws.addEventListener('open', () => {
    // The socket is open but not yet proven: the 'connected' frame is the
    // server's confirmation. Subscribe now; the server queues nothing before.
    for (const subscription of SUBSCRIPTIONS) {
      ws.send(JSON.stringify({ method: 'subscribe', subscription }));
    }
    heartbeat = setInterval(() => {
      if (ws.readyState === WebSocket.OPEN) ws.send(JSON.stringify({ method: 'ping' }));
    }, PING_MS);
  });

  ws.addEventListener('message', (event) => {
    const msg = JSON.parse(event.data);
    if (msg.channel === 'connected') {
      attempt = 0; // live: the next disconnect starts backoff from the beginning
      const { tier, limits } = msg.data;
      console.log(`connected on ${tier}: ${limits.connections} connections, ${limits.subscriptions} subscriptions`);
      return;
    }
    if (msg.channel === 'subscriptionResponse') { console.log('ack', msg.data.method, msg.data.id); return; }
    if (msg.channel === 'error') { console.error('server error:', msg.data); return; }
    if (msg.channel === 'pong') return;

    const key = streamKey(msg);
    const previous = lastSeq.get(key);
    // seq must INCREASE, not increase by one: a filtered feed skips the
    // blocks that matched nothing. Going backwards or repeating is the fault.
    if (previous != null && msg.seq <= previous) {
      console.warn(`${key}: sequence went ${msg.seq === previous ? 'nowhere' : 'backwards'} ${previous} -> ${msg.seq} — frames replayed or reordered`);
    }
    lastSeq.set(key, msg.seq);

    for (const [wallet, fill] of msg.data) {
      console.log(wallet, fill.coin, fill.side, fill.sz, '@', fill.px, fill.dir, 'pnl', fill.closedPnl);
    }
  });

  // In a browser a refused handshake arrives as a CLOSE, not a status: 4401
  // invalid or missing key (not retryable — fix the key), 4429 your account
  // is at its connection cap, 4503 entitlement lookup unavailable. Anything
  // else, including 1006, is a drop: reconnect.
  ws.addEventListener('close', (event) => {
    clearInterval(heartbeat);
    if (event.code === 4401) { console.error('refused: invalid or missing key — not retrying'); stopped = true; return; }
    if (event.code === 4429) console.warn('refused: connection cap reached — retrying with backoff');
    else if (event.code === 4503) console.warn('refused: entitlement lookup unavailable — retrying with backoff');
    else console.warn(`closed (${event.code}${event.reason ? ' ' + event.reason : ''}) — reconnecting`);
    const delay = Math.min(BACKOFF.maxMs, BACKOFF.baseMs * 2 ** attempt) + Math.random() * BACKOFF.jitterMs;
    attempt += 1;
    setTimeout(connect, delay);
  });
}

connect();

// Node client for tradesByCoin BTC, tradesByCoin ETH — generated by the Coinversa developer portal.
// npm install ws   (Node's built-in WebSocket cannot set headers; use ws.)
import WebSocket from 'ws';

// The key goes out with the HTTP upgrade, so the socket is authenticated by
// the time it opens. Read it from the environment; never commit it.
const URL = "wss://ws.coinversa.ai";
const apiKey = process.env.COINVERSA_API_KEY;
if (!apiKey) { console.error('set COINVERSA_API_KEY'); process.exit(1); }

const SUBSCRIPTIONS = [
  { type: "tradesByCoin", coin: "BTC" },
  { type: "tradesByCoin", coin: "ETH" },
];

// Reconnect with backoff: 1 s doubling to 30 s, plus jitter so a fleet of
// clients does not reconnect in lockstep. Reset once a connection is live.
const BACKOFF = { baseMs: 1000, maxMs: 30000, jitterMs: 500 };
const PING_MS = 30000;
let attempt = 0;
let stopped = false;

// seq is the SOURCE BLOCK HEIGHT, the same on every feed. A filtered feed
// skips numbers whenever a block matches nothing, so a jump is normal and
// only a repeat or a rewind is a fault. A data frame carries no subscription
// id, so key the check on the stream you can identify from the frame itself:
// channel plus the coin or wallet of its first fill.
const streamKey = (msg) => {
  const [, fill] = msg.data?.[0] || [];
  if (msg.channel === 'tradesByCoin' && fill?.coin) return `tradesByCoin:${fill.coin}`;
  if (msg.channel === 'tradesByUser' && msg.data?.[0]?.[0]) return `tradesByUser:${msg.data[0][0]}`;
  return msg.channel;
};

function scheduleReconnect(why) {
  if (stopped) return;
  const delay = Math.min(BACKOFF.maxMs, BACKOFF.baseMs * 2 ** attempt) + Math.random() * BACKOFF.jitterMs;
  attempt += 1;
  console.warn(`${why} — reconnecting in ${Math.round(delay)} ms`);
  setTimeout(connect, delay);
}

function connect() {
  if (stopped) return;
  const lastSeq = new Map(); // per connection; seq is a block height, so it only moves forward
  const ws = new WebSocket(URL, { headers: { Authorization: `Bearer ${apiKey}` } });
  let heartbeat = null;
  let refused = false;

  // A refused upgrade is an HTTP status, not a frame — no socket ever opens,
  // so it never reaches 'message'. 401 invalid or missing key (not retryable:
  // fix the key), 429 your account is at its connection cap, 503 entitlement
  // lookup unavailable — both retried with backoff.
  ws.on('unexpected-response', (request, response) => {
    refused = true;
    request.destroy();
    if (response.statusCode === 401) {
      console.error('refused: HTTP 401 invalid or missing key — not retrying');
      stopped = true;
      process.exitCode = 1; // a supervisor must see this as a failure, not success
      return;
    }
    scheduleReconnect(`refused: HTTP ${response.statusCode}`);
  });

  ws.on('open', () => {
    for (const subscription of SUBSCRIPTIONS) {
      ws.send(JSON.stringify({ method: 'subscribe', subscription }));
    }
    heartbeat = setInterval(() => {
      if (ws.readyState === WebSocket.OPEN) ws.send(JSON.stringify({ method: 'ping' }));
    }, PING_MS);
  });

  ws.on('message', (raw) => {
    const msg = JSON.parse(raw.toString());
    if (msg.channel === 'connected') {
      attempt = 0; // live: the next disconnect starts backoff from the beginning
      const { tier, limits } = msg.data;
      console.log(`connected on ${tier}: ${limits.connections} connections, ${limits.subscriptions} subscriptions`);
      return;
    }
    if (msg.channel === 'subscriptionResponse') { console.log('ack', msg.data.method, msg.data.id); return; }
    if (msg.channel === 'error') { console.error('server error:', msg.data); return; }
    if (msg.channel === 'pong') return;

    const key = streamKey(msg);
    const previous = lastSeq.get(key);
    // seq must INCREASE, not increase by one: a filtered feed skips the
    // blocks that matched nothing. Going backwards or repeating is the fault.
    if (previous != null && msg.seq <= previous) {
      console.warn(`${key}: sequence went ${msg.seq === previous ? 'nowhere' : 'backwards'} ${previous} -> ${msg.seq} — frames replayed or reordered`);
    }
    lastSeq.set(key, msg.seq);

    for (const [wallet, fill] of msg.data) {
      console.log(wallet, fill.coin, fill.side, fill.sz, '@', fill.px, fill.dir, 'pnl', fill.closedPnl);
    }
  });

  ws.on('error', (err) => { if (!refused) console.error('socket error:', err.message); });
  ws.on('close', (code, reason) => {
    clearInterval(heartbeat);
    if (refused) return; // unexpected-response already decided
    scheduleReconnect(`closed (${code}${reason?.length ? ' ' + reason.toString() : ''})`);
  });
}

connect();

# Python client for tradesByCoin BTC, tradesByCoin ETH — generated by the Coinversa developer portal.
# pip install websockets   (15.x; on websockets < 14 the header kwarg is
# extra_headers= and the exception below is InvalidStatusCode with .status_code)
import asyncio
import json
import os
import random

import websockets

URL = "wss://ws.coinversa.ai"
# The key goes out with the HTTP upgrade, so the socket is authenticated by
# the time it opens. Read it from the environment; never commit it.
API_KEY = os.environ["COINVERSA_API_KEY"]

SUBSCRIPTIONS = [
    {"type": "tradesByCoin", "coin": "BTC"},
    {"type": "tradesByCoin", "coin": "ETH"},
]

# Reconnect with backoff: 1 s doubling to 30 s, plus jitter so a fleet of
# clients does not reconnect in lockstep. Reset once a connection is live.
BACKOFF_BASE = 1.0
BACKOFF_MAX = 30.0
BACKOFF_JITTER = 0.5
PING_SECONDS = 30.0


def stream_key(msg):
    # seq is the SOURCE BLOCK HEIGHT, the same on every feed. A filtered feed
    # skips numbers whenever a block matches nothing, so a jump is normal and
    # only a repeat or a rewind is a fault. A data frame carries no
    # subscription id, so key the check on the stream the frame itself
    # identifies: channel plus the coin or wallet of its first fill.
    entry = (msg.get("data") or [[None, {}]])[0]
    wallet, fill = entry if isinstance(entry, list) and len(entry) == 2 else (None, entry if isinstance(entry, dict) else {})
    if msg["channel"] == "tradesByCoin" and fill.get("coin"):
        return "tradesByCoin:" + fill["coin"]
    if msg["channel"] == "tradesByUser" and wallet:
        return "tradesByUser:" + wallet
    return msg["channel"]


async def heartbeat(ws):
    while True:
        await asyncio.sleep(PING_SECONDS)
        await ws.send(json.dumps({"method": "ping"}))


async def session(attempt_reset):
    # One connection: subscribe, then read until it drops. Returns why it ended.
    headers = {"Authorization": "Bearer " + API_KEY}
    async with websockets.connect(URL, additional_headers=headers) as ws:
        for subscription in SUBSCRIPTIONS:
            await ws.send(json.dumps({"method": "subscribe", "subscription": subscription}))
        pinger = asyncio.create_task(heartbeat(ws))
        last_seq = {}  # per connection; seq is a block height, so it only moves forward
        try:
            async for raw in ws:
                msg = json.loads(raw)
                channel = msg.get("channel")
                if channel == "connected":
                    attempt_reset()  # live: the next disconnect starts backoff from the beginning
                    limits = msg["data"]["limits"]
                    print(f'connected on {msg["data"]["tier"]}:',
                          f'{limits["connections"]} connections,',
                          f'{limits["subscriptions"]} subscriptions')
                    continue
                if channel == "subscriptionResponse":
                    print("ack", msg["data"]["method"], msg["data"]["id"])
                    continue
                if channel == "error":
                    print("server error:", msg["data"])
                    continue
                if channel == "pong":
                    continue

                key = stream_key(msg)
                seq = msg["seq"]
                previous = last_seq.get(key)
                # seq must INCREASE, not increase by one: a filtered feed
                # skips the blocks that matched nothing.
                if previous is not None and seq <= previous:
                    went = "nowhere" if seq == previous else "backwards"
                    print(f"{key}: sequence went {went} {previous} -> {seq} - frames replayed or reordered")
                last_seq[key] = seq

                # .get(): a fill that lacks a field must not take the client down.
                for wallet, fill in msg["data"]:
                    print(wallet, fill.get("coin"), fill.get("side"), fill.get("sz"), "@", fill.get("px"),
                          fill.get("dir"), "pnl", fill.get("closedPnl"))
        finally:
            pinger.cancel()
        reason = f" {ws.close_reason}" if ws.close_reason else ""
        return f"closed ({ws.close_code}{reason})"


async def main():
    attempt = 0

    def reset():
        nonlocal attempt
        attempt = 0

    while True:
        try:
            why = await session(reset)
        except websockets.exceptions.InvalidStatus as exc:
            # A refused upgrade is an HTTP status, not a frame - no socket ever
            # opens. 401 invalid or missing key (not retryable: fix the key),
            # 429 account at its connection cap, 503 entitlements down.
            status = exc.response.status_code
            if status == 401:
                print("refused: HTTP 401 invalid or missing key - not retrying")
                raise SystemExit(1)  # a supervisor must see this as a failure, not success
            why = f"refused: HTTP {status}"
        except websockets.exceptions.ConnectionClosed as exc:
            why = f"closed ({exc.rcvd.code if exc.rcvd else 'no close frame'})"
        except OSError as exc:
            why = f"network error: {exc}"

        delay = min(BACKOFF_MAX, BACKOFF_BASE * (2 ** attempt)) + random.random() * BACKOFF_JITTER
        attempt += 1
        print(f"{why} - reconnecting in {delay:.1f}s")
        await asyncio.sleep(delay)


asyncio.run(main())

// Go client for tradesByCoin BTC, tradesByCoin ETH — generated by the Coinversa developer portal.
// go get github.com/coder/websocket
package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"math"
	"math/rand"
	"net/http"
	"os"
	"time"

	"github.com/coder/websocket"
)

const wsURL = "wss://ws.coinversa.ai"

// Reconnect with backoff: 1 s doubling to 30 s, plus jitter so a fleet of
// clients does not reconnect in lockstep. Reset once a connection is live.
const (
	backoffBase   = 1000 * time.Millisecond
	backoffMax    = 30000 * time.Millisecond
	backoffJitter = 500 * time.Millisecond
	pingEvery     = 30000 * time.Millisecond
)

var subscriptions = []map[string]string{
	{"type": "tradesByCoin", "coin": "BTC"},
	{"type": "tradesByCoin", "coin": "ETH"},
}

type frame struct {
	Channel string          `json:"channel"`
	Seq     int64           `json:"seq"`
	Data    json.RawMessage `json:"data"`
}

type fill struct {
	Coin      string `json:"coin"`
	Side      string `json:"side"`
	Px        string `json:"px"`
	Sz        string `json:"sz"`
	Dir       string `json:"dir"`
	ClosedPnl string `json:"closedPnl"`
}

// streamKey: seq is the SOURCE BLOCK HEIGHT, the same on every feed, so a
// forward jump is a block that matched nothing rather than a lost frame. A
// data frame carries no subscription id, so the check is keyed on what the
// frame itself identifies: channel plus the coin or wallet of its first fill.
func streamKey(f frame, pairs [][2]json.RawMessage) string {
	if len(pairs) == 0 {
		return f.Channel
	}
	var wallet string
	var first fill
	_ = json.Unmarshal(pairs[0][0], &wallet)
	_ = json.Unmarshal(pairs[0][1], &first)
	switch {
	case f.Channel == "tradesByCoin" && first.Coin != "":
		return "tradesByCoin:" + first.Coin
	case f.Channel == "tradesByUser" && wallet != "":
		return "tradesByUser:" + wallet
	}
	return f.Channel
}

// session runs one connection until it drops. It returns errNotRetryable for
// a refusal that retrying cannot fix.
var errNotRetryable = errors.New("not retryable")

func session(ctx context.Context, apiKey string, live func()) error {
	// The key goes out with the HTTP upgrade, so the socket is authenticated
	// by the time it opens. A refused upgrade is an HTTP status, not a frame:
	// 401 invalid or missing key, 429 account at its connection cap, 503
	// entitlement lookup unavailable.
	header := http.Header{}
	header.Set("Authorization", "Bearer "+apiKey)
	conn, resp, err := websocket.Dial(ctx, wsURL, &websocket.DialOptions{HTTPHeader: header})
	if err != nil {
		if resp != nil {
			if resp.StatusCode == http.StatusUnauthorized {
				return fmt.Errorf("refused: HTTP 401 invalid or missing key — not retrying: %w", errNotRetryable)
			}
			return fmt.Errorf("refused: HTTP %d", resp.StatusCode)
		}
		return err
	}
	defer conn.CloseNow()
	conn.SetReadLimit(1 << 20)

	for _, sub := range subscriptions {
		body, _ := json.Marshal(map[string]any{"method": "subscribe", "subscription": sub})
		if err := conn.Write(ctx, websocket.MessageText, body); err != nil {
			return err
		}
	}

	pingCtx, stopPing := context.WithCancel(ctx)
	defer stopPing()
	go func() {
		t := time.NewTicker(pingEvery)
		defer t.Stop()
		for {
			select {
			case <-pingCtx.Done():
				return
			case <-t.C:
				_ = conn.Write(pingCtx, websocket.MessageText, []byte(`{"method":"ping"}`))
			}
		}
	}()

	lastSeq := map[string]int64{} // per connection; seq is a block height, so it only moves forward
	for {
		_, raw, err := conn.Read(ctx)
		if err != nil {
			return err
		}
		var f frame
		if err := json.Unmarshal(raw, &f); err != nil {
			log.Printf("unparsable frame: %v", err)
			continue
		}
		switch f.Channel {
		case "connected":
			live() // the next disconnect starts backoff from the beginning
			var d struct {
				Tier   string `json:"tier"`
				Limits struct {
					Connections   int `json:"connections"`
					Subscriptions int `json:"subscriptions"`
				} `json:"limits"`
			}
			_ = json.Unmarshal(f.Data, &d)
			log.Printf("connected on %s: %d connections, %d subscriptions", d.Tier, d.Limits.Connections, d.Limits.Subscriptions)
			continue
		case "subscriptionResponse":
			var ack struct {
				Method string `json:"method"`
				ID     string `json:"id"`
			}
			_ = json.Unmarshal(f.Data, &ack)
			log.Printf("ack %s %s", ack.Method, ack.ID)
			continue
		case "error":
			log.Printf("server error: %s", string(f.Data))
			continue
		case "pong":
			continue
		}

		var pairs [][2]json.RawMessage
		if err := json.Unmarshal(f.Data, &pairs); err != nil {
			continue
		}
		key := streamKey(f, pairs)
		// seq must INCREASE, not increase by one: a filtered feed skips the
		// blocks that matched nothing. Backwards or repeated is the fault.
		if previous, ok := lastSeq[key]; ok && f.Seq <= previous {
			went := "backwards"
			if f.Seq == previous {
				went = "nowhere"
			}
			log.Printf("%s: sequence went %s %d -> %d — frames replayed or reordered", key, went, previous, f.Seq)
		}
		lastSeq[key] = f.Seq

		for _, pair := range pairs {
			var wallet string
			var fl fill
			_ = json.Unmarshal(pair[0], &wallet)
			_ = json.Unmarshal(pair[1], &fl)
			log.Printf("%s %s %s %s @ %s %s pnl %s", wallet, fl.Coin, fl.Side, fl.Sz, fl.Px, fl.Dir, fl.ClosedPnl)
		}
	}
}

func main() {
	apiKey := os.Getenv("COINVERSA_API_KEY")
	if apiKey == "" {
		log.Fatal("set COINVERSA_API_KEY")
	}
	ctx := context.Background()
	attempt := 0
	for {
		err := session(ctx, apiKey, func() { attempt = 0 })
		if errors.Is(err, errNotRetryable) {
			log.Fatal(err) // exits 1: a supervisor must see this as a failure, not success
		}
		if code := websocket.CloseStatus(err); code != -1 {
			var ce websocket.CloseError
			_ = errors.As(err, &ce)
			err = fmt.Errorf("closed (%d %s)", code, ce.Reason)
		}
		delay := time.Duration(math.Min(float64(backoffMax), float64(backoffBase)*math.Pow(2, float64(attempt)))) +
			time.Duration(rand.Int63n(int64(backoffJitter)))
		attempt++
		log.Printf("%v — reconnecting in %s", err, delay.Round(time.Millisecond))
		time.Sleep(delay)
	}
}

// Browser client for l4Book ZEC — generated by the Coinversa developer portal.
//
// A browser WebSocket cannot set headers, so the key goes in the subprotocol
// list: the literal 'bearer' first, the key second. It is still a request
// header (Sec-WebSocket-Protocol), never part of the URL. A key shipped to
// other people's browsers is a key you have published — use this for your own
// dashboards and tools; for anything served to others, hold the socket on a
// server (see the Node, Python and Go clients).
const URL = "wss://ws.coinversa.ai";
const apiKey = 'cvsa_...'; // paste your key here; never commit it

const SUBSCRIPTIONS = [
  { type: "l4Book", coin: "ZEC" },
];

// Reconnect with backoff: 1 s doubling to 30 s, plus jitter so a fleet of
// clients does not reconnect in lockstep. Reset once a connection is live.
const BACKOFF = { baseMs: 1000, maxMs: 30000, jitterMs: 500 };
const PING_MS = 30000;
let attempt = 0;
let stopped = false;

// seq is the SOURCE BLOCK HEIGHT, the same on every feed. A filtered feed
// skips numbers whenever a block matches nothing, so a jump is normal and
// only a repeat or a rewind is a fault. A data frame carries no subscription
// id, so key the check on the stream you can identify from the frame itself:
// channel plus the coin or wallet of its first fill.
const streamKey = (msg) => {
  const [, fill] = msg.data?.[0] || [];
  if (msg.channel === 'tradesByCoin' && fill?.coin) return `tradesByCoin:${fill.coin}`;
  if (msg.channel === 'tradesByUser' && msg.data?.[0]?.[0]) return `tradesByUser:${msg.data[0][0]}`;
  return msg.channel;
};

// Order book (l4Book). The server holds no book: the first message on the
// subscription is the whole book as ONE frame — about 30 MB for BTC (HYPE ~15
// MB, ETH ~11 MB, the median coin ~5 KB) — and every block after it is a
// frame of diffs. A browser WebSocket has no message-size limit, so there is
// nothing to configure, but expect JSON.parse to take a moment on the snapshot.
//
// The client rule, from the protocol reference: "Apply an update only if its prev equals the last applied height; on any mismatch, or partial:true when attribution matters, reconnect with the last successfully applied seq. If that cursor is no longer retained, subscribe without it to rebuild from a fresh snapshot."
//
// This client takes the rule's FALLBACK path: it resubscribes without a
// cursor, which costs a fresh snapshot (~30 MB on BTC, 10-70 s). Passing
// {"sequence": <last applied seq>} on the resubscribe resumes instead, with
// no snapshot — add it if that cost matters, and fall back to this path when
// the server answers "sequence is not retained".
//
// A book subscription that fails after it starts — unknown coin, the gateway
// unreachable, the snapshot buffer never filling — is dropped and reported on
// the ordinary error channel as {"channel":"error","data":"<reason>"}. That
// frame does NOT name the coin, so with more than one l4Book subscription on
// a socket you cannot tell which one died from the frame alone; treat an
// error frame as a reason to re-check every book you hold.
const bookHeight = new Map(); // coin -> last applied height

function resubscribeBook(ws, coin, why) {
  console.warn(`l4Book ${coin}: ${why} — resubscribing for a fresh snapshot`);
  bookHeight.delete(coin);
  const subscription = { type: 'l4Book', coin };
  ws.send(JSON.stringify({ method: 'unsubscribe', subscription }));
  ws.send(JSON.stringify({ method: 'subscribe', subscription }));
}

function onBookFrame(ws, msg) {
  if (msg.snapshot) {
    const [bids, asks] = msg.data.book_orders;
    bookHeight.set(msg.coin, msg.height);
    // Build your book from bids, asks and msg.data.untriggered_orders here:
    // every entry is a [wallet, order] pair. The book is the state after
    // block msg.height.
    console.log(`l4Book ${msg.coin}: snapshot at height ${msg.height} — ${bids.length} bids, ${asks.length} asks, ${msg.data.untriggered_orders.length} triggers`);
    return;
  }
  const last = bookHeight.get(msg.coin);
  if (last == null) return; // nothing to apply it to: no snapshot yet, or the subscription ended
  if (msg.prev !== last) { resubscribeBook(ws, msg.coin, `prev ${msg.prev} is not the last applied height ${last}`); return; }
  if (msg.partial) { resubscribeBook(ws, msg.coin, `block ${msg.height} is partial`); return; }
  // Apply msg.data.raw_book_diffs, then msg.data.order_statuses, to your book here.
  bookHeight.set(msg.coin, msg.height);
  console.log(`l4Book ${msg.coin}: height ${msg.height} — ${msg.data.raw_book_diffs.length} diffs, ${msg.data.order_statuses.length} statuses`);
}

function connect() {
  if (stopped) return;
  const lastSeq = new Map(); // per connection; seq is a block height, so it only moves forward
  bookHeight.clear(); // fresh per connection: a new socket starts with a new snapshot
  const ws = new WebSocket(URL, ['bearer', apiKey]);
  let heartbeat = null;

  ws.addEventListener('open', () => {
    // The socket is open but not yet proven: the 'connected' frame is the
    // server's confirmation. Subscribe now; the server queues nothing before.
    for (const subscription of SUBSCRIPTIONS) {
      ws.send(JSON.stringify({ method: 'subscribe', subscription }));
    }
    heartbeat = setInterval(() => {
      if (ws.readyState === WebSocket.OPEN) ws.send(JSON.stringify({ method: 'ping' }));
    }, PING_MS);
  });

  ws.addEventListener('message', (event) => {
    const msg = JSON.parse(event.data);
    if (msg.channel === 'connected') {
      attempt = 0; // live: the next disconnect starts backoff from the beginning
      const { tier, limits } = msg.data;
      console.log(`connected on ${tier}: ${limits.connections} connections, ${limits.subscriptions} subscriptions`);
      return;
    }
    if (msg.channel === 'subscriptionResponse') { console.log('ack', msg.data.method, msg.data.id); return; }
    if (msg.channel === 'error') { console.error('server error:', msg.data); return; }
    if (msg.channel === 'pong') return;
    if (msg.channel === 'l4Book') { onBookFrame(ws, msg); return; } // gated on prev, not seq

    const key = streamKey(msg);
    const previous = lastSeq.get(key);
    // seq must INCREASE, not increase by one: a filtered feed skips the
    // blocks that matched nothing. Going backwards or repeating is the fault.
    if (previous != null && msg.seq <= previous) {
      console.warn(`${key}: sequence went ${msg.seq === previous ? 'nowhere' : 'backwards'} ${previous} -> ${msg.seq} — frames replayed or reordered`);
    }
    lastSeq.set(key, msg.seq);

    for (const [wallet, fill] of msg.data) {
      console.log(wallet, fill.coin, fill.side, fill.sz, '@', fill.px, fill.dir, 'pnl', fill.closedPnl);
    }
  });

  // In a browser a refused handshake arrives as a CLOSE, not a status: 4401
  // invalid or missing key (not retryable — fix the key), 4429 your account
  // is at its connection cap, 4503 entitlement lookup unavailable. Anything
  // else, including 1006, is a drop: reconnect.
  ws.addEventListener('close', (event) => {
    clearInterval(heartbeat);
    if (event.code === 4401) { console.error('refused: invalid or missing key — not retrying'); stopped = true; return; }
    if (event.code === 4429) console.warn('refused: connection cap reached — retrying with backoff');
    else if (event.code === 4503) console.warn('refused: entitlement lookup unavailable — retrying with backoff');
    else console.warn(`closed (${event.code}${event.reason ? ' ' + event.reason : ''}) — reconnecting`);
    const delay = Math.min(BACKOFF.maxMs, BACKOFF.baseMs * 2 ** attempt) + Math.random() * BACKOFF.jitterMs;
    attempt += 1;
    setTimeout(connect, delay);
  });
}

connect();

// Node client for l4Book ZEC — generated by the Coinversa developer portal.
// npm install ws   (Node's built-in WebSocket cannot set headers; use ws.)
import WebSocket from 'ws';

// The key goes out with the HTTP upgrade, so the socket is authenticated by
// the time it opens. Read it from the environment; never commit it.
const URL = "wss://ws.coinversa.ai";
const apiKey = process.env.COINVERSA_API_KEY;
if (!apiKey) { console.error('set COINVERSA_API_KEY'); process.exit(1); }

const SUBSCRIPTIONS = [
  { type: "l4Book", coin: "ZEC" },
];

// Reconnect with backoff: 1 s doubling to 30 s, plus jitter so a fleet of
// clients does not reconnect in lockstep. Reset once a connection is live.
const BACKOFF = { baseMs: 1000, maxMs: 30000, jitterMs: 500 };
const PING_MS = 30000;
let attempt = 0;
let stopped = false;

// seq is the SOURCE BLOCK HEIGHT, the same on every feed. A filtered feed
// skips numbers whenever a block matches nothing, so a jump is normal and
// only a repeat or a rewind is a fault. A data frame carries no subscription
// id, so key the check on the stream you can identify from the frame itself:
// channel plus the coin or wallet of its first fill.
const streamKey = (msg) => {
  const [, fill] = msg.data?.[0] || [];
  if (msg.channel === 'tradesByCoin' && fill?.coin) return `tradesByCoin:${fill.coin}`;
  if (msg.channel === 'tradesByUser' && msg.data?.[0]?.[0]) return `tradesByUser:${msg.data[0][0]}`;
  return msg.channel;
};

// Order book (l4Book). The server holds no book: the first message on the
// subscription is the whole book as ONE frame — about 30 MB for BTC (HYPE ~15
// MB, ETH ~11 MB, the median coin ~5 KB) — and every block after it is a
// frame of diffs. ws has no message-size limit by default, so there is
// nothing to configure, but expect JSON.parse to take a moment on the snapshot.
//
// The client rule, from the protocol reference: "Apply an update only if its prev equals the last applied height; on any mismatch, or partial:true when attribution matters, reconnect with the last successfully applied seq. If that cursor is no longer retained, subscribe without it to rebuild from a fresh snapshot."
//
// This client takes the rule's FALLBACK path: it resubscribes without a
// cursor, which costs a fresh snapshot (~30 MB on BTC, 10-70 s). Passing
// {"sequence": <last applied seq>} on the resubscribe resumes instead, with
// no snapshot — add it if that cost matters, and fall back to this path when
// the server answers "sequence is not retained".
//
// A book subscription that fails after it starts — unknown coin, the gateway
// unreachable, the snapshot buffer never filling — is dropped and reported on
// the ordinary error channel as {"channel":"error","data":"<reason>"}. That
// frame does NOT name the coin, so with more than one l4Book subscription on
// a socket you cannot tell which one died from the frame alone; treat an
// error frame as a reason to re-check every book you hold.
const bookHeight = new Map(); // coin -> last applied height

function resubscribeBook(ws, coin, why) {
  console.warn(`l4Book ${coin}: ${why} — resubscribing for a fresh snapshot`);
  bookHeight.delete(coin);
  const subscription = { type: 'l4Book', coin };
  ws.send(JSON.stringify({ method: 'unsubscribe', subscription }));
  ws.send(JSON.stringify({ method: 'subscribe', subscription }));
}

function onBookFrame(ws, msg) {
  if (msg.snapshot) {
    const [bids, asks] = msg.data.book_orders;
    bookHeight.set(msg.coin, msg.height);
    // Build your book from bids, asks and msg.data.untriggered_orders here:
    // every entry is a [wallet, order] pair. The book is the state after
    // block msg.height.
    console.log(`l4Book ${msg.coin}: snapshot at height ${msg.height} — ${bids.length} bids, ${asks.length} asks, ${msg.data.untriggered_orders.length} triggers`);
    return;
  }
  const last = bookHeight.get(msg.coin);
  if (last == null) return; // nothing to apply it to: no snapshot yet, or the subscription ended
  if (msg.prev !== last) { resubscribeBook(ws, msg.coin, `prev ${msg.prev} is not the last applied height ${last}`); return; }
  if (msg.partial) { resubscribeBook(ws, msg.coin, `block ${msg.height} is partial`); return; }
  // Apply msg.data.raw_book_diffs, then msg.data.order_statuses, to your book here.
  bookHeight.set(msg.coin, msg.height);
  console.log(`l4Book ${msg.coin}: height ${msg.height} — ${msg.data.raw_book_diffs.length} diffs, ${msg.data.order_statuses.length} statuses`);
}

function scheduleReconnect(why) {
  if (stopped) return;
  const delay = Math.min(BACKOFF.maxMs, BACKOFF.baseMs * 2 ** attempt) + Math.random() * BACKOFF.jitterMs;
  attempt += 1;
  console.warn(`${why} — reconnecting in ${Math.round(delay)} ms`);
  setTimeout(connect, delay);
}

function connect() {
  if (stopped) return;
  const lastSeq = new Map(); // per connection; seq is a block height, so it only moves forward
  bookHeight.clear(); // fresh per connection: a new socket starts with a new snapshot
  const ws = new WebSocket(URL, { headers: { Authorization: `Bearer ${apiKey}` } });
  let heartbeat = null;
  let refused = false;

  // A refused upgrade is an HTTP status, not a frame — no socket ever opens,
  // so it never reaches 'message'. 401 invalid or missing key (not retryable:
  // fix the key), 429 your account is at its connection cap, 503 entitlement
  // lookup unavailable — both retried with backoff.
  ws.on('unexpected-response', (request, response) => {
    refused = true;
    request.destroy();
    if (response.statusCode === 401) {
      console.error('refused: HTTP 401 invalid or missing key — not retrying');
      stopped = true;
      process.exitCode = 1; // a supervisor must see this as a failure, not success
      return;
    }
    scheduleReconnect(`refused: HTTP ${response.statusCode}`);
  });

  ws.on('open', () => {
    for (const subscription of SUBSCRIPTIONS) {
      ws.send(JSON.stringify({ method: 'subscribe', subscription }));
    }
    heartbeat = setInterval(() => {
      if (ws.readyState === WebSocket.OPEN) ws.send(JSON.stringify({ method: 'ping' }));
    }, PING_MS);
  });

  ws.on('message', (raw) => {
    const msg = JSON.parse(raw.toString());
    if (msg.channel === 'connected') {
      attempt = 0; // live: the next disconnect starts backoff from the beginning
      const { tier, limits } = msg.data;
      console.log(`connected on ${tier}: ${limits.connections} connections, ${limits.subscriptions} subscriptions`);
      return;
    }
    if (msg.channel === 'subscriptionResponse') { console.log('ack', msg.data.method, msg.data.id); return; }
    if (msg.channel === 'error') { console.error('server error:', msg.data); return; }
    if (msg.channel === 'pong') return;
    if (msg.channel === 'l4Book') { onBookFrame(ws, msg); return; } // gated on prev, not seq

    const key = streamKey(msg);
    const previous = lastSeq.get(key);
    // seq must INCREASE, not increase by one: a filtered feed skips the
    // blocks that matched nothing. Going backwards or repeating is the fault.
    if (previous != null && msg.seq <= previous) {
      console.warn(`${key}: sequence went ${msg.seq === previous ? 'nowhere' : 'backwards'} ${previous} -> ${msg.seq} — frames replayed or reordered`);
    }
    lastSeq.set(key, msg.seq);

    for (const [wallet, fill] of msg.data) {
      console.log(wallet, fill.coin, fill.side, fill.sz, '@', fill.px, fill.dir, 'pnl', fill.closedPnl);
    }
  });

  ws.on('error', (err) => { if (!refused) console.error('socket error:', err.message); });
  ws.on('close', (code, reason) => {
    clearInterval(heartbeat);
    if (refused) return; // unexpected-response already decided
    scheduleReconnect(`closed (${code}${reason?.length ? ' ' + reason.toString() : ''})`);
  });
}

connect();

# Python client for l4Book ZEC — generated by the Coinversa developer portal.
# pip install websockets   (15.x; on websockets < 14 the header kwarg is
# extra_headers= and the exception below is InvalidStatusCode with .status_code)
import asyncio
import json
import os
import random

import websockets

URL = "wss://ws.coinversa.ai"
# The key goes out with the HTTP upgrade, so the socket is authenticated by
# the time it opens. Read it from the environment; never commit it.
API_KEY = os.environ["COINVERSA_API_KEY"]

SUBSCRIPTIONS = [
    {"type": "l4Book", "coin": "ZEC"},
]

# Reconnect with backoff: 1 s doubling to 30 s, plus jitter so a fleet of
# clients does not reconnect in lockstep. Reset once a connection is live.
BACKOFF_BASE = 1.0
BACKOFF_MAX = 30.0
BACKOFF_JITTER = 0.5
PING_SECONDS = 30.0
# A book snapshot is one message of about 30 MB for BTC; the library's default
# limit is 1 MiB and would close the socket on it.
BOOK_MAX_SIZE = 64 * 1024 * 1024


def stream_key(msg):
    # seq is the SOURCE BLOCK HEIGHT, the same on every feed. A filtered feed
    # skips numbers whenever a block matches nothing, so a jump is normal and
    # only a repeat or a rewind is a fault. A data frame carries no
    # subscription id, so key the check on the stream the frame itself
    # identifies: channel plus the coin or wallet of its first fill.
    entry = (msg.get("data") or [[None, {}]])[0]
    wallet, fill = entry if isinstance(entry, list) and len(entry) == 2 else (None, entry if isinstance(entry, dict) else {})
    if msg["channel"] == "tradesByCoin" and fill.get("coin"):
        return "tradesByCoin:" + fill["coin"]
    if msg["channel"] == "tradesByUser" and wallet:
        return "tradesByUser:" + wallet
    return msg["channel"]


# Order book (l4Book). The server holds no book: the first message on the
# subscription is the whole book as ONE message (about 30 MB for BTC, HYPE
# ~15 MB, ETH ~11 MB, the median coin ~5 KB), then one message of diffs per
# block. BOOK_MAX_SIZE above lets it through.
#
# The client rule, from the protocol reference: "Apply an update only if its prev equals the last applied height; on any mismatch, or partial:true when attribution matters, reconnect with the last successfully applied seq. If that cursor is no longer retained, subscribe without it to rebuild from a fresh snapshot."
#
#
# This client takes the rule's FALLBACK path: it resubscribes without a
# cursor, which costs a fresh snapshot (~30 MB on BTC, 10-70 s). Passing
# {"sequence": <last applied seq>} on the resubscribe resumes instead, with
# no snapshot — add it if that cost matters, and fall back to this path when
# the server answers "sequence is not retained".
#
# A book subscription that fails after it starts - unknown coin, the gateway
# unreachable, the snapshot buffer never filling - is dropped and reported on
# the ordinary error channel as {"channel":"error","data":"<reason>"}. That
# frame does NOT name the coin, so with more than one l4Book subscription on
# a socket you cannot tell which one died from the frame alone; treat an
# error frame as a reason to re-check every book you hold.
BOOK_HEIGHT = {}  # coin -> last applied height


async def resubscribe_book(ws, coin, why):
    print(f"l4Book {coin}: {why} - resubscribing for a fresh snapshot")
    BOOK_HEIGHT.pop(coin, None)
    subscription = {"type": "l4Book", "coin": coin}
    await ws.send(json.dumps({"method": "unsubscribe", "subscription": subscription}))
    await ws.send(json.dumps({"method": "subscribe", "subscription": subscription}))


async def on_book_frame(ws, msg):
    coin = msg.get("coin")
    data = msg.get("data") or {}
    if msg.get("snapshot"):
        bids, asks = data.get("book_orders") or ([], [])
        BOOK_HEIGHT[coin] = msg["height"]
        # Build your book from bids, asks and data["untriggered_orders"] here:
        # every entry is a [wallet, order] pair. The book is the state after
        # block msg["height"].
        print(f'l4Book {coin}: snapshot at height {msg["height"]} -',
              f'{len(bids)} bids, {len(asks)} asks, {len(data.get("untriggered_orders") or [])} triggers')
        return
    last = BOOK_HEIGHT.get(coin)
    if last is None:
        return  # nothing to apply it to: no snapshot yet, or the subscription ended
    if msg.get("prev") != last:
        await resubscribe_book(ws, coin, f'prev {msg.get("prev")} is not the last applied height {last}')
        return
    if msg.get("partial"):
        await resubscribe_book(ws, coin, f'block {msg["height"]} is partial')
        return
    # Apply data["raw_book_diffs"], then data["order_statuses"], to your book here.
    BOOK_HEIGHT[coin] = msg["height"]
    print(f'l4Book {coin}: height {msg["height"]} -',
          f'{len(data.get("raw_book_diffs") or [])} diffs, {len(data.get("order_statuses") or [])} statuses')


async def heartbeat(ws):
    while True:
        await asyncio.sleep(PING_SECONDS)
        await ws.send(json.dumps({"method": "ping"}))


async def session(attempt_reset):
    # One connection: subscribe, then read until it drops. Returns why it ended.
    headers = {"Authorization": "Bearer " + API_KEY}
    async with websockets.connect(URL, additional_headers=headers, max_size=BOOK_MAX_SIZE) as ws:
        for subscription in SUBSCRIPTIONS:
            await ws.send(json.dumps({"method": "subscribe", "subscription": subscription}))
        pinger = asyncio.create_task(heartbeat(ws))
        last_seq = {}  # per connection; seq is a block height, so it only moves forward
        BOOK_HEIGHT.clear()  # fresh per connection: a new socket starts with a new snapshot
        try:
            async for raw in ws:
                msg = json.loads(raw)
                channel = msg.get("channel")
                if channel == "connected":
                    attempt_reset()  # live: the next disconnect starts backoff from the beginning
                    limits = msg["data"]["limits"]
                    print(f'connected on {msg["data"]["tier"]}:',
                          f'{limits["connections"]} connections,',
                          f'{limits["subscriptions"]} subscriptions')
                    continue
                if channel == "subscriptionResponse":
                    print("ack", msg["data"]["method"], msg["data"]["id"])
                    continue
                if channel == "error":
                    print("server error:", msg["data"])
                    continue
                if channel == "pong":
                    continue
                if channel == "l4Book":
                    await on_book_frame(ws, msg)  # gated on prev, not seq
                    continue

                key = stream_key(msg)
                seq = msg["seq"]
                previous = last_seq.get(key)
                # seq must INCREASE, not increase by one: a filtered feed
                # skips the blocks that matched nothing.
                if previous is not None and seq <= previous:
                    went = "nowhere" if seq == previous else "backwards"
                    print(f"{key}: sequence went {went} {previous} -> {seq} - frames replayed or reordered")
                last_seq[key] = seq

                # .get(): a fill that lacks a field must not take the client down.
                for wallet, fill in msg["data"]:
                    print(wallet, fill.get("coin"), fill.get("side"), fill.get("sz"), "@", fill.get("px"),
                          fill.get("dir"), "pnl", fill.get("closedPnl"))
        finally:
            pinger.cancel()
        reason = f" {ws.close_reason}" if ws.close_reason else ""
        return f"closed ({ws.close_code}{reason})"


async def main():
    attempt = 0

    def reset():
        nonlocal attempt
        attempt = 0

    while True:
        try:
            why = await session(reset)
        except websockets.exceptions.InvalidStatus as exc:
            # A refused upgrade is an HTTP status, not a frame - no socket ever
            # opens. 401 invalid or missing key (not retryable: fix the key),
            # 429 account at its connection cap, 503 entitlements down.
            status = exc.response.status_code
            if status == 401:
                print("refused: HTTP 401 invalid or missing key - not retrying")
                raise SystemExit(1)  # a supervisor must see this as a failure, not success
            why = f"refused: HTTP {status}"
        except websockets.exceptions.ConnectionClosed as exc:
            why = f"closed ({exc.rcvd.code if exc.rcvd else 'no close frame'})"
        except OSError as exc:
            why = f"network error: {exc}"

        delay = min(BACKOFF_MAX, BACKOFF_BASE * (2 ** attempt)) + random.random() * BACKOFF_JITTER
        attempt += 1
        print(f"{why} - reconnecting in {delay:.1f}s")
        await asyncio.sleep(delay)


asyncio.run(main())

// Go client for l4Book ZEC — generated by the Coinversa developer portal.
// go get github.com/coder/websocket
package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"math"
	"math/rand"
	"net/http"
	"os"
	"time"

	"github.com/coder/websocket"
)

const wsURL = "wss://ws.coinversa.ai"

// Reconnect with backoff: 1 s doubling to 30 s, plus jitter so a fleet of
// clients does not reconnect in lockstep. Reset once a connection is live.
const (
	backoffBase   = 1000 * time.Millisecond
	backoffMax    = 30000 * time.Millisecond
	backoffJitter = 500 * time.Millisecond
	pingEvery     = 30000 * time.Millisecond
)

var subscriptions = []map[string]string{
	{"type": "l4Book", "coin": "ZEC"},
}

type frame struct {
	Channel string          `json:"channel"`
	Seq     int64           `json:"seq"`
	Data    json.RawMessage `json:"data"`
}

type fill struct {
	Coin      string `json:"coin"`
	Side      string `json:"side"`
	Px        string `json:"px"`
	Sz        string `json:"sz"`
	Dir       string `json:"dir"`
	ClosedPnl string `json:"closedPnl"`
}

// bookFrame is an l4Book message: the snapshot that opens a subscription, an
// update per block after it, or the error that ends it. Orders and events are
// left raw: build your book from them.
type bookFrame struct {
	Coin     string `json:"coin"`
	Snapshot bool   `json:"snapshot"`
	Height   int64  `json:"height"`
	Prev     int64  `json:"prev"`
	Partial  bool   `json:"partial"`
	Data     struct {
		BookOrders        [][]json.RawMessage `json:"book_orders"`
		UntriggeredOrders []json.RawMessage   `json:"untriggered_orders"`
		RawBookDiffs      []json.RawMessage   `json:"raw_book_diffs"`
		OrderStatuses     []json.RawMessage   `json:"order_statuses"`
	} `json:"data"`
}

// streamKey: seq is the SOURCE BLOCK HEIGHT, the same on every feed, so a
// forward jump is a block that matched nothing rather than a lost frame. A
// data frame carries no subscription id, so the check is keyed on what the
// frame itself identifies: channel plus the coin or wallet of its first fill.
func streamKey(f frame, pairs [][2]json.RawMessage) string {
	if len(pairs) == 0 {
		return f.Channel
	}
	var wallet string
	var first fill
	_ = json.Unmarshal(pairs[0][0], &wallet)
	_ = json.Unmarshal(pairs[0][1], &first)
	switch {
	case f.Channel == "tradesByCoin" && first.Coin != "":
		return "tradesByCoin:" + first.Coin
	case f.Channel == "tradesByUser" && wallet != "":
		return "tradesByUser:" + wallet
	}
	return f.Channel
}

// Order book (l4Book). The server holds no book: the first message on the
// subscription is the whole book as ONE message — about 30 MB for BTC (HYPE
// ~15 MB, ETH ~11 MB, the median coin ~5 KB) — then one message of diffs per
// block. The read limit in session lets it through.
//
// The client rule, from the protocol reference: "Apply an update only if its prev equals the last applied height; on any mismatch, or partial:true when attribution matters, reconnect with the last successfully applied seq. If that cursor is no longer retained, subscribe without it to rebuild from a fresh snapshot."
//
// This client takes the rule's FALLBACK path: it resubscribes without a
// cursor, which costs a fresh snapshot (~30 MB on BTC, 10-70 s). Passing
// {"sequence": <last applied seq>} on the resubscribe resumes instead, with
// no snapshot — add it if that cost matters, and fall back to this path when
// the server answers "sequence is not retained".
//
// A book subscription that fails after it starts — unknown coin, the gateway
// unreachable, the snapshot buffer never filling — is dropped and reported on
// the ordinary error channel as {"channel":"error","data":"<reason>"}. That
// frame does NOT name the coin, so with more than one l4Book subscription on
// a socket you cannot tell which one died from the frame alone; treat an
// error frame as a reason to re-check every book you hold.
func resubscribeBook(ctx context.Context, conn *websocket.Conn, height map[string]int64, coin, why string) {
	log.Printf("l4Book %s: %s — resubscribing for a fresh snapshot", coin, why)
	delete(height, coin)
	sub := map[string]string{"type": "l4Book", "coin": coin}
	for _, method := range []string{"unsubscribe", "subscribe"} {
		body, _ := json.Marshal(map[string]any{"method": method, "subscription": sub})
		_ = conn.Write(ctx, websocket.MessageText, body)
	}
}

func onBookFrame(ctx context.Context, conn *websocket.Conn, height map[string]int64, raw []byte) {
	var b bookFrame
	if err := json.Unmarshal(raw, &b); err != nil {
		log.Printf("unparsable l4Book frame: %v", err)
		return
	}
	if b.Snapshot {
		bids, asks := 0, 0
		if len(b.Data.BookOrders) == 2 {
			bids, asks = len(b.Data.BookOrders[0]), len(b.Data.BookOrders[1])
		}
		height[b.Coin] = b.Height
		// Build your book from b.Data.BookOrders[0] (bids), [1] (asks) and
		// b.Data.UntriggeredOrders here: every entry is a [wallet, order]
		// pair. The book is the state after block b.Height.
		log.Printf("l4Book %s: snapshot at height %d — %d bids, %d asks, %d triggers", b.Coin, b.Height, bids, asks, len(b.Data.UntriggeredOrders))
		return
	}
	last, ok := height[b.Coin]
	if !ok {
		return // nothing to apply it to: no snapshot yet, or the subscription ended
	}
	if b.Prev != last {
		resubscribeBook(ctx, conn, height, b.Coin, fmt.Sprintf("prev %d is not the last applied height %d", b.Prev, last))
		return
	}
	if b.Partial {
		resubscribeBook(ctx, conn, height, b.Coin, fmt.Sprintf("block %d is partial", b.Height))
		return
	}
	// Apply b.Data.RawBookDiffs, then b.Data.OrderStatuses, to your book here.
	height[b.Coin] = b.Height
	log.Printf("l4Book %s: height %d — %d diffs, %d statuses", b.Coin, b.Height, len(b.Data.RawBookDiffs), len(b.Data.OrderStatuses))
}

// session runs one connection until it drops. It returns errNotRetryable for
// a refusal that retrying cannot fix.
var errNotRetryable = errors.New("not retryable")

func session(ctx context.Context, apiKey string, live func()) error {
	// The key goes out with the HTTP upgrade, so the socket is authenticated
	// by the time it opens. A refused upgrade is an HTTP status, not a frame:
	// 401 invalid or missing key, 429 account at its connection cap, 503
	// entitlement lookup unavailable.
	header := http.Header{}
	header.Set("Authorization", "Bearer "+apiKey)
	conn, resp, err := websocket.Dial(ctx, wsURL, &websocket.DialOptions{HTTPHeader: header})
	if err != nil {
		if resp != nil {
			if resp.StatusCode == http.StatusUnauthorized {
				return fmt.Errorf("refused: HTTP 401 invalid or missing key — not retrying: %w", errNotRetryable)
			}
			return fmt.Errorf("refused: HTTP %d", resp.StatusCode)
		}
		return err
	}
	defer conn.CloseNow()
	conn.SetReadLimit(64 << 20) // a book snapshot is one message of about 30 MB for BTC; the library default is 32 KiB

	for _, sub := range subscriptions {
		body, _ := json.Marshal(map[string]any{"method": "subscribe", "subscription": sub})
		if err := conn.Write(ctx, websocket.MessageText, body); err != nil {
			return err
		}
	}

	pingCtx, stopPing := context.WithCancel(ctx)
	defer stopPing()
	go func() {
		t := time.NewTicker(pingEvery)
		defer t.Stop()
		for {
			select {
			case <-pingCtx.Done():
				return
			case <-t.C:
				_ = conn.Write(pingCtx, websocket.MessageText, []byte(`{"method":"ping"}`))
			}
		}
	}()

	lastSeq := map[string]int64{} // per connection; seq is a block height, so it only moves forward
	// fresh per connection too: a new socket starts with a new snapshot
	bookHeight := map[string]int64{}
	for {
		_, raw, err := conn.Read(ctx)
		if err != nil {
			return err
		}
		var f frame
		if err := json.Unmarshal(raw, &f); err != nil {
			log.Printf("unparsable frame: %v", err)
			continue
		}
		switch f.Channel {
		case "connected":
			live() // the next disconnect starts backoff from the beginning
			var d struct {
				Tier   string `json:"tier"`
				Limits struct {
					Connections   int `json:"connections"`
					Subscriptions int `json:"subscriptions"`
				} `json:"limits"`
			}
			_ = json.Unmarshal(f.Data, &d)
			log.Printf("connected on %s: %d connections, %d subscriptions", d.Tier, d.Limits.Connections, d.Limits.Subscriptions)
			continue
		case "subscriptionResponse":
			var ack struct {
				Method string `json:"method"`
				ID     string `json:"id"`
			}
			_ = json.Unmarshal(f.Data, &ack)
			log.Printf("ack %s %s", ack.Method, ack.ID)
			continue
		case "error":
			log.Printf("server error: %s", string(f.Data))
			continue
		case "pong":
			continue
		case "l4Book":
			onBookFrame(ctx, conn, bookHeight, raw) // gated on prev, not seq
			continue
		}

		var pairs [][2]json.RawMessage
		if err := json.Unmarshal(f.Data, &pairs); err != nil {
			continue
		}
		key := streamKey(f, pairs)
		// seq must INCREASE, not increase by one: a filtered feed skips the
		// blocks that matched nothing. Backwards or repeated is the fault.
		if previous, ok := lastSeq[key]; ok && f.Seq <= previous {
			went := "backwards"
			if f.Seq == previous {
				went = "nowhere"
			}
			log.Printf("%s: sequence went %s %d -> %d — frames replayed or reordered", key, went, previous, f.Seq)
		}
		lastSeq[key] = f.Seq

		for _, pair := range pairs {
			var wallet string
			var fl fill
			_ = json.Unmarshal(pair[0], &wallet)
			_ = json.Unmarshal(pair[1], &fl)
			log.Printf("%s %s %s %s @ %s %s pnl %s", wallet, fl.Coin, fl.Side, fl.Sz, fl.Px, fl.Dir, fl.ClosedPnl)
		}
	}
}

func main() {
	apiKey := os.Getenv("COINVERSA_API_KEY")
	if apiKey == "" {
		log.Fatal("set COINVERSA_API_KEY")
	}
	ctx := context.Background()
	attempt := 0
	for {
		err := session(ctx, apiKey, func() { attempt = 0 })
		if errors.Is(err, errNotRetryable) {
			log.Fatal(err) // exits 1: a supervisor must see this as a failure, not success
		}
		if code := websocket.CloseStatus(err); code != -1 {
			var ce websocket.CloseError
			_ = errors.As(err, &ce)
			err = fmt.Errorf("closed (%d %s)", code, ce.Reason)
		}
		delay := time.Duration(math.Min(float64(backoffMax), float64(backoffBase)*math.Pow(2, float64(attempt)))) +
			time.Duration(rand.Int63n(int64(backoffJitter)))
		attempt++
		log.Printf("%v — reconnecting in %s", err, delay.Round(time.Millisecond))
		time.Sleep(delay)
	}
}