Binance Spot order book support

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];
 

workspace/data-proxy/src/modules/shared/venue-ws.test.ts

@@ -28,6 +28,19 @@ const startClient = (options?: {
 	maxMessagesWindow?: Duration.Duration;
 	reconnectSchedule?: Schedule.Schedule<unknown, unknown, never>;
 	preSubscribed?: string[];
+	seed?: {
+		start: (keys: string[]) => Effect.Effect<void, never, never>;
+		stop: (keys: string[]) => Effect.Effect<void, never, never>;
+		restart: (keys: string[]) => Effect.Effect<void, never, never>;
+	};
+	parseInboundFrame?: (raw: string) => {
+		kind: "tickers";
+		frames: Array<{
+			key: string;
+			frame: Record<string, unknown>;
+			action?: "write" | "ignore" | "resubscribe" | "reseed";
+		}>;
+	} | null;
 }) =>
 	Effect.gen(function* () {
 		const cache = yield* createPriceCache<string, Record<string, unknown>>();
@@ -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<string, unknown>;
-				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<string, unknown>;
+					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<void, never, never> =>
+				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<void>((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();

workspace/data-proxy/src/modules/shared/venue-ws.ts

@@ -44,14 +44,16 @@ export interface VenueWS<TKey = string> {
 	hasError(): Effect.Effect<boolean, never, never>;
 }
 
-export type InboundFrameAction = "write" | "ignore" | "resubscribe";
+export type InboundFrameAction = "write" | "ignore" | "resubscribe" | "reseed";
 
 export type ParsedTickersFrame<TKey, TFrame> = {
 	key: TKey;
 	frame: TFrame;
 	/**
 	 * 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<TKey, TFrame> {
 	buildSubscribeFrame: (keys: TKey[]) => string | string[];
 	buildUnsubscribeFrame: (keys: TKey[]) => string | string[];
 	parseInboundFrame: (raw: string) => VenueParsedInbound<TKey, TFrame> | 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<void, never, never>;
+		stop: (keys: TKey[]) => Effect.Effect<void, never, never>;
+		restart: (keys: TKey[]) => Effect.Effect<void, never, never>;
+	};
 }
 
 export const createVenueWS = <TKey, TFrame>(
@@ -101,6 +114,7 @@ export const createVenueWS = <TKey, TFrame>(
 			buildSubscribeFrame,
 			buildUnsubscribeFrame,
 			parseInboundFrame,
+			seed,
 		} = params;
 		const { name, wsUrl, maxMessages, maxMessagesWindow } = config;
 		const reconnectSchedule =
@@ -118,6 +132,9 @@ export const createVenueWS = <TKey, TFrame>(
 		const desiredKeys = MutableHashMap.empty<TKey, true>();
 		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<OutboundMessage>();
 		const activeSubscriptions = Metric.gauge(`${venue}_active_subscriptions`, {
 			description: `Number of active ${venue} WebSocket stream subscriptions`,
@@ -152,6 +169,12 @@ export const createVenueWS = <TKey, TFrame>(
 		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 = <TKey, TFrame>(
 				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 = <TKey, TFrame>(
 				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 = <TKey, TFrame>(
 				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<void, void>) =>
 			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 = <TKey, TFrame>(
 			}
 
 			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 = <TKey, TFrame>(
 					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);
 		};
 

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.