Repository Public
1// Hyperliquid websocket client: l2Book, trades, activeAssetCtx, allMids and the user feeds (userFills, orderUpdates, userEvents).2// https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/websocket (+ /subscriptions, /timeouts-and-heartbeats)3// "The server will close any connection if it hasn't sent a message to it in the last 60 seconds" → { "method": "ping" } /4// { "channel": "pong" }. "all automated users should handle disconnects from the server side and gracefully reconnect …5// Missed data during the reconnect will be present in the snapshot ack on reconnect."6// One connection per adapter, shared by every subscription. It reconnects by itself (exponential backoff with jitter),7// subscribes again, and tells each consumer that a GAP may exist: status 'reconnecting' → 'connected' plus a `{ type: 'gap' }`8// event. A consumer must then take a fresh snapshot (REST) before trusting its local state. Fills and public trades are9// de-duplicated by id across messages, snapshots and reconnects.10import type { z } from 'zod';11import type { DecStr } from '../../decimal.ts';12import type { Book, Fill, FundingPayment, PublicTrade, Ticker } from '../../models.ts';13import { VenueError } from './error.ts';14import { hlAccount, hlBook, hlFillId, hlFillOf, hlOrderStatus, hlPublicTrade, hlTicker, hlTradeId } from './mappers.ts';15import { HL_WS } from './http.ts';16import { hlL2Book, hlTrades, hlWsActiveAssetCtx, hlWsAllMids, hlWsEnvelope, hlWsOrders, hlWsUserEvent, hlWsUserFills } from './schemas.ts';17import type { HlNetwork } from './signing.ts';1819/** the part of the WHATWG WebSocket that is used; tests inject a fake */20export interface HlWebSocketLike {21 send(data: string): void; close(code?: number, reason?: string): void;22 onopen: ((ev: unknown) => void) | null; onmessage: ((ev: { data: unknown }) => void) | null;23 onclose: ((ev: unknown) => void) | null; onerror: ((ev: unknown) => void) | null;24}25export type HlWebSocketCtor = new (url: string) => HlWebSocketLike;2627export interface HlStreamOptions {28 network: HlNetwork; now?: () => number; url?: string;29 /** defaults to the global WebSocket (Node ≥ 22, browsers) */30 WebSocket?: HlWebSocketCtor;31 /** heartbeat period; the server drops a connection it has sent nothing to for 60 s. Default 15 s. */32 pingIntervalMs?: number;33 /** no message at all (data or pong) for this long → 'stale'. Default 25 s. */34 staleAfterMs?: number;35 /** still nothing this long after 'stale' → the connection is dropped and re-opened. Default 10 s. */36 staleGraceMs?: number;37 /** reconnect backoff bounds; the venue allows 30 new connections per minute. Default 1 s … 30 s. */38 backoffMinMs?: number; backoffMaxMs?: number;39 random?: () => number;40 /** how many fill / trade ids are remembered per feed. Default 5000. */41 dedupeSize?: number;42}4344export interface HlOrderUpdate { instrument: string; venueOrderId: string; clientId: string | null; side: 'buy' | 'sell'; price: DecStr; size: DecStr; remaining: DecStr;45 status: 'open' | 'filled' | 'partially_filled' | 'cancelled' | 'rejected' | 'triggered' | 'pending'; venueStatus: string; createdAt: number; updatedAt: number }46export type HlStreamEvent =47 | { type: 'book'; book: Book }48 | { type: 'trades'; trades: PublicTrade[] }49 | { type: 'ticker'; ticker: Ticker }50 | { type: 'mids'; mids: Record<string, DecStr>; ts: number }51 | { type: 'fills'; fills: Fill[]; snapshot: boolean }52 | { type: 'orders'; orders: HlOrderUpdate[] }53 | { type: 'funding'; payment: FundingPayment }54 | { type: 'liquidation'; raw: unknown }55 | { type: 'cancelled_by_venue'; orders: { instrument: string; venueOrderId: string }[] }56 /** the connection was lost and re-established: anything may have been missed — take a snapshot again */57 | { type: 'gap'; reason: 'reconnected' }58 /** a message that failed validation was dropped */59 | { type: 'invalid'; channel: string; reason: string };6061export type HlStreamSub = { type: 'book' | 'trades' | 'ticker'; symbol: string } | { type: 'account'; account: string } | { type: 'mids' };62export type HlStreamStatus = 'connected' | 'reconnecting' | 'stale';6364export interface HlStream {65 subscribe(sub: HlStreamSub, on: (ev: unknown) => void, onStatus: (s: HlStreamStatus) => void): () => void;66 close(): void;67 stats(): { connects: number; messages: number; invalid: number; duplicates: number; subscriptions: number; status: HlStreamStatus | 'idle' };68}6970type WireSub = Record<string, string>;71interface Consumer { keys: string[]; on: (ev: HlStreamEvent) => void; onStatus: (s: HlStreamStatus) => void; status: HlStreamStatus | null }7273/** a connection must have lasted this long before the reconnect backoff starts again from its minimum */74const STABLE_MS = 10_000;75const keyOf = (s: WireSub): string => `${s.type}:${s.coin ?? s.user ?? ''}`;7677class Seen {78 private readonly set = new Set<string>(); private readonly max: number;79 constructor(max: number) { this.max = max; }80 /** true the first time an id is offered */81 add(id: string): boolean {82 if (this.set.has(id)) return false;83 this.set.add(id);84 if (this.set.size > this.max) this.set.delete(this.set.values().next().value!);85 return true;86 }87}8889export function createHlStream(o: HlStreamOptions): HlStream {90 const now = o.now ?? Date.now, url = o.url ?? HL_WS[o.network], random = o.random ?? Math.random;91 const pingIntervalMs = o.pingIntervalMs ?? 15_000, staleAfterMs = o.staleAfterMs ?? 25_000, staleGraceMs = o.staleGraceMs ?? 10_000;92 const backoffMinMs = o.backoffMinMs ?? 1000, backoffMaxMs = o.backoffMaxMs ?? 30_000, dedupeSize = o.dedupeSize ?? 5000;9394 const consumers = new Set<Consumer>();95 const wire = new Map<string, { sub: WireSub; refs: number }>();96 const seen = new Map<string, Seen>(); // per feed, kept across reconnects97 const seenFor = (k: string): Seen => { let s = seen.get(k); if (!s) { s = new Seen(dedupeSize); seen.set(k, s); } return s; };98 let ws: HlWebSocketLike | null = null, open = false, everConnected = false, attempt = 0, lastMessageAt = 0, openedAt = 0, user: string | null = null;99 let pingTimer: ReturnType<typeof setInterval> | null = null, watchTimer: ReturnType<typeof setInterval> | null = null, retryTimer: ReturnType<typeof setTimeout> | null = null;100 const st = { connects: 0, messages: 0, invalid: 0, duplicates: 0 };101 let status: HlStreamStatus | 'idle' = 'idle';102103 const setStatus = (s: HlStreamStatus): void => {104 status = s;105 for (const c of consumers) if (c.status !== s) { c.status = s; try { c.onStatus(s); } catch { /* a consumer must not break the stream */ } }106 };107 const emit = (key: string | null, ev: HlStreamEvent): void => {108 for (const c of consumers) if (key === null || c.keys.includes(key)) { try { c.on(ev); } catch { /* idem */ } }109 };110 const send = (msg: unknown): void => { if (ws && open) { try { ws.send(JSON.stringify(msg)); } catch { /* the close handler reconnects */ } } };111112 const stopTimers = (): void => {113 if (pingTimer) clearInterval(pingTimer); if (watchTimer) clearInterval(watchTimer);114 pingTimer = watchTimer = null;115 };116 const drop = (): void => { // forget the current socket without reacting to its late events117 stopTimers();118 if (ws) { const old = ws; ws = null; open = false; old.onopen = old.onmessage = old.onclose = old.onerror = null; try { old.close(); } catch { /* already closed */ } }119 };120 const scheduleReconnect = (): void => {121 // the backoff starts again only after a connection that lasted: a server that accepts and drops at once (which is what it122 // does on a subscription to an unknown coin) must not be hammered — the venue allows 30 new connections per minute123 if (open && now() - openedAt >= STABLE_MS) attempt = 0;124 drop();125 if (consumers.size === 0) { status = 'idle'; return; }126 setStatus('reconnecting');127 if (retryTimer) return;128 const delay = Math.min(backoffMaxMs, backoffMinMs * 2 ** Math.min(attempt, 16)) * (0.5 + 0.5 * random());129 attempt += 1;130 retryTimer = setTimeout(() => { retryTimer = null; connect(); }, delay);131 };132133 const invalid = (channel: string, key: string | null, reason: string): void => { st.invalid += 1; emit(key, { type: 'invalid', channel, reason: reason.slice(0, 160) }); };134 const parse = <S extends z.ZodType>(schema: S, data: unknown, channel: string, key: (d: z.output<S>) => string | null): { data: z.output<S>; key: string } | null => {135 const r = schema.safeParse(data);136 if (!r.success) { invalid(channel, null, r.error.issues.slice(0, 2).map((i) => `${i.path.join('.')} ${i.message}`).join('; ')); return null; }137 const k = key(r.data);138 if (k === null || !wire.has(k)) return null; // not (or no longer) subscribed: ignore139 return { data: r.data, key: k };140 };141 const fresh = <T>(feed: string, items: T[], id: (x: T) => string): T[] => {142 const s = seenFor(feed);143 const out = items.filter((x) => s.add(id(x)));144 st.duplicates += items.length - out.length;145 return out;146 };147 const fillsOut = (fills: z.output<typeof hlWsUserFills>['fills'], snapshot: boolean): void => {148 if (user === null) return;149 const key = `userFills:${user}`, out = fresh(`fills:${user}`, fills, hlFillId);150 if (out.length || snapshot) emit(key, { type: 'fills', fills: out.map(hlFillOf).sort((a, b) => a.ts - b.ts), snapshot });151 };152153 const onMessage = (raw: unknown): void => {154 lastMessageAt = now(); st.messages += 1;155 if (status === 'stale') setStatus('connected'); // same connection, the transport is ordered and lossless: no gap156 let msg: unknown;157 try { msg = JSON.parse(typeof raw === 'string' ? raw : String(raw)); } catch { invalid('?', null, 'message is not JSON'); return; }158 const env = hlWsEnvelope.safeParse(msg);159 if (!env.success) { invalid('?', null, 'message has no channel'); return; }160 const { channel, data } = env.data;161 try {162 switch (channel) {163 case 'pong': case 'subscriptionResponse': return;164 case 'error': invalid('error', null, typeof data === 'string' ? data : JSON.stringify(data ?? null)); return;165 case 'l2Book': { const p = parse(hlL2Book, data, channel, (d) => `l2Book:${d.coin}`); if (p) emit(p.key, { type: 'book', book: hlBook(p.data) }); return; }166 case 'trades': {167 const p = parse(hlTrades, data, channel, (d) => (d[0] ? `trades:${d[0].coin}` : null));168 if (!p) return;169 if (p.data.some((t) => t.coin !== p.data[0]!.coin)) { invalid(channel, p.key, 'trades of several coins in one message'); return; }170 const out = fresh(p.key, p.data, hlTradeId);171 if (out.length) emit(p.key, { type: 'trades', trades: out.map(hlPublicTrade).sort((a, b) => a.ts - b.ts) });172 return;173 }174 case 'activeAssetCtx': { const p = parse(hlWsActiveAssetCtx, data, channel, (d) => `activeAssetCtx:${d.coin}`); if (p) emit(p.key, { type: 'ticker', ticker: hlTicker(p.data.coin, p.data.ctx, now()) }); return; }175 case 'allMids': { const p = parse(hlWsAllMids, data, channel, () => 'allMids:'); if (p) emit(p.key, { type: 'mids', mids: p.data.mids, ts: now() }); return; }176 case 'userFills': { const p = parse(hlWsUserFills, data, channel, (d) => `userFills:${d.user}`); if (p && p.data.user === user) fillsOut(p.data.fills, p.data.isSnapshot === true); return; }177 case 'orderUpdates': {178 const p = parse(hlWsOrders, data, channel, () => (user ? `orderUpdates:${user}` : null));179 if (!p) return;180 emit(p.key, { type: 'orders', orders: p.data.map((u): HlOrderUpdate => {181 const s = hlOrderStatus(u.status).status, partly = s === 'open' && u.order.sz !== u.order.origSz;182 return { instrument: `hyperliquid:${u.order.coin}`, venueOrderId: String(u.order.oid), clientId: u.order.cloid ?? null, side: u.order.side === 'B' ? 'buy' : 'sell',183 price: u.order.limitPx, size: u.order.origSz, remaining: u.order.sz, status: partly ? 'partially_filled' : s, venueStatus: u.status, createdAt: u.order.timestamp, updatedAt: u.statusTimestamp };184 }) });185 return;186 }187 case 'user': { // userEvents: fills (also delivered by userFills — de-duplicated), funding, liquidation, nonUserCancel188 const p = parse(hlWsUserEvent, data, channel, () => (user ? `userEvents:${user}` : null));189 if (!p) return;190 const d = p.data;191 if (d.fills) fillsOut(d.fills, false);192 if (d.funding) emit(p.key, { type: 'funding', payment: { instrument: `hyperliquid:${d.funding.coin}`, amount: d.funding.usdc, rate: d.funding.fundingRate, size: d.funding.szi, ts: d.funding.time } });193 if (d.liquidation) emit(p.key, { type: 'liquidation', raw: d.liquidation });194 if (d.nonUserCancel) emit(p.key, { type: 'cancelled_by_venue', orders: d.nonUserCancel.map((c) => ({ instrument: `hyperliquid:${c.coin}`, venueOrderId: String(c.oid) })) });195 return;196 }197 default: return; // a channel we did not ask for198 }199 } catch (e) {200 invalid(channel, null, e instanceof Error ? e.message : 'handler error'); // e.g. a crossed book or an unknown order status201 }202 };203204 function connect(): void {205 if (ws || consumers.size === 0) return;206 const Ctor = o.WebSocket ?? (globalThis as { WebSocket?: HlWebSocketCtor }).WebSocket;207 if (!Ctor) throw new VenueError('hyperliquid', 'unsupported', 'no WebSocket implementation available');208 let sock: HlWebSocketLike;209 try { sock = new Ctor(url); } catch { scheduleReconnect(); return; }210 ws = sock; st.connects += 1;211 sock.onopen = () => {212 if (ws !== sock) return;213 open = true; lastMessageAt = openedAt = now();214 for (const w of wire.values()) send({ method: 'subscribe', subscription: w.sub });215 pingTimer = setInterval(() => send({ method: 'ping' }), pingIntervalMs);216 watchTimer = setInterval(() => {217 const quiet = now() - lastMessageAt;218 if (quiet > staleAfterMs + staleGraceMs) scheduleReconnect();219 else if (quiet > staleAfterMs && status === 'connected') setStatus('stale');220 }, Math.max(250, Math.min(1000, Math.floor(staleGraceMs / 4))));221 const again = everConnected;222 everConnected = true;223 setStatus('connected');224 if (again) emit(null, { type: 'gap', reason: 'reconnected' });225 };226 sock.onmessage = (ev) => { if (ws === sock) onMessage(ev.data); };227 sock.onclose = () => { if (ws === sock) scheduleReconnect(); };228 sock.onerror = () => { if (ws === sock) scheduleReconnect(); };229 }230231 const wireSubs = (sub: HlStreamSub): WireSub[] => {232 if (sub.type === 'mids') return [{ type: 'allMids' }];233 if (sub.type === 'account') {234 const u = hlAccount(sub.account);235 // orderUpdates and userEvents messages do not name their user: one account per connection keeps them unambiguous236 if (user !== null && user !== u) throw new VenueError('hyperliquid', 'unsupported', 'one account per stream connection; create another adapter for a second account');237 return [{ type: 'userFills', user: u }, { type: 'orderUpdates', user: u }, { type: 'userEvents', user: u }];238 }239 if (typeof sub.symbol !== 'string' || !/^[A-Za-z0-9:_./@-]{1,40}$/.test(sub.symbol)) throw new VenueError('hyperliquid', 'unsupported', 'bad symbol');240 if (sub.type === 'book') return [{ type: 'l2Book', coin: sub.symbol }];241 if (sub.type === 'trades') return [{ type: 'trades', coin: sub.symbol }];242 if (sub.type === 'ticker') return [{ type: 'activeAssetCtx', coin: sub.symbol }];243 throw new VenueError('hyperliquid', 'unsupported', `unknown subscription type '${String((sub as { type: unknown }).type)}'`);244 };245246 return {247 subscribe(sub, on, onStatus) {248 const subs = wireSubs(sub);249 if (sub.type === 'account') user = hlAccount(sub.account);250 const c: Consumer = { keys: subs.map(keyOf), on: on as (ev: HlStreamEvent) => void, onStatus, status: null };251 consumers.add(c);252 for (const s of subs) {253 const k = keyOf(s), w = wire.get(k);254 if (w) w.refs += 1; else { wire.set(k, { sub: s, refs: 1 }); send({ method: 'subscribe', subscription: s }); }255 }256 if (status !== 'idle') { c.status = status; onStatus(status); } // late joiner on a live connection257 if (!ws && !retryTimer) connect();258 let done = false;259 return () => {260 if (done) return;261 done = true; consumers.delete(c);262 for (const k of c.keys) {263 const w = wire.get(k);264 if (!w) continue;265 w.refs -= 1;266 if (w.refs <= 0) { wire.delete(k); send({ method: 'unsubscribe', subscription: w.sub }); }267 }268 if (![...wire.keys()].some((k) => k.startsWith('userFills:'))) user = null;269 if (consumers.size === 0) { if (retryTimer) { clearTimeout(retryTimer); retryTimer = null; } drop(); status = 'idle'; everConnected = false; attempt = 0; }270 };271 },272 close() {273 consumers.clear(); wire.clear(); user = null;274 if (retryTimer) { clearTimeout(retryTimer); retryTimer = null; }275 drop(); status = 'idle'; everConnected = false; attempt = 0;276 },277 stats: () => ({ ...st, subscriptions: wire.size, status }),278 };279}280