Repository Public
1// dYdX v4 indexer websocket (https://docs.dydx.xyz/indexer-client/websockets): one connection, many subscriptions.2// Channels: v4_orderbook, v4_trades (per ticker), v4_markets (all markets), v4_subaccounts (`<address>/<subaccountNumber>`).3//4// Reliability model5// - Every server message carries a per-connection `message_id`. Ids were observed OUT OF ORDER around subscriptions (the6// snapshot of a slow channel arrives after later ids), so order is not enforced; instead an id that is still missing7// `gapMs` after a later one arrived is a gap → status `stale` → reconnect and resubscribe (fresh snapshots).8// - An id already seen is a duplicate delivery and is dropped. Trades are also de-duplicated by trade id across reconnects.9// - No message at all for `staleMs` (the client pings every `pingMs`) → `stale` → reconnect.10// - Reconnect uses exponential backoff; every subscription is re-sent after `connected`.11import { z } from 'zod';12import { D } from '../../decimal.ts';13import type { Book, PublicTrade, Ticker } from '../../models.ts';14import type { StreamStatus } from '../../venue.ts';15import { instrumentId, orderOf, tickerOf, tradeOf, type DydxOrder } from './map.ts';16import * as W from './wire.ts';1718export interface WebSocketLike {19 send(data: string): void; close(): void;20 onopen: ((ev: unknown) => void) | null; onmessage: ((ev: { data: unknown }) => void) | null;21 onclose: ((ev: unknown) => void) | null; onerror: ((ev: unknown) => void) | null;22}23export type WebSocketCtor = new (url: string) => WebSocketLike;2425export type DydxStreamEvent =26 | { type: 'book'; symbol: string; snapshot: boolean; crossed: boolean; book: Book }27 | { type: 'trades'; symbol: string; snapshot: boolean; trades: PublicTrade[] }28 | { type: 'ticker'; symbol: string; snapshot: boolean; ticker: Ticker }29 /** account events are a trigger to re-read the account over REST: they carry the changed orders and flags, not a full state */30 | { type: 'account'; account: string; snapshot: boolean; subaccount: W.Subaccount | null;31 /** complete open orders, in the snapshot only */ orders: DydxOrder[];32 /** order changes: the indexer sends PARTIAL objects here (fields vary by cause), so only the identity and the venue status are normalized */33 orderUpdates: DydxOrderUpdate[]; hasFills: boolean; hasPositionChange: boolean; blockHeight: string | null; raw: unknown }34 | { type: 'error'; message: string };3536export interface DydxStreamOptions {37 url: string; WebSocket?: WebSocketCtor; now?: () => number;38 /** levels per side in emitted books (default 50) */ depth?: number;39 pingMs?: number; staleMs?: number; gapMs?: number; backoffMs?: number; maxBackoffMs?: number;40}41export interface DydxOrderUpdate { venueOrderId: string; clientId: string; instrument: string; venueStatus: string; filled: string | null; removalReason: string | null; raw: unknown }42type Handler = { symbol: string | null; on: (ev: DydxStreamEvent) => void; onStatus: (s: StreamStatus) => void };43interface Sub { channel: 'v4_orderbook' | 'v4_trades' | 'v4_markets' | 'v4_subaccounts'; id: string | null; handlers: Set<Handler>;44 bids: Map<string, string>; asks: Map<string, string>; markets: Map<string, Record<string, unknown>>; live: boolean }4546const envelope = z.object({ type: z.string(), message_id: z.number().int().nonnegative().optional(), channel: z.string().optional(), id: z.string().optional(), contents: z.unknown().optional(), message: z.string().optional() });47const bookSnapshot = W.orderbook;48const bookDelta = z.object({ bids: z.array(z.tuple([W.dec, W.dec]).rest(z.unknown())).optional(), asks: z.array(z.tuple([W.dec, W.dec]).rest(z.unknown())).optional() });49const marketsDelta = z.object({50 trading: z.record(z.string(), z.record(z.string(), z.unknown())).optional(),51 oraclePrices: z.record(z.string(), z.object({ oraclePrice: W.dec })).optional(),52});53const accountSnapshot = z.object({ subaccount: W.subaccount, orders: z.array(W.order).optional(), blockHeight: z.string().optional() });54const orderUpdate = z.looseObject({ id: z.string().min(1), clientId: z.string().regex(/^\d+$/), ticker: z.string().min(1), status: z.string().min(1),55 totalFilled: W.dec.optional(), totalOptimisticFilled: W.dec.optional(), removalReason: z.string().optional() });56const accountDelta = z.object({ orders: z.array(orderUpdate).optional(), fills: z.array(z.unknown()).optional(), perpetualPositions: z.array(z.unknown()).optional(),57 assetPositions: z.array(z.unknown()).optional(), blockHeight: z.string().optional() });5859const cmpDesc = (a: [string, string], b: [string, string]) => D.cmp(D.from(b[0]), D.from(a[0]));60const cmpAsc = (a: [string, string], b: [string, string]) => D.cmp(D.from(a[0]), D.from(b[0]));6162export class DydxStream {63 private readonly o: Required<Omit<DydxStreamOptions, 'WebSocket'>> & { WebSocket: WebSocketCtor };64 private readonly subs = new Map<string, Sub>();65 private ws: WebSocketLike | null = null;66 private connected = false;67 private attempt = 0;68 private nextId = 0;69 private readonly ahead = new Set<number>();70 private lastMessageAt = 0;71 private gapTimer: ReturnType<typeof setTimeout> | null = null;72 private retryTimer: ReturnType<typeof setTimeout> | null = null;73 private tick: ReturnType<typeof setInterval> | null = null;74 private readonly seenTrades = new Map<string, Set<string>>();7576 constructor(o: DydxStreamOptions) {77 const WS = o.WebSocket ?? (globalThis as { WebSocket?: WebSocketCtor }).WebSocket;78 if (!WS) throw new Error('no WebSocket implementation available');79 this.o = { url: o.url, WebSocket: WS, now: o.now ?? Date.now, depth: o.depth ?? 50, pingMs: o.pingMs ?? 15_000, staleMs: o.staleMs ?? 40_000,80 gapMs: o.gapMs ?? 5_000, backoffMs: o.backoffMs ?? 500, maxBackoffMs: o.maxBackoffMs ?? 30_000 };81 }8283 /** Subscribe; returns the unsubscribe function. `symbol` filters v4_markets events to one ticker. */84 subscribe(channel: Sub['channel'], id: string | null, symbol: string | null, on: Handler['on'], onStatus: Handler['onStatus']): () => void {85 const key = `${channel}|${id ?? ''}`;86 let sub = this.subs.get(key);87 const h: Handler = { symbol, on, onStatus };88 if (!sub) {89 sub = { channel, id, handlers: new Set(), bids: new Map(), asks: new Map(), markets: new Map(), live: false };90 this.subs.set(key, sub);91 if (this.connected) this.sendSub(sub);92 } else if (sub.live) {93 // late joiner on a channel that is already live: give it the current state as a snapshot94 queueMicrotask(() => { if (sub!.handlers.has(h)) this.replay(sub!, h); });95 }96 sub.handlers.add(h);97 if (!this.ws && !this.retryTimer) this.open();98 else if (this.connected) onStatus('connected');99 return () => {100 const s = this.subs.get(key);101 if (!s || !s.handlers.delete(h)) return;102 if (s.handlers.size === 0) {103 this.subs.delete(key);104 if (this.connected) this.send({ type: 'unsubscribe', channel: s.channel, ...(s.id === null ? {} : { id: s.id }) });105 }106 if (this.subs.size === 0) this.shutdown();107 };108 }109110 private send(m: unknown): void { try { this.ws?.send(JSON.stringify(m)); } catch { /* the close handler reconnects */ } }111 private sendSub(s: Sub): void { this.send({ type: 'subscribe', channel: s.channel, ...(s.id === null ? {} : { id: s.id }), batched: true }); }112 private status(s: StreamStatus): void { for (const sub of this.subs.values()) for (const h of sub.handlers) h.onStatus(s); }113114 private open(): void {115 this.retryTimer = null;116 const ws = new this.o.WebSocket(this.o.url);117 this.ws = ws;118 this.connected = false;119 this.lastMessageAt = this.o.now();120 ws.onmessage = (ev) => { if (this.ws === ws) this.onMessage(ev.data); };121 ws.onclose = () => { if (this.ws === ws) this.reconnect('reconnecting'); };122 ws.onerror = () => { if (this.ws === ws) this.reconnect('reconnecting'); };123 ws.onopen = () => { /* the server announces readiness with a `connected` message */ };124 if (!this.tick) {125 this.tick = setInterval(() => {126 if (!this.ws) return;127 if (this.o.now() - this.lastMessageAt > this.o.staleMs) { this.reconnect('stale'); return; }128 if (this.connected) this.send({ type: 'ping' });129 }, this.o.pingMs);130 }131 }132133 private drop(): void {134 const ws = this.ws;135 this.ws = null;136 this.connected = false;137 if (this.gapTimer) { clearTimeout(this.gapTimer); this.gapTimer = null; }138 this.ahead.clear();139 for (const s of this.subs.values()) s.live = false;140 if (ws) { ws.onmessage = ws.onclose = ws.onerror = ws.onopen = null; try { ws.close(); } catch { /* already closed */ } }141 }142 private reconnect(why: StreamStatus): void {143 this.drop();144 if (this.subs.size === 0) { this.shutdown(); return; }145 this.status(why);146 if (why === 'stale') this.status('reconnecting');147 if (this.retryTimer) return;148 const delay = Math.min(this.o.maxBackoffMs, this.o.backoffMs * 2 ** this.attempt);149 this.attempt += 1;150 this.retryTimer = setTimeout(() => this.open(), delay);151 }152 private shutdown(): void {153 this.drop();154 if (this.retryTimer) { clearTimeout(this.retryTimer); this.retryTimer = null; }155 if (this.tick) { clearInterval(this.tick); this.tick = null; }156 this.attempt = 0;157 }158159 /** true when this message id was already delivered */160 private track(id: number): boolean {161 if (id < this.nextId || this.ahead.has(id)) return true;162 if (id === this.nextId) {163 this.nextId += 1;164 while (this.ahead.delete(this.nextId)) this.nextId += 1;165 if (this.ahead.size === 0 && this.gapTimer) { clearTimeout(this.gapTimer); this.gapTimer = null; }166 } else {167 this.ahead.add(id);168 if (!this.gapTimer) this.gapTimer = setTimeout(() => { this.gapTimer = null; if (this.ahead.size > 0) this.reconnect('stale'); }, this.o.gapMs);169 }170 return false;171 }172173 private onMessage(data: unknown): void {174 this.lastMessageAt = this.o.now();175 let env: z.infer<typeof envelope>;176 try { env = envelope.parse(JSON.parse(typeof data === 'string' ? data : String(data))); }177 catch { this.reconnect('stale'); return; } // an unreadable frame may hide a state change178 if (env.type === 'connected') {179 this.connected = true;180 this.attempt = 0;181 this.nextId = (env.message_id ?? 0) + 1;182 this.ahead.clear();183 for (const s of this.subs.values()) this.sendSub(s);184 this.status('connected');185 return;186 }187 if (env.message_id !== undefined && this.track(env.message_id)) return;188 if (env.type === 'pong' || env.type === 'unsubscribed') return;189 if (env.type === 'error') {190 // e.g. "Invalid subscription id for channel: (v4_orderbook-NOPE-USD)"191 const m = /\((v4_[a-z]+)-([^)]+)\)/.exec(env.message ?? '');192 const target = m ? this.subs.get(`${m[1]}|${m[2]}`) : undefined;193 for (const s of target ? [target] : this.subs.values()) for (const h of s.handlers) h.on({ type: 'error', message: env.message ?? 'error' });194 return;195 }196 const sub = env.channel ? this.subs.get(`${env.channel}|${env.id ?? ''}`) : undefined;197 if (!sub) return; // late data for a channel we left198 try {199 if (env.type === 'subscribed') this.snapshot(sub, env.contents);200 else if (env.type === 'channel_data') this.delta(sub, env.contents);201 else if (env.type === 'channel_batch_data') for (const c of z.array(z.unknown()).parse(env.contents)) this.delta(sub, c);202 } catch { this.reconnect('stale'); } // malformed contents: the local state can no longer be trusted203 }204205 private book(sub: Sub): { book: Book; crossed: boolean } {206 const bids = [...sub.bids.entries()].sort(cmpDesc).slice(0, this.o.depth), asks = [...sub.asks.entries()].sort(cmpAsc).slice(0, this.o.depth);207 const crossed = bids.length > 0 && asks.length > 0 && D.cmp(D.from(bids[0]![0]), D.from(asks[0]![0])) >= 0;208 return { book: { instrument: instrumentId(sub.id!), bids, asks, ts: this.o.now() }, crossed };209 }210 private emit(sub: Sub, ev: DydxStreamEvent, only?: Handler): void { for (const h of only ? [only] : sub.handlers) h.on(ev); }211 private emitTicker(sub: Sub, symbol: string, snapshot: boolean, only?: Handler): void {212 const raw = sub.markets.get(symbol);213 if (!raw) return;214 const m = W.perpetualMarket.parse(raw);215 for (const h of only ? [only] : sub.handlers) if (h.symbol === symbol) h.on({ type: 'ticker', symbol, snapshot, ticker: tickerOf(m, this.o.now()) });216 }217 private replay(sub: Sub, h: Handler): void {218 if (sub.channel === 'v4_orderbook') this.emit(sub, { type: 'book', symbol: sub.id!, snapshot: true, ...this.book(sub) }, h);219 else if (sub.channel === 'v4_markets' && h.symbol) this.emitTicker(sub, h.symbol, true, h);220 h.onStatus('connected');221 }222223 private newTrades(symbol: string, list: z.infer<typeof W.trade>[]): PublicTrade[] {224 let seen = this.seenTrades.get(symbol);225 if (!seen) { seen = new Set(); this.seenTrades.set(symbol, seen); }226 const out: PublicTrade[] = [];227 for (const t of list) { if (seen.has(t.id)) continue; seen.add(t.id); out.push(tradeOf(symbol, t)); }228 if (seen.size > 4000) { const keep = [...seen].slice(-2000); seen.clear(); for (const k of keep) seen.add(k); }229 return out.sort((a, b) => a.ts - b.ts);230 }231232 private snapshot(sub: Sub, contents: unknown): void {233 sub.live = true;234 switch (sub.channel) {235 case 'v4_orderbook': {236 const b = bookSnapshot.parse(contents);237 sub.bids = new Map(b.bids.map((l) => [l.price, l.size]));238 sub.asks = new Map(b.asks.map((l) => [l.price, l.size]));239 this.emit(sub, { type: 'book', symbol: sub.id!, snapshot: true, ...this.book(sub) });240 return;241 }242 case 'v4_trades': {243 const t = W.trades.parse(contents);244 this.emit(sub, { type: 'trades', symbol: sub.id!, snapshot: true, trades: this.newTrades(sub.id!, t.trades) });245 return;246 }247 case 'v4_markets': {248 const m = z.object({ markets: z.record(z.string(), z.record(z.string(), z.unknown())) }).parse(contents);249 sub.markets = new Map(Object.entries(m.markets));250 for (const symbol of new Set([...sub.handlers].map((h) => h.symbol))) if (symbol) this.emitTicker(sub, symbol, true);251 return;252 }253 case 'v4_subaccounts': {254 const a = accountSnapshot.parse(contents);255 this.emit(sub, { type: 'account', account: sub.id!, snapshot: true, subaccount: a.subaccount, orders: (a.orders ?? []).map(orderOf), orderUpdates: [],256 hasFills: false, hasPositionChange: false, blockHeight: a.blockHeight ?? null, raw: contents });257 return;258 }259 }260 }261262 private delta(sub: Sub, contents: unknown): void {263 if (!sub.live) return; // data before the snapshot cannot be applied264 switch (sub.channel) {265 case 'v4_orderbook': {266 const d = bookDelta.parse(contents);267 for (const [side, levels] of [[sub.bids, d.bids], [sub.asks, d.asks]] as const) {268 for (const [price, size] of levels ?? []) {269 const p = W.canon(price), s = W.canon(size);270 if (D.isZero(D.from(s))) side.delete(p); else side.set(p, s);271 }272 }273 this.emit(sub, { type: 'book', symbol: sub.id!, snapshot: false, ...this.book(sub) });274 return;275 }276 case 'v4_trades': {277 const fresh = this.newTrades(sub.id!, W.trades.parse(contents).trades);278 if (fresh.length) this.emit(sub, { type: 'trades', symbol: sub.id!, snapshot: false, trades: fresh });279 return;280 }281 case 'v4_markets': {282 const d = marketsDelta.parse(contents);283 const touched = new Set<string>();284 for (const [t, patch] of Object.entries(d.trading ?? {})) { sub.markets.set(t, { ...(sub.markets.get(t) ?? {}), ...patch }); touched.add(t); }285 for (const [t, p] of Object.entries(d.oraclePrices ?? {})) { const cur = sub.markets.get(t); if (cur) { sub.markets.set(t, { ...cur, oraclePrice: p.oraclePrice }); touched.add(t); } }286 for (const t of touched) this.emitTicker(sub, t, false);287 return;288 }289 case 'v4_subaccounts': {290 const d = accountDelta.parse(contents);291 this.emit(sub, { type: 'account', account: sub.id!, snapshot: false, subaccount: null, orders: [],292 orderUpdates: (d.orders ?? []).map((u): DydxOrderUpdate => ({ venueOrderId: u.id, clientId: u.clientId, instrument: instrumentId(u.ticker), venueStatus: u.status,293 filled: u.totalFilled !== undefined ? W.canon(u.totalFilled) : u.totalOptimisticFilled !== undefined ? W.canon(u.totalOptimisticFilled) : null,294 removalReason: u.removalReason ?? null, raw: u })),295 hasFills: (d.fills?.length ?? 0) > 0, hasPositionChange: (d.perpetualPositions?.length ?? 0) > 0 || (d.assetPositions?.length ?? 0) > 0,296 blockHeight: d.blockHeight ?? null, raw: contents });297 return;298 }299 }300 }301}302