From 083f0984290cda1577949e6db780f00432125ece Mon Sep 17 00:00:00 2001 From: nktkas Date: Wed, 17 Jun 2026 14:29:30 +0300 Subject: [PATCH] feat(subscription): add `fastAssetCtxs` subscription Co-authored-by: Claude Opus 4.8 --- .../subscription/_methods/fastAssetCtxs.ts | 117 ++++++++++++++++++ src/api/subscription/client.ts | 36 ++++++ src/api/subscription/mod.ts | 1 + tests/api/subscription/fastAssetCtxs.test.ts | 19 +++ 4 files changed, 173 insertions(+) create mode 100644 src/api/subscription/_methods/fastAssetCtxs.ts create mode 100644 tests/api/subscription/fastAssetCtxs.test.ts diff --git a/src/api/subscription/_methods/fastAssetCtxs.ts b/src/api/subscription/_methods/fastAssetCtxs.ts new file mode 100644 index 00000000..cb04679c --- /dev/null +++ b/src/api/subscription/_methods/fastAssetCtxs.ts @@ -0,0 +1,117 @@ +import * as v from "@valibot/valibot"; + +// ============================================================ +// API Schemas +// ============================================================ + +/** + * Subscription to mark and mid price events for all assets. + * @see https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/websocket/subscriptions + */ +export const FastAssetCtxsRequest = /* @__PURE__ */ (() => { + return v.object({ + /** Type of subscription. */ + type: v.literal("fastAssetCtxs"), + }); +})(); +export type FastAssetCtxsRequest = v.InferOutput; + +/** + * Event of mark and mid prices, keyed by coin. + * + * The first message after subscribing is a full snapshot; later messages contain only the changed coins. + * @see https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/websocket/subscriptions + */ +export type FastAssetCtxsEvent = { + /** Mark and mid prices for a single asset. */ + [coin: string]: { + /** + * Mark price. + * @pattern ^[0-9]+(\.[0-9]+)?$ + */ + markPx?: string; + /** + * Mid price. + * @pattern ^[0-9]+(\.[0-9]+)?$ + */ + midPx?: string | null; + }; +}; + +// ============================================================ +// Execution Logic +// ============================================================ + +import { parse } from "../../../_base.ts"; +import type { ISubscription, TransportError } from "../../../transport/mod.ts"; +import type { SubscriptionConfig } from "./_base/mod.ts"; + +/** + * Subscribe to mark and mid prices for all assets. + * + * NOTE: payloads are decompressed with [`DecompressionStream`](https://developer.mozilla.org/en-US/docs/Web/API/DecompressionStream), + * which React Native does not provide; add a [polyfill](https://www.npmjs.com/package/compression-streams-polyfill) to use this subscription there. + * + * @param config General configuration for Subscription API subscriptions. + * @param listener A callback function to be called when the event is received. + * @param onError An optional callback function to be called when the subscription fails. + * @return A request-promise that resolves with a {@link ISubscription} object to manage the subscription lifecycle. + * + * @throws {ValidationError} When the request parameters fail validation (before sending). + * @throws {TransportError} When the transport layer throws an error. + * + * @example + * ```ts + * import { WebSocketTransport } from "@nktkas/hyperliquid"; + * import { fastAssetCtxs } from "@nktkas/hyperliquid/api/subscription"; + * + * const transport = new WebSocketTransport(); + * + * const sub = await fastAssetCtxs( + * { transport }, + * (data) => console.log(data), + * ); + * ``` + * + * @see https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/websocket/subscriptions + */ +export function fastAssetCtxs( + config: SubscriptionConfig, + listener: (data: FastAssetCtxsEvent) => void, + onError?: (error: TransportError) => void, +): Promise { + const payload = parse(FastAssetCtxsRequest, { type: "fastAssetCtxs" }); + // The server pushes each update as a base64 + raw DEFLATE (RFC 1951) compressed JSON string (assumed to be valid). + // Decompress sequentially so events reach the listener in arrival order. + let queue = Promise.resolve(); + return config.transport.subscribe(payload.type, payload, (e) => { + queue = queue.then(async () => listener(await decompress(e.detail))); + }, { onError }); +} + +/** Decode a base64 + raw DEFLATE (RFC 1951) payload into a {@linkcode FastAssetCtxsEvent}. */ +async function decompress(data: string): Promise { + const bytes = Uint8Array.from(atob(data), (c) => c.charCodeAt(0)); + + const stream = new DecompressionStream("deflate-raw"); + const writer = stream.writable.getWriter(); + // Do not await write/close before draining: backpressure on multi-chunk output would deadlock. + writer.write(bytes); + writer.close(); + + const reader = stream.readable.getReader(); + const chunks: Uint8Array[] = []; + let result = await reader.read(); + while (!result.done) { + chunks.push(result.value); + result = await reader.read(); + } + + const merged = new Uint8Array(chunks.reduce((total, chunk) => total + chunk.length, 0)); + let offset = 0; + for (const chunk of chunks) { + merged.set(chunk, offset); + offset += chunk.length; + } + return JSON.parse(new TextDecoder().decode(merged)); +} diff --git a/src/api/subscription/client.ts b/src/api/subscription/client.ts index 712b34f5..16743d6d 100644 --- a/src/api/subscription/client.ts +++ b/src/api/subscription/client.ts @@ -36,6 +36,7 @@ import { type ClearinghouseStateEvent, type ClearinghouseStateParameters, } from "./_methods/clearinghouseState.ts"; +import { fastAssetCtxs, type FastAssetCtxsEvent } from "./_methods/fastAssetCtxs.ts"; import { l2Book, type L2BookEvent, type L2BookParameters } from "./_methods/l2Book.ts"; import { notification, type NotificationEvent, type NotificationParameters } from "./_methods/notification.ts"; import { openOrders, type OpenOrdersEvent, type OpenOrdersParameters } from "./_methods/openOrders.ts"; @@ -461,6 +462,40 @@ export class SubscriptionClient { + * console.log(data); + * }); + * ``` + * + * @see https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/websocket/subscriptions + */ + fastAssetCtxs( + listener: (data: FastAssetCtxsEvent) => void, + onError?: (error: TransportError) => void, + ): Promise { + return fastAssetCtxs(this.config_, listener, onError); + } + /** * Subscribe to L2 order book updates for a specific asset. * @@ -1088,6 +1123,7 @@ export type { ClearinghouseStateEvent as ClearinghouseStateWsEvent, ClearinghouseStateParameters as ClearinghouseStateWsParameters, } from "./_methods/clearinghouseState.ts"; +export type { FastAssetCtxsEvent as FastAssetCtxsWsEvent } from "./_methods/fastAssetCtxs.ts"; export type { L2BookEvent as L2BookWsEvent, L2BookParameters as L2BookWsParameters } from "./_methods/l2Book.ts"; export type { NotificationEvent as NotificationWsEvent, diff --git a/src/api/subscription/mod.ts b/src/api/subscription/mod.ts index e7406c29..113f3538 100644 --- a/src/api/subscription/mod.ts +++ b/src/api/subscription/mod.ts @@ -35,6 +35,7 @@ export * from "./_methods/assetCtxs.ts"; export * from "./_methods/bbo.ts"; export * from "./_methods/candle.ts"; export * from "./_methods/clearinghouseState.ts"; +export * from "./_methods/fastAssetCtxs.ts"; export * from "./_methods/l2Book.ts"; export * from "./_methods/notification.ts"; export * from "./_methods/openOrders.ts"; diff --git a/tests/api/subscription/fastAssetCtxs.test.ts b/tests/api/subscription/fastAssetCtxs.test.ts new file mode 100644 index 00000000..babbd624 --- /dev/null +++ b/tests/api/subscription/fastAssetCtxs.test.ts @@ -0,0 +1,19 @@ +import type { FastAssetCtxsEvent } from "@nktkas/hyperliquid/api/subscription"; +import { schemaCoverage } from "../_utils/schemaCoverage.ts"; +import { typeToJsonSchema } from "../_utils/typeToJsonSchema.ts"; +import { collectEventsOverTime, runTest } from "./_t.ts"; + +const sourceFile = new URL("../../../src/api/subscription/_methods/fastAssetCtxs.ts", import.meta.url).pathname; +const responseSchema = typeToJsonSchema(sourceFile, "FastAssetCtxsEvent"); + +runTest({ + name: "fastAssetCtxs", + mode: "api", + fn: async (_t, client) => { + const data = await collectEventsOverTime(async (cb) => { + await client.fastAssetCtxs(cb); + }, 10_000); + + schemaCoverage(responseSchema, data); + }, +});