diff --git a/workspace/data-proxy/src/config/binance-module-config.ts b/workspace/data-proxy/src/config/binance-module-config.ts index ca42775..a055ba4 100644 --- a/workspace/data-proxy/src/config/binance-module-config.ts +++ b/workspace/data-proxy/src/config/binance-module-config.ts @@ -5,7 +5,12 @@ import { validateTickerModuleRoute, } from "./ticker-module-config"; -export const BINANCE_STREAM_TYPES = ["bookTicker", "trade"] as const; +export const BINANCE_STREAM_TYPES = [ + "bookTicker", + "trade", + "depth@100ms", + "depth@1000ms", +] as const; export type BinanceStreamType = (typeof BINANCE_STREAM_TYPES)[number]; diff --git a/workspace/data-proxy/src/modules/shared/venue-ws.test.ts b/workspace/data-proxy/src/modules/shared/venue-ws.test.ts index b5d8213..2a77e49 100644 --- a/workspace/data-proxy/src/modules/shared/venue-ws.test.ts +++ b/workspace/data-proxy/src/modules/shared/venue-ws.test.ts @@ -28,6 +28,19 @@ const startClient = (options?: { maxMessagesWindow?: Duration.Duration; reconnectSchedule?: Schedule.Schedule; preSubscribed?: string[]; + seed?: { + start: (keys: string[]) => Effect.Effect; + stop: (keys: string[]) => Effect.Effect; + restart: (keys: string[]) => Effect.Effect; + }; + parseInboundFrame?: (raw: string) => { + kind: "tickers"; + frames: Array<{ + key: string; + frame: Record; + action?: "write" | "ignore" | "resubscribe" | "reseed"; + }>; + } | null; }) => Effect.gen(function* () { const cache = yield* createPriceCache>(); @@ -54,21 +67,24 @@ const startClient = (options?: { buildSubscribeFrame: (keys) => JSON.stringify({ op: "subscribe", keys }), buildUnsubscribeFrame: (keys) => JSON.stringify({ op: "unsubscribe", keys }), - parseInboundFrame: (raw) => { - let json: unknown; - try { - json = JSON.parse(raw); - } catch { - return null; - } - if (typeof json !== "object" || json === null) return null; - const record = json as Record; - if (typeof record.s !== "string") return null; - return { - kind: "tickers" as const, - frames: [{ key: record.s, frame: record }], - }; - }, + seed: options?.seed, + parseInboundFrame: + options?.parseInboundFrame ?? + ((raw) => { + let json: unknown; + try { + json = JSON.parse(raw); + } catch { + return null; + } + if (typeof json !== "object" || json === null) return null; + const record = json as Record; + if (typeof record.s !== "string") return null; + return { + kind: "tickers" as const, + frames: [{ key: record.s, frame: record }], + }; + }), }); if (options?.preSubscribed !== undefined) { yield* ws.subscribe(options.preSubscribed); @@ -476,6 +492,67 @@ describe("createVenueWS", () => { await Effect.runPromise(Fiber.interrupt(fiber)); }); + it("seeds new keys, reseeds without a socket frame, and reseeds on reconnect", async () => { + const calls: string[] = []; + const note = + (label: string) => + (keys: string[]): Effect.Effect => + Effect.sync(() => { + calls.push(`${label}:${keys.join(",")}`); + }); + const { cache, fiber } = await Effect.runPromise( + startClient({ + preSubscribed: ["BTC"], + reconnectSchedule: Schedule.spaced(Duration.millis(10)), + seed: { + start: note("start"), + stop: note("stop"), + restart: note("restart"), + }, + parseInboundFrame: () => ({ + kind: "tickers", + frames: [ + { + key: "BTC", + frame: { s: "BTC" }, + action: "reseed", + }, + ], + }), + }), + ); + await flush(); + const ws1 = FakeWebSocket.instances[0]; + ws1.triggerOpen(); + await flush(); + + expect(calls).toEqual(["start:BTC"]); + expect(ws1.sent.length).toBe(1); + + ws1.triggerMessage(JSON.stringify({ s: "BTC" })); + await flush(); + expect(calls).toEqual(["start:BTC", "restart:BTC"]); + expect(cache.size()).toBe(0); + expect(ws1.sent.length).toBe(1); + + ws1.triggerClose(); + await flush(); + expect(calls).toEqual(["start:BTC", "restart:BTC", "stop:BTC"]); + + await new Promise((r) => setTimeout(r, 40)); + const ws2 = FakeWebSocket.instances[1]; + ws2.triggerOpen(); + await flush(); + expect(calls).toEqual([ + "start:BTC", + "restart:BTC", + "stop:BTC", + "start:BTC", + ]); + + await Effect.runPromise(Fiber.interrupt(fiber)); + }); + it("does not send keepalive when it is omitted", async () => { const { fiber } = await Effect.runPromise(startClient()); await flush(); diff --git a/workspace/data-proxy/src/modules/shared/venue-ws.ts b/workspace/data-proxy/src/modules/shared/venue-ws.ts index 70502cb..d52c11f 100644 --- a/workspace/data-proxy/src/modules/shared/venue-ws.ts +++ b/workspace/data-proxy/src/modules/shared/venue-ws.ts @@ -44,7 +44,7 @@ export interface VenueWS { hasError(): Effect.Effect; } -export type InboundFrameAction = "write" | "ignore" | "resubscribe"; +export type InboundFrameAction = "write" | "ignore" | "resubscribe" | "reseed"; export type ParsedTickersFrame = { key: TKey; @@ -52,6 +52,8 @@ export type ParsedTickersFrame = { /** * Omitted means write. `ignore` drops the frame. `resubscribe` * clears the cached value and subscribes again for a snapshot. + * `reseed` clears the cached value and asks `seed.restart` for + * a fresh snapshot without changing the socket subscription. */ action?: InboundFrameAction; }; @@ -87,6 +89,17 @@ export interface CreateVenueWSParams { buildSubscribeFrame: (keys: TKey[]) => string | string[]; buildUnsubscribeFrame: (keys: TKey[]) => string | string[]; parseInboundFrame: (raw: string) => VenueParsedInbound | null; + /** + * Optional snapshot source for venues whose socket does not send one. + * `start` runs for newly subscribed keys and for every desired key on + * reconnect. `stop` runs for removed keys and for every desired key when + * the socket closes after it has opened. `restart` runs on `reseed`. + */ + seed?: { + start: (keys: TKey[]) => Effect.Effect; + stop: (keys: TKey[]) => Effect.Effect; + restart: (keys: TKey[]) => Effect.Effect; + }; } export const createVenueWS = ( @@ -101,6 +114,7 @@ export const createVenueWS = ( buildSubscribeFrame, buildUnsubscribeFrame, parseInboundFrame, + seed, } = params; const { name, wsUrl, maxMessages, maxMessagesWindow } = config; const reconnectSchedule = @@ -118,6 +132,9 @@ export const createVenueWS = ( const desiredKeys = MutableHashMap.empty(); let currentWS: WebSocket | null = null; let unhealthy = false; + // A close before the first open must not cancel a seed that subscribe + // already started; there is no reconnect yet to start it again. + let opened = false; const outbound = yield* Queue.unbounded(); const activeSubscriptions = Metric.gauge(`${venue}_active_subscriptions`, { description: `Number of active ${venue} WebSocket stream subscriptions`, @@ -152,6 +169,12 @@ export const createVenueWS = ( const isDesired = (key: TKey): boolean => Option.isSome(MutableHashMap.get(desiredKeys, key)); + const desiredKeyList = (): TKey[] => { + const keys: TKey[] = []; + for (const [key] of desiredKeys) keys.push(key); + return keys; + }; + const enqueue = (frame: string, type: OutboundMessageType) => Queue.offer(outbound, { frame, type }).pipe(Effect.asVoid); @@ -252,6 +275,7 @@ export const createVenueWS = ( if (fresh.length === 0) return; yield* setActiveSubscriptions(); yield* enqueueFrames(buildSubscribeFrame(fresh), "subscribe"); + if (seed !== undefined) yield* seed.start(fresh); }); const unsubscribe = (keys: TKey[]) => @@ -265,6 +289,7 @@ export const createVenueWS = ( if (removed.length === 0) return; yield* setActiveSubscriptions(); yield* enqueueFrames(buildUnsubscribeFrame(removed), "unsubscribe"); + if (seed !== undefined) yield* seed.stop(removed); }); const hasError = () => Effect.sync(() => unhealthy); @@ -274,17 +299,22 @@ export const createVenueWS = ( yield* Effect.logInfo(`${venue} WS open`, { name }); unhealthy = false; currentWS = ws; - const keys: TKey[] = []; - for (const [key] of desiredKeys) keys.push(key); + const keys = desiredKeyList(); if (keys.length > 0) { yield* enqueueFrames(buildSubscribeFrame(keys), "subscribe"); + if (opened && seed !== undefined) yield* seed.start(keys); } + opened = true; }); const handleDisconnect = (closed: Deferred.Deferred) => Effect.gen(function* () { unhealthy = true; currentWS = null; + if (opened && seed !== undefined) { + const keys = desiredKeyList(); + if (keys.length > 0) yield* seed.stop(keys); + } // Drop queued frames so a reconnect isn't preceded by a stale backlog. yield* clearOutbound(); yield* Deferred.fail(closed, undefined); @@ -307,6 +337,7 @@ export const createVenueWS = ( } const resubs: TKey[] = []; + const reseeds: TKey[] = []; let applied = 0; for (const { key, frame, action = "write" } of parsed.frames) { if (!isDesired(key)) continue; @@ -316,12 +347,20 @@ export const createVenueWS = ( resubs.push(key); continue; } + if (action === "reseed") { + cache.deleteCached(key); + reseeds.push(key); + continue; + } cache.setPriceSync(key, frame); applied += 1; } if (resubs.length > 0) { Runtime.runSync(runtime, resubscribeKeys(resubs)); } + if (reseeds.length > 0 && seed !== undefined) { + Runtime.runSync(runtime, seed.restart(reseeds)); + } recordTickHandle(venue, name, performance.now() - started, applied); }; diff --git a/workspace/data-proxy/src/modules/ticker-module/README.md b/workspace/data-proxy/src/modules/ticker-module/README.md index c891a35..3652416 100644 --- a/workspace/data-proxy/src/modules/ticker-module/README.md +++ b/workspace/data-proxy/src/modules/ticker-module/README.md @@ -1,6 +1,6 @@ # Ticker module -Shared HTTP + cache + idle-cleanup loop for public ticker venues. Each module supplies only its WebSocket protocol and a `parseKey` that turns request tokens into cache keys; request handling, price caching, and subscription lifecycle are the same. +A ticker module is created with `createTickerModuleService`, which takes a `CreateTickerModuleServiceParams` object. It provides shared functionality for public ticker modules, such as subscription lifecycle handling, caching, and HTTP request handling. ## Overview @@ -45,7 +45,7 @@ Duration fields accept a number (ms) or a duration string (`"30 seconds"`). | Venue | Default `wsUrl` | Default `maxMessages` / `maxMessagesWindow` | Extra fields | | --- | --- | --- | --- | -| Binance | `wss://stream.binance.com:9443/stream` | `5` / `"1 second"` | `streamType` (default `"bookTicker"`): `bookTicker`, `trade`. No keepalive. `tradesKeepSeconds` (default `60`) is how many seconds of trades to keep when `streamType` is `"trade"`, using each trade's `T` (ms). | +| Binance | `wss://stream.binance.com:9443/stream` | `5` / `"1 second"` | `streamType` (default `"bookTicker"`): `bookTicker`, `trade`, `depth@100ms`, `depth@1000ms`. No keepalive. `tradesKeepSeconds` (default `60`) is how many seconds of trades to keep when `streamType` is `"trade"`, using each trade's `T` (ms). | | OKX | `wss://ws.okx.com:8443/ws/v5/public` | `480` / `"1 hour"` | `keepaliveInterval` (default `"20 seconds"`). | | Bybit | `wss://stream.bybit.com/v5/public/spot` | `5` / `"1 second"` | `keepaliveInterval` (default `"15 seconds"`). Subscribe and unsubscribe send one symbol per frame since Bybit rejects the whole args list when any symbol is invalid. | | Lighter | `wss://mainnet.zklighter.elliot.ai/stream?readonly=true` (The `readonly` query parameter avoids geo-restriction) | `180` / `"1 minute"` | `keepaliveInterval` (default `"60 seconds"`). `streamType` (default `"ticker"`): `ticker`, `order_book`, `trade`. Keys are numeric market ids. `tradesKeepSeconds` (default `60`) is how many seconds of executed trades to keep when `streamType` is `"trade"`. | @@ -182,7 +182,7 @@ Lighter (identity field `marketId`): | Field | Present when | Description | | --- | --- | --- | | Identity (`symbol`, `instId`, or `marketId`) | always | The raw request token from `fetchFromModule` (original casing). | -| Venue ticker fields | `__sedaHasPrice: true` | Relayed from the cached frame. | +| Venue ticker fields | `__sedaHasPrice: true` | Relayed from the cached data. | | `__sedaHasPrice` | always | `true` when a cached price was returned; `false` on wait timeout, miss, or unhealthy socket. | Requests with more symbols than `maxSymbolsPerRequest` return HTTP 400. diff --git a/workspace/data-proxy/src/modules/ticker-module/binance-depth.test.ts b/workspace/data-proxy/src/modules/ticker-module/binance-depth.test.ts new file mode 100644 index 0000000..8c210ee --- /dev/null +++ b/workspace/data-proxy/src/modules/ticker-module/binance-depth.test.ts @@ -0,0 +1,152 @@ +import { describe, expect, it } from "bun:test"; +import { + type BinanceDepthSnapshot, + type BinanceDepthUpdate, + type DepthSyncState, + onDepthUpdate, + tryPromotePendingSnapshot, +} from "./binance-depth"; + +const book = ( + lastUpdateId: number, + bids: Array<[string, string]> = [], + asks: Array<[string, string]> = [], +): BinanceDepthSnapshot => ({ s: "BTCUSDT", lastUpdateId, bids, asks }); + +const update = ( + U: number, + u: number, + b: Array<[string, string]> = [], + a: Array<[string, string]> = [], +): BinanceDepthUpdate => ({ U, u, b, a }); + +const emptyDepthSyncState = (): DepthSyncState => ({ + bufferedUpdates: [], + book: null, + pendingSnapshot: null, +}); + +describe("tryPromotePendingSnapshot", () => { + it("holds the snapshot until a diff covers the next update id", () => { + const state = emptyDepthSyncState(); + state.pendingSnapshot = book(100, [["1", "5"]], [["2", "5"]]); + expect(tryPromotePendingSnapshot(state)).toEqual({ + action: "ignore", + }); + expect(state.book).toBeNull(); + expect(state.pendingSnapshot?.lastUpdateId).toBe(100); + + expect( + onDepthUpdate( + state, + update(101, 102, [ + ["1", "0"], + ["3", "1"], + ]), + ), + ).toEqual({ + action: "write", + book: { + s: "BTCUSDT", + lastUpdateId: 102, + bids: [["3", "1"]], + asks: [["2", "5"]], + }, + }); + expect(state.pendingSnapshot).toBeNull(); + expect(state.bufferedUpdates).toEqual([]); + }); + + it("drops diffs already included in the snapshot and applies the rest", () => { + const state = emptyDepthSyncState(); + state.bufferedUpdates.push(update(90, 95, [["9", "9"]])); + state.bufferedUpdates.push(update(100, 104, [["1", "2"]])); + state.pendingSnapshot = book(100, [["1", "5"]], [["2", "5"]]); + + expect(tryPromotePendingSnapshot(state)).toEqual({ + action: "write", + book: { + s: "BTCUSDT", + lastUpdateId: 104, + bids: [["1", "2"]], + asks: [["2", "5"]], + }, + }); + }); + + it("reseeds when the snapshot is older than the first buffered diff", () => { + const state = emptyDepthSyncState(); + const stale = update(200, 201, [["4", "1"]]); + state.bufferedUpdates.push(stale); + state.pendingSnapshot = book(100); + + expect(tryPromotePendingSnapshot(state)).toEqual({ action: "reseed" }); + expect(state.bufferedUpdates).toEqual([stale]); + expect(state.pendingSnapshot).toBeNull(); + expect(state.book).toBeNull(); + }); + + it("treats a diff that starts at lastUpdateId + 1 as the bridge", () => { + const state = emptyDepthSyncState(); + state.bufferedUpdates.push(update(101, 101, [["1", "8"]])); + state.pendingSnapshot = book(100, [["1", "5"]]); + + expect(tryPromotePendingSnapshot(state)).toEqual({ + action: "write", + book: { s: "BTCUSDT", lastUpdateId: 101, bids: [["1", "8"]], asks: [] }, + }); + }); +}); + +describe("onDepthUpdate", () => { + it("ignores a diff the book already includes", () => { + const state = emptyDepthSyncState(); + state.book = book(105, [["1", "1"]]); + + expect(onDepthUpdate(state, update(100, 105, [["1", "9"]]))).toEqual({ + action: "ignore", + }); + expect(state.book).toEqual(book(105, [["1", "1"]])); + }); + + it("merges a continuing diff", () => { + const state = emptyDepthSyncState(); + state.book = book(105, [["1", "1"]], [["2", "2"]]); + + expect( + onDepthUpdate( + state, + update( + 106, + 108, + [], + [ + ["2", "9"], + ["4", "1"], + ], + ), + ), + ).toEqual({ + action: "write", + book: { + s: "BTCUSDT", + lastUpdateId: 108, + bids: [["1", "1"]], + asks: [ + ["2", "9"], + ["4", "1"], + ], + }, + }); + }); + + it("clears the book and keeps the gapped diff for the next snapshot", () => { + const state = emptyDepthSyncState(); + state.book = book(105, [["1", "1"]]); + const gap = update(110, 111, [["8", "1"]]); + + expect(onDepthUpdate(state, gap)).toEqual({ action: "reseed" }); + expect(state.book).toBeNull(); + expect(state.bufferedUpdates).toEqual([gap]); + }); +}); diff --git a/workspace/data-proxy/src/modules/ticker-module/binance-depth.ts b/workspace/data-proxy/src/modules/ticker-module/binance-depth.ts new file mode 100644 index 0000000..1657952 --- /dev/null +++ b/workspace/data-proxy/src/modules/ticker-module/binance-depth.ts @@ -0,0 +1,436 @@ +import { Duration, Effect, Fiber } from "effect"; +import type { PriceCache } from "../shared/price-cache"; +import type { ParsedTickerFrames } from "../shared/venue-ws"; +import type { BinancePriceFrame } from "./binance"; + +export const BINANCE_DEPTH_SNAPSHOT_URL = + "https://api.binance.com/api/v3/depth"; +export const BINANCE_DEPTH_SNAPSHOT_LIMIT = 5000; +const DEPTH_SNAPSHOT_RETRY = Duration.seconds(1); + +/** Pair [price, quantity] */ +export type BinanceDepthLevel = [string, string]; + +export type BinanceDepthSnapshot = { + s: string; + /** Indicates that update ID up to this value is fully synced. */ + lastUpdateId: number; + bids: BinanceDepthLevel[]; + asks: BinanceDepthLevel[]; +}; + +export interface BinanceDepthUpdate { + /** The first update ID in this event. */ + U: number; + /** The last update ID in this event. */ + u: number; + /** Bids. */ + b: BinanceDepthLevel[]; + /** Asks. */ + a: BinanceDepthLevel[]; +} + +export interface DepthSyncState { + /** Synchronized live order book. */ + book: BinanceDepthSnapshot | null; + /** Updates buffered while the book is not synced (book is null). */ + bufferedUpdates: BinanceDepthUpdate[]; + /** Snapshot held until it is promoted to book. */ + pendingSnapshot: BinanceDepthSnapshot | null; +} + +export type DepthStep = + | { action: "ignore" } + | { action: "write"; book: BinanceDepthSnapshot } + | { action: "reseed" }; + +const isUpdateId = (value: unknown): value is number => + typeof value === "number" && Number.isSafeInteger(value); + +const isZeroQty = (qty: string): boolean => Number(qty) === 0; + +const readLevel = (value: unknown): BinanceDepthLevel | null => { + if (!Array.isArray(value) || value.length < 2) return null; + if (typeof value[0] !== "string" || typeof value[1] !== "string") return null; + return [value[0], value[1]]; +}; + +const readSide = (value: unknown): BinanceDepthLevel[] | null => { + if (!Array.isArray(value)) return null; + const levels: BinanceDepthLevel[] = []; + for (const item of value) { + const level = readLevel(item); + if (level === null) return null; + levels.push(level); + } + return levels; +}; + +const copySide = (levels: BinanceDepthLevel[]): BinanceDepthLevel[] => + levels.map((level) => [level[0], level[1]]); + +const byPrice = + (direction: 1 | -1) => + (left: BinanceDepthLevel, right: BinanceDepthLevel): number => + (Number(left[0]) - Number(right[0])) * direction; + +const sortSides = ( + bids: BinanceDepthLevel[], + asks: BinanceDepthLevel[], +): Pick => ({ + bids: copySide(bids).sort(byPrice(-1)), + asks: copySide(asks).sort(byPrice(1)), +}); + +const sortBook = (book: BinanceDepthSnapshot): BinanceDepthSnapshot => { + const sides = sortSides(book.bids, book.asks); + return { + s: book.s, + lastUpdateId: book.lastUpdateId, + bids: sides.bids, + asks: sides.asks, + }; +}; + +/** Absolute quantity. `"0"` removes the price. New prices append. */ +const mergeSide = ( + existing: BinanceDepthLevel[], + updates: BinanceDepthLevel[], +): BinanceDepthLevel[] => { + const levels = copySide(existing); + for (const [price, qty] of updates) { + const index = levels.findIndex((level) => level[0] === price); + if (isZeroQty(qty)) { + if (index !== -1) levels.splice(index, 1); + continue; + } + if (index === -1) levels.push([price, qty]); + else levels[index] = [price, qty]; + } + return levels; +}; + +const mergeBook = ( + book: BinanceDepthSnapshot, + event: BinanceDepthUpdate, +): BinanceDepthSnapshot => + sortBook({ + s: book.s, + lastUpdateId: event.u, + bids: mergeSide(book.bids, event.b), + asks: mergeSide(book.asks, event.a), + }); + +/** + * An update continues the book when its range covers `lastUpdateId + 1`. + * `u <= lastUpdateId` is already included in the snapshot. + */ +const continues = (lastUpdateId: number, event: BinanceDepthUpdate): boolean => + event.u > lastUpdateId && event.U <= lastUpdateId + 1; + +/** Promotes `pendingSnapshot` into `book` when there is an applicable update. */ +export const tryPromotePendingSnapshot = (state: DepthSyncState): DepthStep => { + if (state.pendingSnapshot === null) return { action: "ignore" }; + + // Find the updates that can be applied to the snapshot. + // Binance Spot sends updates in order, and they are stored in arrival order. + let startIndex = 0; + for (const update of state.bufferedUpdates) { + if (update.u > state.pendingSnapshot.lastUpdateId) break; + startIndex++; + } + const applicableUpdates = state.bufferedUpdates.slice(startIndex); + + // If all buffered events are stale, drop them and wait for new updates. + if (applicableUpdates.length === 0) { + state.bufferedUpdates = []; + return { action: "ignore" }; + } + + // Apply the buffered updates to the pending snapshot and promote to book. + let next = state.pendingSnapshot; + for (let index = 0; index < applicableUpdates.length; index++) { + const event = applicableUpdates[index]; + if (event.u <= next.lastUpdateId) continue; + if (!continues(next.lastUpdateId, event)) { + state.bufferedUpdates = applicableUpdates.slice(index); + state.book = null; + state.pendingSnapshot = null; + return { action: "reseed" }; + } + + next = mergeBook(next, event); + } + state.bufferedUpdates = []; + state.pendingSnapshot = null; + state.book = next; + return { action: "write", book: next }; +}; + +export const onDepthUpdate = ( + state: DepthSyncState, + event: BinanceDepthUpdate, +): DepthStep => { + // No live book yet. + if (state.book === null) { + state.bufferedUpdates.push(event); + if (state.pendingSnapshot === null) return { action: "ignore" }; + return tryPromotePendingSnapshot(state); + } + + // The update event brings nothing new. + if (event.u <= state.book.lastUpdateId) return { action: "ignore" }; + + // A gap. Clear the book and reseed. + if (!continues(state.book.lastUpdateId, event)) { + state.bufferedUpdates = [event]; + state.book = null; + state.pendingSnapshot = null; + return { action: "reseed" }; + } + + // The update event continues the book. + state.book = mergeBook(state.book, event); + return { action: "write", book: state.book }; +}; + +export const readDepthBook = ( + symbol: string, + value: unknown, +): BinanceDepthSnapshot | null => { + if (typeof value !== "object" || value === null || Array.isArray(value)) { + return null; + } + const record = value as Record; + if (!isUpdateId(record.lastUpdateId)) return null; + + const bids = readSide(record.bids); + const asks = readSide(record.asks); + if (bids === null || asks === null) return null; + + const sides = sortSides( + bids.filter((level) => !isZeroQty(level[1])), + asks.filter((level) => !isZeroQty(level[1])), + ); + return { + s: symbol.toUpperCase(), + lastUpdateId: record.lastUpdateId, + bids: sides.bids, + asks: sides.asks, + }; +}; + +export const readDepthUpdate = ( + payload: Record, + symbol: string, +): { key: string; event: BinanceDepthUpdate } | "invalid" | null => { + if (payload.e !== "depthUpdate") return null; + if (!isUpdateId(payload.U) || !isUpdateId(payload.u)) return "invalid"; + + const bids = readSide(payload.b); + const asks = readSide(payload.a); + if (bids === null || asks === null) return "invalid"; + return { + key: symbol.toUpperCase(), + event: { U: payload.U, u: payload.u, b: bids, a: asks }, + }; +}; + +const emptyBook = (symbol: string): BinanceDepthSnapshot => ({ + s: symbol.toUpperCase(), + lastUpdateId: 0, + bids: [], + asks: [], +}); + +interface KeyState extends DepthSyncState { + generation: number; +} + +export interface DepthSync { + start: (keys: string[]) => Effect.Effect; + stop: (keys: string[]) => Effect.Effect; + restart: (keys: string[]) => Effect.Effect; + ingest: ( + payload: Record, + symbol: string, + ) => ParsedTickerFrames | null; +} + +export const fetchBinanceDepthSnapshot = (symbol: string) => + Effect.async((resume) => { + const controller = new AbortController(); + const url = new URL(BINANCE_DEPTH_SNAPSHOT_URL); + url.searchParams.set("symbol", symbol); + url.searchParams.set("limit", String(BINANCE_DEPTH_SNAPSHOT_LIMIT)); + fetch(url, { signal: controller.signal }) + .then(async (response) => { + if (!response.ok) { + resume( + Effect.fail( + new Error(`Binance depth snapshot failed: ${response.status}`), + ), + ); + return; + } + const body: unknown = await response.json(); + const book = readDepthBook(symbol, body); + if (book === null) { + resume(Effect.fail(new Error("Binance depth snapshot was invalid"))); + return; + } + resume(Effect.succeed(book)); + }) + .catch((error: unknown) => { + if (controller.signal.aborted) return; + resume( + Effect.fail( + error instanceof Error ? error : new Error(String(error)), + ), + ); + }); + return Effect.sync(() => { + controller.abort(); + }); + }); + +export const createDepthSync = ( + venueCache: Pick< + PriceCache, + "setPriceSync" | "deleteCached" + >, +): DepthSync => { + const states = new Map(); + const fibers = new Map>(); + + /** Fetch a seed snapshot until it is promoted to the live book. */ + const snapshotLoop = (key: string, generation: number) => + Effect.gen(function* () { + while (true) { + const state = states.get(key); + if (state === undefined || state.generation !== generation) return; + + const snapshot = yield* fetchBinanceDepthSnapshot(key).pipe( + Effect.catchAll((error) => + Effect.logWarning("Binance depth snapshot failed", { + symbol: key, + error: String(error), + }).pipe(Effect.as(null)), + ), + ); + + // Check the state again after the async fetch. + const current = states.get(key); + if ( + current === undefined || + current !== state || + current.generation !== generation + ) { + return; + } + if (snapshot === null) { + yield* Effect.sleep(DEPTH_SNAPSHOT_RETRY); + continue; + } + + current.pendingSnapshot = snapshot; + const step = tryPromotePendingSnapshot(current); + + if (step.action === "write") { + venueCache.setPriceSync(key, step.book); + return; + } + if (step.action === "ignore") return; + + yield* Effect.sleep(DEPTH_SNAPSHOT_RETRY); + } + }); + + const forkLoop = (key: string, generation: number) => + Effect.gen(function* () { + const prevFiber = fibers.get(key); + const newFiber = yield* Effect.forkDaemon(snapshotLoop(key, generation)); + fibers.set(key, newFiber); + if (prevFiber !== undefined) { + yield* Fiber.interrupt(prevFiber); + } + }); + + const start = (keys: string[]) => + Effect.gen(function* () { + for (const key of keys) { + const generation = (states.get(key)?.generation ?? 0) + 1; + states.set(key, { + generation, + bufferedUpdates: [], + book: null, + pendingSnapshot: null, + }); + venueCache.deleteCached(key); + yield* forkLoop(key, generation); + } + }); + + const stop = (keys: string[]) => + Effect.gen(function* () { + for (const key of keys) { + const fiber = fibers.get(key); + fibers.delete(key); + states.delete(key); + venueCache.deleteCached(key); + if (fiber !== undefined) yield* Fiber.interrupt(fiber); + } + }); + + const restart = (keys: string[]) => + Effect.gen(function* () { + for (const key of keys) { + const previous = states.get(key); + const generation = (previous?.generation ?? 0) + 1; + states.set(key, { + generation, + bufferedUpdates: previous?.bufferedUpdates.slice() ?? [], + book: null, + pendingSnapshot: null, + }); + + venueCache.deleteCached(key); + yield* Effect.logWarning("Binance depth book out of sync", { + symbol: key, + }); + yield* forkLoop(key, generation); + } + }); + + const ingest = ( + payload: Record, + symbol: string, + ): ParsedTickerFrames | null => { + const update = readDepthUpdate(payload, symbol); + if (update === null) return null; + if (update === "invalid") { + const key = symbol.toUpperCase(); + const state = states.get(key); + if (state === undefined) return null; + + state.bufferedUpdates = []; + state.book = null; + state.pendingSnapshot = null; + return [{ key, action: "reseed", frame: emptyBook(symbol) }]; + } + + const state = states.get(update.key); + if (state === undefined) return null; + + const step = onDepthUpdate(state, update.event); + if (step.action === "write") { + return [{ key: update.key, action: "write", frame: step.book }]; + } + if (step.action === "reseed") { + return [{ key: update.key, action: "reseed", frame: emptyBook(symbol) }]; + } + return [{ key: update.key, action: "ignore", frame: emptyBook(symbol) }]; + }; + + return { start, stop, restart, ingest }; +}; diff --git a/workspace/data-proxy/src/modules/ticker-module/binance-ws-client.test.ts b/workspace/data-proxy/src/modules/ticker-module/binance-ws-client.test.ts index 630a767..405076a 100644 --- a/workspace/data-proxy/src/modules/ticker-module/binance-ws-client.test.ts +++ b/workspace/data-proxy/src/modules/ticker-module/binance-ws-client.test.ts @@ -15,6 +15,12 @@ import { parseInboundFrame, } from "./binance"; +type DepthRestSnapshot = { + lastUpdateId: number; + bids: Array<[string, string]>; + asks: Array<[string, string]>; +}; + const btcBook: BinancePriceFrame = { u: 400900217, s: "BTCUSDT", @@ -36,7 +42,13 @@ const ethBook: BinancePriceFrame = { describe("buildStreamName", () => { it("lowercases the symbol and appends the stream type", () => { expect(buildStreamName("BTCUSDT", "bookTicker")).toBe("btcusdt@bookTicker"); - expect(buildStreamName("ethusdt", "aggTrade")).toBe("ethusdt@aggTrade"); + expect(buildStreamName("ethusdt", "trade")).toBe("ethusdt@trade"); + expect(buildStreamName("BTCUSDT", "depth@100ms")).toBe( + "btcusdt@depth@100ms", + ); + expect(buildStreamName("BTCUSDT", "depth@1000ms")).toBe( + "btcusdt@depth@1000ms", + ); }); }); @@ -65,21 +77,24 @@ describe("buildSubscribeFrame / buildUnsubscribeFrame", () => { describe("parseInboundFrame", () => { it("unwraps a combined-stream envelope", () => { const raw = JSON.stringify({ stream: "btcusdt@bookTicker", data: btcBook }); - expect(parseInboundFrame(raw)).toEqual({ + expect(parseInboundFrame(raw, "bookTicker")).toEqual({ kind: "tickers", frames: [{ key: "BTCUSDT", frame: btcBook }], }); }); it("accepts a bare (raw-stream) payload", () => { - expect(parseInboundFrame(JSON.stringify(ethBook))).toEqual({ + expect(parseInboundFrame(JSON.stringify(ethBook), "bookTicker")).toEqual({ kind: "tickers", frames: [{ key: "ETHUSDT", frame: ethBook }], }); }); it("uppercases the symbol so cache keys stay consistent", () => { - const result = parseInboundFrame(JSON.stringify({ s: "btcusdt", b: "1" })); + const result = parseInboundFrame( + JSON.stringify({ s: "btcusdt", b: "1" }), + "bookTicker", + ); expect(result).toMatchObject({ kind: "tickers", frames: [{ key: "BTCUSDT" }], @@ -88,7 +103,7 @@ describe("parseInboundFrame", () => { it("returns null for a control ack with no symbol", () => { expect( - parseInboundFrame(JSON.stringify({ result: null, id: 1 })), + parseInboundFrame(JSON.stringify({ result: null, id: 1 }), "bookTicker"), ).toBeNull(); }); @@ -101,6 +116,7 @@ describe("parseInboundFrame", () => { msg: "Invalid JSON: key must be a string at line 1 column 2", }, }), + "bookTicker", ), ).toEqual({ kind: "error", @@ -110,11 +126,13 @@ describe("parseInboundFrame", () => { }); it("returns null for malformed JSON", () => { - expect(parseInboundFrame("not json")).toBeNull(); + expect(parseInboundFrame("not json", "bookTicker")).toBeNull(); }); it("returns null when the payload has no string symbol", () => { - expect(parseInboundFrame(JSON.stringify({ b: "1", a: "2" }))).toBeNull(); + expect( + parseInboundFrame(JSON.stringify({ b: "1", a: "2" }), "bookTicker"), + ).toBeNull(); }); }); @@ -294,3 +312,216 @@ describe("createBinanceWS", () => { await Effect.runPromise(Fiber.interrupt(fiber)); }); }); + +const depthUpdate = ( + symbol: string, + U: number, + u: number, + b: Array<[string, string]> = [], + a: Array<[string, string]> = [], +) => + JSON.stringify({ + stream: `${symbol.toLowerCase()}@depth@100ms`, + data: { e: "depthUpdate", s: symbol, U, u, b, a }, + }); + +const createSnapshotFeed = () => { + const pending: Array<{ + resolve: (book: DepthRestSnapshot) => void; + reject: (error: Error) => void; + }> = []; + const queued: DepthRestSnapshot[] = []; + let calls = 0; + const original = globalThis.fetch; + globalThis.fetch = (async (_input, init) => { + calls += 1; + const signal = init?.signal ?? undefined; + const book = await new Promise((resolve, reject) => { + const abort = () => { + reject(new DOMException("The operation was aborted", "AbortError")); + }; + if (signal?.aborted) { + abort(); + return; + } + const next = queued.shift(); + if (next !== undefined) { + resolve(next); + return; + } + const waiter = { resolve, reject }; + pending.push(waiter); + signal?.addEventListener( + "abort", + () => { + const index = pending.indexOf(waiter); + if (index !== -1) pending.splice(index, 1); + abort(); + }, + { once: true }, + ); + }); + return new Response(JSON.stringify(book), { + status: 200, + headers: { "Content-Type": "application/json" }, + }); + }) as typeof fetch; + + const deliver = (book: DepthRestSnapshot) => { + const waiter = pending.shift(); + if (waiter !== undefined) waiter.resolve(book); + else queued.push(book); + }; + return { + deliver, + calls: () => calls, + restore: () => { + globalThis.fetch = original; + }, + }; +}; + +describe("createBinanceWS depth", () => { + const depthConfig: BinanceModuleConfig = { + ...baseConfig, + streamType: "depth@100ms", + subscriptionSymbols: ["BTCUSDT"], + }; + let restoreFetch: (() => void) | undefined; + + afterEach(() => { + restoreFetch?.(); + restoreFetch = undefined; + }); + + const startDepth = ( + feed: ReturnType, + preSubscribed: string[] = ["BTCUSDT"], + ) => + Effect.gen(function* () { + restoreFetch = feed.restore; + const cache = yield* createPriceCache(); + const ws = yield* createBinanceWS( + depthConfig, + cache, + Schedule.spaced(Duration.minutes(10)), + ); + yield* ws.subscribe(preSubscribed); + const fiber = yield* ws.start(); + return { cache, ws, fiber }; + }).pipe(Logger.withMinimumLogLevel(LogLevel.None)); + + it("subscribes to the depth stream and waits to publish until a diff bridges the snapshot", async () => { + const feed = createSnapshotFeed(); + const { cache, fiber } = await Effect.runPromise(startDepth(feed)); + await flush(); + const ws = FakeWebSocket.instances[0]; + ws.triggerOpen(); + await flush(); + + expect(parseControl(ws.sent[0]).params).toEqual(["btcusdt@depth@100ms"]); + expect(feed.calls()).toBe(1); + expect(cache.size()).toBe(0); + + ws.triggerMessage( + depthUpdate("BTCUSDT", 101, 102, [ + ["1", "0"], + ["3", "1"], + ]), + ); + await flush(); + expect(cache.size()).toBe(0); + + feed.deliver({ + lastUpdateId: 100, + bids: [ + ["1", "5"], + ["3", "1"], + ], + asks: [["2", "5"]], + }); + await flush(); + + expect(await Effect.runPromise(cache.getOrWaitPrice("BTCUSDT"))).toEqual({ + s: "BTCUSDT", + lastUpdateId: 102, + bids: [["3", "1"]], + asks: [["2", "5"]], + }); + + ws.triggerMessage(depthUpdate("BTCUSDT", 103, 103, [], [["2", "9"]])); + await flush(); + expect(feed.calls()).toBe(1); + expect(await Effect.runPromise(cache.getOrWaitPrice("BTCUSDT"))).toEqual({ + s: "BTCUSDT", + lastUpdateId: 103, + bids: [["3", "1"]], + asks: [["2", "9"]], + }); + + await Effect.runPromise(Fiber.interrupt(fiber)); + }); + + it("refetches the snapshot after a sequence gap and keeps the gapped diff", async () => { + const feed = createSnapshotFeed(); + const { cache, fiber } = await Effect.runPromise(startDepth(feed)); + await flush(); + FakeWebSocket.instances[0].triggerOpen(); + await flush(); + + feed.deliver({ lastUpdateId: 100, bids: [["1", "5"]], asks: [] }); + await flush(); + FakeWebSocket.instances[0].triggerMessage( + depthUpdate("BTCUSDT", 101, 101, [["1", "8"]]), + ); + await flush(); + expect(cache.getCached("BTCUSDT")).toMatchObject({ lastUpdateId: 101 }); + + FakeWebSocket.instances[0].triggerMessage( + depthUpdate("BTCUSDT", 110, 111, [["8", "1"]]), + ); + await flush(); + expect(cache.getCached("BTCUSDT")).toBeUndefined(); + expect(feed.calls()).toBe(2); + + feed.deliver({ lastUpdateId: 109, bids: [["1", "8"]], asks: [] }); + await flush(); + expect(await Effect.runPromise(cache.getOrWaitPrice("BTCUSDT"))).toEqual({ + s: "BTCUSDT", + lastUpdateId: 111, + bids: [ + ["8", "1"], + ["1", "8"], + ], + asks: [], + }); + + await Effect.runPromise(Fiber.interrupt(fiber)); + }); + + it("drops a snapshot that arrives after unsubscribe", async () => { + const feed = createSnapshotFeed(); + const { + cache, + ws: service, + fiber, + } = await Effect.runPromise(startDepth(feed)); + await flush(); + FakeWebSocket.instances[0].triggerOpen(); + await flush(); + expect(feed.calls()).toBe(1); + + await Effect.runPromise(service.unsubscribe(["BTCUSDT"])); + await flush(); + feed.deliver({ lastUpdateId: 100, bids: [["1", "5"]], asks: [] }); + await flush(); + FakeWebSocket.instances[0].triggerMessage( + depthUpdate("BTCUSDT", 101, 101, [["1", "8"]]), + ); + await flush(); + + expect(cache.size()).toBe(0); + + await Effect.runPromise(Fiber.interrupt(fiber)); + }); +}); diff --git a/workspace/data-proxy/src/modules/ticker-module/binance.ts b/workspace/data-proxy/src/modules/ticker-module/binance.ts index c4646b5..016fc7b 100644 --- a/workspace/data-proxy/src/modules/ticker-module/binance.ts +++ b/workspace/data-proxy/src/modules/ticker-module/binance.ts @@ -1,4 +1,4 @@ -import type { Effect, Schedule } from "effect"; +import { type Effect, Match, type Schedule } from "effect"; import type { BinanceModuleConfig, BinanceStreamType, @@ -6,10 +6,12 @@ import type { import { isRecord, parseJsonRecord } from "../shared/json"; import type { PriceCache } from "../shared/price-cache"; import { + type ParsedTickerFrames, type VenueParsedInbound, type VenueWS, createVenueWS, } from "../shared/venue-ws"; +import { type DepthSync, createDepthSync } from "./binance-depth"; import { type CreateTickerModuleServiceParams, createTickerModuleService, @@ -117,41 +119,81 @@ const applyBinanceTradeFrame = ( return { s: symbol ?? "", trades }; }; +const readErrorFrame = ( + json: Record, +): { + kind: "error"; + code: string | number | null; + message: string | null; +} | null => { + const err = json.error; + if (!isRecord(err)) return null; + return { + kind: "error", + code: typeof err.code === "number" ? err.code : null, + message: typeof err.msg === "string" ? err.msg : null, + }; +}; + +const unwrapPayload = ( + json: Record, +): Record => + typeof json.stream === "string" && isRecord(json.data) ? json.data : json; + +const readSymbolFrame = ( + payload: Record, + symbol: string, +): ParsedTickerFrames => [ + { + key: symbol.toUpperCase(), + frame: payload as BinancePriceFrame, + }, +]; + +const readTradeFrame = ( + payload: Record, + symbol: string, +): ParsedTickerFrames => [ + { + key: symbol.toUpperCase(), + frame: { s: symbol, trades: [payload] }, + }, +]; + +const readDepthFrame = ( + payload: Record, + symbol: string, + depth: DepthSync | undefined, +): ParsedTickerFrames | null => { + if (depth === undefined) return null; + return depth.ingest(payload, symbol); +}; + export const parseInboundFrame = ( raw: string, streamType: BinanceStreamType, + depth?: DepthSync, ): VenueParsedInbound | null => { const json = parseJsonRecord(raw); if (!json) return null; - const err = json.error; - if (isRecord(err)) { - return { - kind: "error", - code: typeof err.code === "number" ? err.code : null, - message: typeof err.msg === "string" ? err.msg : null, - }; - } - - // Combined streams wrap the payload as { stream, data }; raw streams send it bare. - let payload: unknown = json; - if (typeof json.stream === "string" && isRecord(json.data)) { - payload = json.data; - } + const errorFrame = readErrorFrame(json); + if (errorFrame) return errorFrame; - if (!isRecord(payload) || typeof payload.s !== "string") { - return null; - } + const payload = unwrapPayload(json); + if (typeof payload.s !== "string" || payload.s.length === 0) return null; - const frame = - streamType === "trade" - ? { s: payload.s, trades: [payload] } - : (payload as BinancePriceFrame); + const symbol = payload.s; + const frames = Match.value(streamType).pipe( + Match.when("bookTicker", () => readSymbolFrame(payload, symbol)), + Match.when("trade", () => readTradeFrame(payload, symbol)), + Match.when("depth@100ms", () => readDepthFrame(payload, symbol, depth)), + Match.when("depth@1000ms", () => readDepthFrame(payload, symbol, depth)), + Match.exhaustive, + ); + if (frames === null) return null; - return { - kind: "tickers", - frames: [{ key: payload.s.toUpperCase(), frame }], - }; + return { kind: "tickers", frames }; }; export const createBinanceWS = ( @@ -164,15 +206,22 @@ export const createBinanceWS = ( const streamNamesFor = (symbols: string[]) => symbols.map((symbol) => buildStreamName(symbol, config.streamType)); + const depth = + config.streamType.split("@")[0] === "depth" + ? createDepthSync(cache) + : undefined; + return createVenueWS({ venue: "binance", config, cache, reconnectSchedule, + seed: depth, buildSubscribeFrame: (keys) => buildSubscribeFrame(streamNamesFor(keys), nextControlId()), buildUnsubscribeFrame: (keys) => buildUnsubscribeFrame(streamNamesFor(keys), nextControlId()), - parseInboundFrame: (raw) => parseInboundFrame(raw, config.streamType), + parseInboundFrame: (raw) => + parseInboundFrame(raw, config.streamType, depth), }); };