diff --git a/workspace/data-proxy/src/config/hydromancer-module-config.ts b/workspace/data-proxy/src/config/hydromancer-module-config.ts index 8b8d5bc..a5733aa 100644 --- a/workspace/data-proxy/src/config/hydromancer-module-config.ts +++ b/workspace/data-proxy/src/config/hydromancer-module-config.ts @@ -95,6 +95,36 @@ export const HydromancerModuleConfigSchema = v.strictObject({ ), ), ), + tradesSubscriptionCoins: v.optional(v.array(v.string()), []), + tradesMaxCoinsPerRequest: v.optional(v.number(), 20), + tradesKeepSeconds: v.optional(v.number(), 60), + tradesWaitTimeout: v.pipe( + v.optional(v.union([v.number(), v.string()]), "1 second"), + v.transform((value) => + Option.getOrThrowWith( + Duration.decodeUnknown(value), + () => new Error("Invalid tradesWaitTimeout duration"), + ), + ), + ), + tradesCleanupTtl: v.pipe( + v.optional(v.union([v.number(), v.string()]), "2 minutes"), + v.transform((value) => + Option.getOrThrowWith( + Duration.decodeUnknown(value), + () => new Error("Invalid tradesCleanupTtl duration"), + ), + ), + ), + tradesCleanupInterval: v.pipe( + v.optional(v.union([v.number(), v.string()]), "30 seconds"), + v.transform((value) => + Option.getOrThrowWith( + Duration.decodeUnknown(value), + () => new Error("Invalid tradesCleanupInterval duration"), + ), + ), + ), }); export interface HydromancerModuleConfig @@ -168,11 +198,31 @@ export const L2BookRequestBodySchema = v.object({ coins: v.union([v.string(), v.array(v.string())]), }); -// Request body the module accepts, discriminated on `type`. Any shape that -// matches neither variant is rejected with 400. +export const TradeSchema = v.object({ + coin: v.string(), + side: v.union([v.literal("B"), v.literal("A")]), + px: v.string(), + sz: v.string(), + hash: v.string(), + time: v.number(), + tid: v.number(), + users: v.tuple([v.string(), v.string()]), +}); + +export type Trade = v.InferOutput; + +export const TradesRequestBodySchema = v.object({ + type: v.literal("trades"), + // Array of tickers, or a single comma-delimited string (e.g. "BTC,ETH"). + coins: v.union([v.string(), v.array(v.string())]), +}); + +// Request body the module accepts, discriminated on `type`. Any other shape +// is forwarded to upstream REST. export const HydromancerRequestBodySchema = v.variant("type", [ AssetContextRequestBodySchema, L2BookRequestBodySchema, + TradesRequestBodySchema, ]); export type ParsedHydromancerBody = v.InferOutput< diff --git a/workspace/data-proxy/src/modules/hydromancer/README.md b/workspace/data-proxy/src/modules/hydromancer/README.md index 60623db..9ebb7fe 100644 --- a/workspace/data-proxy/src/modules/hydromancer/README.md +++ b/workspace/data-proxy/src/modules/hydromancer/README.md @@ -1,15 +1,15 @@ # Hydromancer module -Caches Hydromancer `activeAssetCtx` and `l2Book` updates over WebSocket and serves `assetContext` / `l2Book` requests from those caches. `assetContext` falls back to a REST endpoint when data is missing or stale. `l2Book` has no REST fallback: the request waits briefly for a snapshot and returns `null` on timeout. Bodies that match neither type are forwarded to upstream REST as-is. +Caches Hydromancer `activeAssetCtx`, `l2Book`, or `trades` updates over WebSocket and serves requests from those caches. `assetContext` falls back to a REST endpoint when data is missing or stale. `l2Book` and `trades` have no REST fallback: the request waits briefly for the first update and returns `null` on timeout. Bodies that match none of these types are forwarded to upstream REST as-is. ## Overview On startup the module: 1. Connects to `wsUrl` with the API key as a `token` query parameter. -2. Subscribes to every coin in `subscriptionCoins` (`activeAssetCtx`) and `l2BookSubscriptionCoins` (`l2Book`). -3. Caches inbound `activeAssetCtx` frames by coin (freshness-keyed) and inbound `l2Book` snapshots by coin (waiter-keyed). -4. Idle-unsubscribes each channel independently: `activeAssetCtx` coins after `coinsCleanupTtl`, `l2Book` coins after `l2BookCleanupTtl` (demand-driven subscriptions from HTTP requests are cleaned up the same way). +2. Subscribes to every coin in `subscriptionCoins` (`activeAssetCtx`), `l2BookSubscriptionCoins` (`l2Book`), and `tradesSubscriptionCoins` (`trades`). +3. Caches inbound frames. The cache for `activeAssetCtx` frames is freshness-keyed, whereas the caches for `l2Book` and `trades` frames are waiter-keyed. +4. Idle-unsubscribes each channel independently based on `coinsCleanupTtl`, `l2BookCleanupTtl`, and `tradesCleanupTtl`. For `assetContext` HTTP requests the handler: @@ -29,6 +29,15 @@ For `l2Book` HTTP requests the handler: There is no REST fallback for `l2Book`. Requests with more coins than `l2BookMaxCoinsPerRequest` return HTTP 400. +For `trades` HTTP requests the handler: + +1. Parses a batch (`coins`) body. There is no single-coin `coin` field. +2. Expands comma-separated values inside `coins` into individual tickers (same path-param expansion as batch `assetContext`). +3. Subscribes to each coin over WebSocket (idempotent). Re-subscribing adds coins to the connection's single trades subscription. +4. Returns the cached recent trades for that coin immediately when present; otherwise waits up to `tradesWaitTimeout` for the next inbound frame. A timeout or unsubscribe during the wait yields `null` for that coin. Trades executed outside `tradesKeepSeconds` are dropped on cache write or when the request is served. + +There is no REST fallback for `trades`, and the all-coins firehose is not used. Requests with more coins than `tradesMaxCoinsPerRequest` return HTTP 400. + ## Environment variables Set the env var named by `hydromancerApiKeyEnvKey`. Config parsing fails if it is unset; the value is treated as a secret and redacted from logs. @@ -59,6 +68,12 @@ Set the env var named by `hydromancerApiKeyEnvKey`. Config parsing fails if it i | `l2BookWaitTimeout` | no | `"1 second"` | How long an `l2Book` request waits for a snapshot before returning `null`. | | `l2BookCleanupTtl` | no | `"2 minutes"` | Idle time before an unused `l2Book` subscription is cleaned up. | | `l2BookCleanupInterval` | no | `"30 seconds"` | How often `l2Book` idle cleanup runs. | +| `tradesSubscriptionCoins` | no | `[]` | Coins to subscribe to `trades` on start. | +| `tradesMaxCoinsPerRequest` | no | `20` | Max coins allowed in a single `trades` request. | +| `tradesKeepSeconds` | no | `60` | How many seconds of executed trades to keep per coin. Older trades are dropped. | +| `tradesWaitTimeout` | no | `"1 second"` | How long a `trades` request waits for the first trade before returning `null`. | +| `tradesCleanupTtl` | no | `"2 minutes"` | Idle time before an unused `trades` subscription is cleaned up. | +| `tradesCleanupInterval` | no | `"30 seconds"` | How often `trades` idle cleanup runs. | | `reconnectMaxBackoff` | no | `"30 seconds"` | Cap on WS reconnect backoff. | | `reconnectStableThreshold` | no | `"30 seconds"` | Connected duration before reconnect backoff resets. | | `restFetchTimeout` | no | `"15 seconds"` | Timeout for REST `/info` calls. | @@ -86,7 +101,8 @@ Hydromancer routes do not use `fetchFromModule`. The request body is the Hydroma "restBaseUrl": "https://api.hydromancer.xyz", "hydromancerApiKeyEnvKey": "HYDROMANCER_API_KEY_MAINNET", "subscriptionCoins": ["BTC", "ETH"], - "l2BookSubscriptionCoins": ["BTC", "ETH"] + "l2BookSubscriptionCoins": ["BTC", "ETH"], + "tradesSubscriptionCoins": ["BTC", "ETH"] } ], "routes": [ @@ -120,6 +136,11 @@ curl -s "http://127.0.0.1:5384/proxy/hydro" \ curl -s "http://127.0.0.1:5384/proxy/hydro" \ -H 'Content-Type: application/json' \ -d '{"type":"l2Book","coins":["BTC","ETH"]}' | jq . + +# trades — response is a map of coin → Trade[] | null +curl -s "http://127.0.0.1:5384/proxy/hydro" \ + -H 'Content-Type: application/json' \ + -d '{"type":"trades","coins":["BTC","ETH"]}' | jq . ``` ## Request body @@ -130,6 +151,7 @@ curl -s "http://127.0.0.1:5384/proxy/hydro" \ | `{"type":"assetContext","coins":["...", "..."]}` | Batch path. Response is `{ [coin]: AssetCtx \| null }`. Comma-separated entries inside the array are expanded. | | `{"type":"assetContext","coins":"BTC,ETH"}` | Same batch path; a comma-delimited string is expanded to individual tickers. | | `{"type":"l2Book","coins":["...", "..."]}` | Batch path. Response is `{ [coin]: BookSnapshot \| null }`. Comma-separated entries inside the array are expanded. There is no REST fallback; a miss waits up to `l2BookWaitTimeout` then returns `null`. | +| `{"type":"trades","coins":["...", "..."]}` | Batch path. Response is `{ [coin]: Trade[] \| null }`. Comma-separated entries inside the array are expanded. There is no REST fallback; a miss waits up to `tradesWaitTimeout` then returns `null`. | | Anything else | Forwarded unchanged to `POST {restBaseUrl}/info` with the bearer token. | ## Response shape @@ -176,3 +198,25 @@ Unresolved coins stay `null`, matching Hydromancer’s native `/info` batch shap ``` `levels` is `[bids, asks]`. Each level has `px` (price), `sz` (size), and `n` (number of orders). Unresolved coins (timeout, unsubscribe during wait, or no snapshot yet) stay `null`. + +### trades (`coins`) + +```jsonc +{ + "BTC": [ + { + "coin": "BTC", + "side": "B", + "px": "62541.0", + "sz": "0.0006", + "hash": "0x...", + "time": 1782203279565, + "tid": 414334974001319, + "users": ["0x...", "0x..."] + } + ], + "ETH": null +} +``` + +Each entry matches Hydromancer's `WsTrade`. `side` is the taker's side (`"B"` bought, `"A"` sold). `users` is `[buyer, seller]`. `time` is a millisecond execution timestamp. Trades stay in arrival order; any trade older than `tradesKeepSeconds` is dropped. Unresolved coins stay `null`. A coin whose trades have all aged out is `[]`. diff --git a/workspace/data-proxy/src/modules/hydromancer/hydromancer.test.ts b/workspace/data-proxy/src/modules/hydromancer/hydromancer.test.ts index 9be3a2d..b353f3d 100644 --- a/workspace/data-proxy/src/modules/hydromancer/hydromancer.test.ts +++ b/workspace/data-proxy/src/modules/hydromancer/hydromancer.test.ts @@ -1,10 +1,18 @@ import { afterEach, beforeEach, describe, expect, it, mock } from "bun:test"; -import { Duration, Effect, LogLevel, Logger } from "effect"; +import { + Duration, + Effect, + LogLevel, + Logger, + TestClock, + TestContext, +} from "effect"; import * as v from "valibot"; import { type BookSnapshot, type HydromancerModuleConfig, HydromancerModuleRouteSchema, + type Trade, } from "../../config/hydromancer-module-config"; import { ModuleService } from "../module"; import { HydromancerModuleService } from "./hydromancer"; @@ -35,6 +43,12 @@ const baseConfig: HydromancerModuleConfig = { l2BookWaitTimeout: Duration.seconds(1), l2BookCleanupTtl: Duration.minutes(2), l2BookCleanupInterval: Duration.seconds(30), + tradesSubscriptionCoins: [], + tradesMaxCoinsPerRequest: 20, + tradesKeepSeconds: 60, + tradesWaitTimeout: Duration.seconds(1), + tradesCleanupTtl: Duration.minutes(2), + tradesCleanupInterval: Duration.seconds(30), }; const btcCtx = { @@ -72,6 +86,30 @@ const buildL2BookRequest = (coins: string[]) => const l2BookBody = (coins: string[]) => JSON.stringify({ type: "l2Book", coins }); +const btcTrade: Trade = { + coin: "BTC", + side: "B", + px: "62541.0", + sz: "0.0006", + hash: "0xabc", + time: Date.now(), + tid: 1, + users: [ + "0xf83fc34248744a304872e1ed40b9b10f54a64f6f", + "0x2ca4927174ba283d8a57f60ef3589844035a2930", + ], +}; + +const buildTradesRequest = (coins: string[]) => + new Request("http://proxy.local/info", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ type: "trades", coins }), + }); + +const tradesBody = (coins: string[]) => + JSON.stringify({ type: "trades", coins }); + const buildRoute = () => v.parse(HydromancerModuleRouteSchema, { type: "hydromancer", @@ -929,3 +967,232 @@ describe("HydromancerModuleService l2Book flow", () => { expect(response.status).toBe(400); }); }); + +describe("HydromancerModuleService trades flow", () => { + const originalWebSocket = globalThis.WebSocket; + + beforeEach(() => { + FakeWebSocket.instances = []; + globalThis.WebSocket = FakeWebSocket as unknown as typeof WebSocket; + }); + + afterEach(() => { + globalThis.WebSocket = originalWebSocket; + }); + + it("subscribes the pre-seeded coin on open and returns its trades", async () => { + const config: HydromancerModuleConfig = { + ...baseConfig, + tradesSubscriptionCoins: ["BTC"], + }; + + const program = Effect.gen(function* () { + const svc = yield* ModuleService; + yield* svc.start(); + const ws = yield* waitForSocket(); + ws.triggerOpen(); + yield* Effect.yieldNow(); + const sentOnOpen = [...ws.sent]; + + ws.triggerMessage( + JSON.stringify({ channel: "trades", trades: [btcTrade] }), + ); + yield* Effect.yieldNow(); + + const route = buildRoute(); + const response = yield* svc.handleRequest( + route, + {}, + buildTradesRequest(["BTC"]), + tradesBody(["BTC"]), + ); + return { response, sentOnOpen }; + }); + + const { response, sentOnOpen } = await runSilently( + program.pipe(Effect.provide(HydromancerModuleService(config))), + ); + + expect(sentOnOpen).toEqual([buildSubscribeFrame("trades", "BTC")]); + expect(response.status).toBe(200); + expect(await response.json()).toEqual({ BTC: [btcTrade] }); + }); + + it("returns null for an unseeded coin once tradesWaitTimeout elapses", async () => { + const config: HydromancerModuleConfig = { + ...baseConfig, + tradesSubscriptionCoins: [], + tradesWaitTimeout: Duration.millis(50), + }; + + const program = Effect.gen(function* () { + const svc = yield* ModuleService; + yield* svc.start(); + const ws = yield* waitForSocket(); + ws.triggerOpen(); + yield* Effect.yieldNow(); + + const route = buildRoute(); + return yield* svc.handleRequest( + route, + {}, + buildTradesRequest(["BTC"]), + tradesBody(["BTC"]), + ); + }); + + const response = await runSilently( + program.pipe(Effect.provide(HydromancerModuleService(config))), + ); + + expect(response.status).toBe(200); + expect(await response.json()).toEqual({ BTC: null }); + }); + + it("splits a mixed trades frame into one list per coin", async () => { + const ethTrade: Trade = { ...btcTrade, coin: "ETH", tid: 2 }; + const config: HydromancerModuleConfig = { + ...baseConfig, + tradesSubscriptionCoins: ["BTC", "ETH"], + }; + + const program = Effect.gen(function* () { + const svc = yield* ModuleService; + yield* svc.start(); + const ws = yield* waitForSocket(); + ws.triggerOpen(); + yield* Effect.yieldNow(); + + ws.triggerMessage( + JSON.stringify({ channel: "trades", trades: [btcTrade, ethTrade] }), + ); + yield* Effect.yieldNow(); + + const route = buildRoute(); + return yield* svc.handleRequest( + route, + {}, + buildTradesRequest(["BTC", "ETH"]), + tradesBody(["BTC", "ETH"]), + ); + }); + + const response = await runSilently( + program.pipe(Effect.provide(HydromancerModuleService(config))), + ); + + expect(response.status).toBe(200); + expect(await response.json()).toEqual({ + BTC: [btcTrade], + ETH: [ethTrade], + }); + }); + + it("drops cached trades that fall outside tradesKeepSeconds before the request", async () => { + const config: HydromancerModuleConfig = { + ...baseConfig, + tradesSubscriptionCoins: ["BTC", "ETH"], // so we can send frames without sending a request first + tradesKeepSeconds: 60, + }; + + const requestTrades = Effect.gen(function* () { + const svc = yield* ModuleService; + const route = buildRoute(); + const response = yield* svc.handleRequest( + route, + {}, + buildTradesRequest(["BTC", "ETH"]), + tradesBody(["BTC", "ETH"]), + ); + return yield* Effect.promise(() => response.json()); + }); + + const program = Effect.gen(function* () { + const now = yield* TestClock.currentTimeMillis; + const trade = ( + coin: "BTC" | "ETH", + tid: number, + time: number, + side: Trade["side"] = "B", + ): Trade => ({ ...btcTrade, coin, side, tid, time }); + + const ethEarly = trade("ETH", 1, now - 50_000); + const btcEarly = trade("BTC", 2, now - 45_000); + const btcBatch = [ + trade("BTC", 3, now - 15_000, "A"), + trade("BTC", 4, now - 15_000, "A"), + trade("BTC", 5, now - 15_000, "A"), + ]; + const ethLater = trade("ETH", 6, now - 5_000); + const btcLater = trade("BTC", 7, now - 1_000, "A"); + + const svc = yield* ModuleService; + yield* svc.start(); + const ws = yield* waitForSocket(); + ws.triggerOpen(); + yield* Effect.yieldNow(); + + for (const frame of [ + { channel: "trades", trades: [ethEarly] }, + { channel: "trades", trades: [btcEarly] }, + { type: "ping" }, + { channel: "trades", trades: btcBatch }, + { channel: "trades", trades: [ethLater] }, + { channel: "trades", trades: [btcLater] }, + ]) { + ws.triggerMessage(JSON.stringify(frame)); + } + yield* Effect.yieldNow(); + + yield* TestClock.adjust(Duration.seconds(40)); + const withinWindow = yield* requestTrades; + + yield* TestClock.adjust(Duration.seconds(30)); + const agedOut = yield* requestTrades; + return { withinWindow, agedOut, btcBatch, ethLater, btcLater }; + }); + + const { withinWindow, agedOut, btcBatch, ethLater, btcLater } = + await runSilently( + program.pipe( + Effect.provide(HydromancerModuleService(config)), + Effect.provide(TestContext.TestContext), + ), + ); + + expect(withinWindow).toEqual({ + BTC: [...btcBatch, btcLater], + ETH: [ethLater], + }); + expect(agedOut).toEqual({ BTC: [], ETH: [] }); + }); + + it("rejects a trades batch larger than tradesMaxCoinsPerRequest", async () => { + const config: HydromancerModuleConfig = { + ...baseConfig, + tradesMaxCoinsPerRequest: 1, + }; + + const program = Effect.gen(function* () { + const svc = yield* ModuleService; + yield* svc.start(); + const ws = yield* waitForSocket(); + ws.triggerOpen(); + yield* Effect.yieldNow(); + + const route = buildRoute(); + return yield* svc.handleRequest( + route, + {}, + buildTradesRequest(["BTC", "ETH"]), + tradesBody(["BTC", "ETH"]), + ); + }); + + const response = await runSilently( + program.pipe(Effect.provide(HydromancerModuleService(config))), + ); + + expect(response.status).toBe(400); + }); +}); diff --git a/workspace/data-proxy/src/modules/hydromancer/hydromancer.ts b/workspace/data-proxy/src/modules/hydromancer/hydromancer.ts index 8f9f2bf..071f1e8 100644 --- a/workspace/data-proxy/src/modules/hydromancer/hydromancer.ts +++ b/workspace/data-proxy/src/modules/hydromancer/hydromancer.ts @@ -12,6 +12,7 @@ import { type AssetCtx, type BookSnapshot, type HydromancerModuleConfig, + type Trade, parseHydromancerBody, } from "../../config/hydromancer-module-config"; import { createErrorResponse } from "../../controllers/create-error-response"; @@ -26,6 +27,17 @@ import { } from "./rest-fallback"; import { type HydromancerChannel, createHydromancerWS } from "./ws-client"; +/** Returns trades whose execution time is inside the window. */ +export const retainTradesWithinWindow = ( + trades: readonly Trade[], + keepMillis: number, + now: number, +): readonly Trade[] => { + const cutoff = now - keepMillis; + const kept = trades.filter((trade) => trade.time >= cutoff); + return kept.length === trades.length ? trades : kept; +}; + export const HydromancerModuleService = (config: HydromancerModuleConfig) => Layer.effect( ModuleService, @@ -36,18 +48,36 @@ export const HydromancerModuleService = (config: HydromancerModuleConfig) => restBaseUrl: config.restBaseUrl, }); - // Two caches with deliberately different shapes. + // Three caches with deliberately different shapes. // assetContext is freshness-keyed: a stale read falls back to a REST fetch. // l2Book is waiter-keyed: a request waits briefly and returns null on timeout. + // trades is waiter-keyed too, and each coin keeps trades executed inside tradesKeepSeconds. const cache = yield* createFreshnessCache(); const bookCache = yield* createPriceCache({ timeout: config.l2BookWaitTimeout, }); + const clock = yield* Clock.clockWith((clock) => Effect.succeed(clock)); + const tradesKeepMillis = config.tradesKeepSeconds * 1000; + const tradesCache = yield* createPriceCache({ + timeout: config.tradesWaitTimeout, + apply: (prev, next) => + retainTradesWithinWindow( + prev === undefined ? next : prev.concat(next), + tradesKeepMillis, + clock.unsafeCurrentTimeMillis(), + ), + }); const assetCtxStaleAfterMillis = Duration.toMillis(config.staleAfter); - const ws = yield* createHydromancerWS(config, cache, bookCache); + const ws = yield* createHydromancerWS( + config, + cache, + bookCache, + tradesCache, + ); const lastRequestToCoin = MutableHashMap.empty(); const lastRequestToBookCoin = MutableHashMap.empty(); + const lastRequestToTradesCoin = MutableHashMap.empty(); const subscriptionKinds = [ { @@ -73,6 +103,18 @@ export const HydromancerModuleService = (config: HydromancerModuleConfig) => yield* bookCache.deletePrice(coin); }), }, + { + name: "trades", + channel: "trades" as HydromancerChannel, + lastRequest: lastRequestToTradesCoin, + ttl: config.tradesCleanupTtl, + interval: config.tradesCleanupInterval, + onEvict: (coin: string) => + Effect.gen(function* () { + yield* tradesCache.setPriceToError(coin, "unsubscribed"); + yield* tradesCache.deletePrice(coin); + }), + }, ]; const start = () => @@ -89,6 +131,9 @@ export const HydromancerModuleService = (config: HydromancerModuleConfig) => for (const coin of config.l2BookSubscriptionCoins) { yield* ws.subscribe("l2Book", coin); } + for (const coin of config.tradesSubscriptionCoins) { + yield* ws.subscribe("trades", coin); + } for (const kind of subscriptionKinds) { yield* forkIdleCleanup({ @@ -217,6 +262,35 @@ export const HydromancerModuleService = (config: HydromancerModuleConfig) => Effect.withSpan("handleL2BookRequest", { attributes: { coins } }), ); + const handleTradesRequest = (coins: string[]) => + Effect.gen(function* () { + const { resolved } = yield* prepareRequest( + coins, + config.tradesMaxCoinsPerRequest, + "trades", + lastRequestToTradesCoin, + ); + + for (const coin of coins) { + const trades = yield* tradesCache.getOrWaitPrice(coin); + if (trades === null) continue; + const kept = retainTradesWithinWindow( + trades, + tradesKeepMillis, + yield* Clock.currentTimeMillis, + ); + if (kept !== trades) tradesCache.replaceCached(coin, kept); + resolved[coin] = kept; + } + + return new Response(JSON.stringify(resolved), { + status: 200, + headers: { "Content-Type": "application/json" }, + }); + }).pipe( + Effect.withSpan("handleTradesRequest", { attributes: { coins } }), + ); + const handleRequest = ( route: Route, _params: Record, @@ -277,6 +351,9 @@ export const HydromancerModuleService = (config: HydromancerModuleConfig) => Match.when({ type: "l2Book", coins: Match.any }, (body) => handleL2BookRequest(expandCoins(body.coins)), ), + Match.when({ type: "trades", coins: Match.any }, (body) => + handleTradesRequest(expandCoins(body.coins)), + ), Match.exhaustive, ); }).pipe( diff --git a/workspace/data-proxy/src/modules/hydromancer/ws-client.test.ts b/workspace/data-proxy/src/modules/hydromancer/ws-client.test.ts index d01e4a6..4f3be3c 100644 --- a/workspace/data-proxy/src/modules/hydromancer/ws-client.test.ts +++ b/workspace/data-proxy/src/modules/hydromancer/ws-client.test.ts @@ -1,5 +1,6 @@ import { afterEach, beforeEach, describe, expect, it } from "bun:test"; import { + Clock, Duration, Effect, Fiber, @@ -14,9 +15,11 @@ import type { AssetCtx, BookSnapshot, HydromancerModuleConfig, + Trade, } from "../../config/hydromancer-module-config"; import { createFreshnessCache } from "../shared/freshness-cache"; import { createPriceCache } from "../shared/price-cache"; +import { retainTradesWithinWindow } from "./hydromancer"; import { buildSubscribeFrame, buildUnsubscribeFrame, @@ -53,6 +56,20 @@ describe("buildSubscribeFrame / buildUnsubscribeFrame", () => { subscription: { type: "l2Book", coin: "BTC" }, }); }); + + it("produces the documented trades subscribe shape", () => { + expect(JSON.parse(buildSubscribeFrame("trades", "BTC"))).toEqual({ + method: "subscribe", + subscription: { type: "trades", coin: "BTC" }, + }); + }); + + it("produces the documented trades unsubscribe shape", () => { + expect(JSON.parse(buildUnsubscribeFrame("trades", "ETH"))).toEqual({ + method: "unsubscribe", + subscription: { type: "trades", coin: "ETH" }, + }); + }); }); describe("parseInboundFrame", () => { @@ -144,6 +161,40 @@ describe("parseInboundFrame", () => { }); expect(parseInboundFrame(frame)).toBeNull(); }); + + it("extracts trades from a valid trades frame", () => { + const trades: Trade[] = [ + { + coin: "BTC", + side: "B", + px: "62541.0", + sz: "0.0006", + hash: "0xabc", + time: 1782203279565, + tid: 414334974001319, + users: [ + "0xf83fc34248744a304872e1ed40b9b10f54a64f6f", + "0x2ca4927174ba283d8a57f60ef3589844035a2930", + ], + }, + ]; + const frame = JSON.stringify({ + type: "trades", + channel: "trades", + seq: 1, + cursor: "500:1704067200000:0", + trades, + }); + expect(parseInboundFrame(frame)).toEqual({ kind: "trades", trades }); + }); + + it("returns null for a trades frame with a malformed trade", () => { + const frame = JSON.stringify({ + channel: "trades", + trades: [{ coin: "BTC", side: "B", px: "1" }], + }); + expect(parseInboundFrame(frame)).toBeNull(); + }); }); class FakeWebSocket extends EventTarget { @@ -244,6 +295,12 @@ const baseConfig: HydromancerModuleConfig = { l2BookWaitTimeout: Duration.seconds(1), l2BookCleanupTtl: Duration.minutes(2), l2BookCleanupInterval: Duration.seconds(30), + tradesSubscriptionCoins: [], + tradesMaxCoinsPerRequest: 20, + tradesKeepSeconds: 60, + tradesWaitTimeout: Duration.seconds(1), + tradesCleanupTtl: Duration.minutes(2), + tradesCleanupInterval: Duration.seconds(30), }; const originalWebSocket = globalThis.WebSocket; @@ -261,22 +318,33 @@ afterEach(() => { const startService = ( config: HydromancerModuleConfig, preSubscribed: string[] = config.subscriptionCoins, - options?: Parameters[3], + options?: Parameters[4], ) => Effect.gen(function* () { const cache = yield* createFreshnessCache(); const bookCache = yield* createPriceCache(); + const clock = yield* Clock.clockWith((clock) => Effect.succeed(clock)); + const tradesKeepMillis = config.tradesKeepSeconds * 1000; + const tradesCache = yield* createPriceCache({ + apply: (prev, next) => + retainTradesWithinWindow( + prev === undefined ? next : prev.concat(next), + tradesKeepMillis, + clock.unsafeCurrentTimeMillis(), + ), + }); const ws = yield* createHydromancerWS( config, cache, bookCache, + tradesCache, options ?? { reconnectSchedule: Schedule.spaced(Duration.minutes(10)) }, ); for (const coin of preSubscribed) { yield* ws.subscribe("activeAssetCtx", coin); } const fiber = yield* ws.start(); - return { cache, bookCache, ws, fiber }; + return { cache, bookCache, tradesCache, ws, fiber }; }); describe("createHydromancerWS", () => { @@ -397,6 +465,105 @@ describe("createHydromancerWS", () => { ); }); + it("routes trades frames to the trades cache, split by coin", async () => { + await runWithTestClock( + Effect.gen(function* () { + const now = yield* TestClock.currentTimeMillis; + const btc: Trade = { + coin: "BTC", + side: "B", + px: "100", + sz: "1", + hash: "0x1", + time: now, + tid: 1, + users: ["0xbuyer", "0xseller"], + }; + const eth: Trade = { ...btc, coin: "ETH", tid: 2 }; + const btcNext: Trade = { ...btc, tid: 3, time: now + 1 }; + + const { + tradesCache, + bookCache, + ws: service, + fiber, + } = yield* startService(baseConfig, []); + yield* waitForInstances(1); + const ws = FakeWebSocket.instances[0]; + ws.triggerOpen(); + + yield* service.subscribe("trades", "BTC"); + + ws.triggerMessage( + JSON.stringify({ channel: "trades", trades: [btc, eth] }), + ); + ws.triggerMessage( + JSON.stringify({ channel: "trades", trades: [btcNext] }), + ); + + expect(yield* tradesCache.getOrWaitPrice("BTC")).toEqual([ + btc, + btcNext, + ]); + expect(tradesCache.size()).toBe(1); + expect(bookCache.size()).toBe(0); + + yield* Fiber.interrupt(fiber); + }), + ); + }); + + it("drops trades executed outside tradesKeepSeconds, regardless of order", async () => { + await runWithTestClock( + Effect.gen(function* () { + const now = yield* TestClock.currentTimeMillis; + const trade = (tid: number, time: number): Trade => ({ + coin: "BTC", + side: "A", + px: "100", + sz: "1", + hash: "0x1", + time, + tid, + users: ["0xbuyer", "0xseller"], + }); + const fresh = trade(1, now - 1_000); + const stale = trade(2, now - 120_000); + const freshLater = trade(3, now - 500); + + const { + tradesCache, + ws: service, + fiber, + } = yield* startService({ ...baseConfig, tradesKeepSeconds: 60 }, []); + yield* waitForInstances(1); + const ws = FakeWebSocket.instances[0]; + ws.triggerOpen(); + yield* service.subscribe("trades", "BTC"); + + ws.triggerMessage( + JSON.stringify({ + channel: "trades", + trades: [stale, fresh], + }), + ); + ws.triggerMessage( + JSON.stringify({ + channel: "trades", + trades: [freshLater, stale], + }), + ); + + expect(yield* tradesCache.getOrWaitPrice("BTC")).toEqual([ + fresh, + freshLater, + ]); + + yield* Fiber.interrupt(fiber); + }), + ); + }); + it("marks hasError after the socket closes", async () => { await runWithTestClock( Effect.gen(function* () { diff --git a/workspace/data-proxy/src/modules/hydromancer/ws-client.ts b/workspace/data-proxy/src/modules/hydromancer/ws-client.ts index 3ef5f63..dfe70ea 100644 --- a/workspace/data-proxy/src/modules/hydromancer/ws-client.ts +++ b/workspace/data-proxy/src/modules/hydromancer/ws-client.ts @@ -4,6 +4,7 @@ import { Duration, Effect, type Fiber, + Match, MutableHashMap, Option, Runtime, @@ -16,12 +17,14 @@ import { type BookSnapshot, BookSnapshotSchema, type HydromancerModuleConfig, + type Trade, + TradeSchema, } from "../../config/hydromancer-module-config"; import type { FreshnessCache } from "../shared/freshness-cache"; import type { PriceCache } from "../shared/price-cache"; import { recordTickHandle } from "../shared/tick-metrics"; -export type HydromancerChannel = "activeAssetCtx" | "l2Book"; +export type HydromancerChannel = "activeAssetCtx" | "l2Book" | "trades"; const InboundAssetCtxFrameSchema = v.object({ channel: v.literal("activeAssetCtx"), @@ -36,14 +39,21 @@ const InboundBookFrameSchema = v.object({ data: BookSnapshotSchema, }); +const InboundTradesFrameSchema = v.object({ + channel: v.literal("trades"), + trades: v.array(TradeSchema), +}); + const InboundFrameSchema = v.variant("channel", [ InboundAssetCtxFrameSchema, InboundBookFrameSchema, + InboundTradesFrameSchema, ]); export type ParsedInboundFrame = | { kind: "activeAssetCtx"; coin: string; ctx: AssetCtx } - | { kind: "l2Book"; snapshot: BookSnapshot }; + | { kind: "l2Book"; snapshot: BookSnapshot } + | { kind: "trades"; trades: Trade[] }; export const buildSubscribeFrame = ( channel: HydromancerChannel, @@ -83,7 +93,10 @@ export const parseInboundFrame = (raw: string): ParsedInboundFrame | null => { ctx: parsed.value.data.ctx, }; } - return { kind: "l2Book", snapshot: parsed.value.data }; + if (parsed.value.channel === "l2Book") { + return { kind: "l2Book", snapshot: parsed.value.data }; + } + return { kind: "trades", trades: parsed.value.trades }; }; export const defaultReconnectSchedule = (config: HydromancerModuleConfig) => @@ -123,6 +136,7 @@ export const createHydromancerWS = ( config: HydromancerModuleConfig, assetCache: FreshnessCache, bookCache: PriceCache, + tradesCache: PriceCache, options?: CreateHydromancerWSOptions, ): Effect.Effect => Effect.gen(function* () { @@ -137,6 +151,10 @@ export const createHydromancerWS = ( subscribeFrame: (coin) => buildSubscribeFrame("l2Book", coin, config.l2BookNSigFigs), }, + trades: { + desired: MutableHashMap.empty(), + subscribeFrame: (coin) => buildSubscribeFrame("trades", coin), + }, }; let currentWS: WebSocket | null = null; let unhealthy = false; @@ -198,25 +216,52 @@ export const createHydromancerWS = ( const started = performance.now(); const frame = parseInboundFrame(raw); if (!frame) return; - if (frame.kind === "activeAssetCtx") { - if ( - Option.isNone( - MutableHashMap.get(channels.activeAssetCtx.desired, frame.coin), - ) - ) { - return; - } - assetCache.setSync(frame.coin, frame.ctx, Date.now()); - } else { - if ( - Option.isNone( - MutableHashMap.get(channels.l2Book.desired, frame.snapshot.coin), - ) - ) { - return; - } - bookCache.setPriceSync(frame.snapshot.coin, frame.snapshot); - } + const handled = Match.value(frame).pipe( + Match.when({ kind: "activeAssetCtx" }, (frame) => { + if ( + Option.isNone( + MutableHashMap.get(channels.activeAssetCtx.desired, frame.coin), + ) + ) { + return false; + } + assetCache.setSync(frame.coin, frame.ctx, Date.now()); + return true; + }), + Match.when({ kind: "l2Book" }, (frame) => { + if ( + Option.isNone( + MutableHashMap.get(channels.l2Book.desired, frame.snapshot.coin), + ) + ) { + return false; + } + bookCache.setPriceSync(frame.snapshot.coin, frame.snapshot); + return true; + }), + Match.when({ kind: "trades" }, (frame) => { + const byCoin = new Map(); + for (const trade of frame.trades) { + if ( + Option.isNone( + MutableHashMap.get(channels.trades.desired, trade.coin), + ) + ) { + continue; + } + const batch = byCoin.get(trade.coin); + if (batch === undefined) byCoin.set(trade.coin, [trade]); + else batch.push(trade); + } + if (byCoin.size === 0) return false; + for (const [coin, trades] of byCoin) { + tradesCache.setPriceSync(coin, trades); + } + return true; + }), + Match.exhaustive, + ); + if (!handled) return; recordTickHandle("hydromancer", config.name, performance.now() - started); }; diff --git a/workspace/data-proxy/src/modules/shared/price-cache.ts b/workspace/data-proxy/src/modules/shared/price-cache.ts index e4df7b1..6e4a0ac 100644 --- a/workspace/data-proxy/src/modules/shared/price-cache.ts +++ b/workspace/data-proxy/src/modules/shared/price-cache.ts @@ -31,6 +31,8 @@ const makeWaiter = (): PriceWaiter => { export interface PriceCache { getOrWaitPrice: (key: K) => Effect.Effect; setPriceSync: (key: K, price: V) => void; + /** Overwrites an existing entry without waking waiters. */ + replaceCached: (key: K, price: V) => void; deletePrice: (key: K) => Effect.Effect; setPriceToError: (key: K, error: string) => Effect.Effect; size: () => number; @@ -39,6 +41,7 @@ export interface PriceCache { const replace = (_prev: V | undefined, next: V): V => next; export const createPriceCache = (options?: { + /** The timeout for a price wait. Defaults to 3 seconds. */ timeout?: Duration.Duration; /** Produces the value to store from the cached value, if any, and the incoming one. */ apply?: (prev: V | undefined, next: V) => V; @@ -63,6 +66,11 @@ export const createPriceCache = (options?: { } }; + const replaceCached = (key: K, price: V): void => { + if (Option.isNone(MutableHashMap.get(priceCache, key))) return; + MutableHashMap.set(priceCache, key, price); + }; + const setPriceToError = (key: K, error: string) => Effect.sync(() => { const waiter = MutableHashMap.get(priceWaiters, key); @@ -119,6 +127,7 @@ export const createPriceCache = (options?: { return { getOrWaitPrice, setPriceSync, + replaceCached, deletePrice, setPriceToError, size,