diff --git a/openapi.json b/openapi.json index b0a5df2..a87d22f 100644 --- a/openapi.json +++ b/openapi.json @@ -3057,6 +3057,174 @@ } } }, + "/tokens/{chainId}/{tokenAddress}/net-flow": { + "get": { + "tags": [ + "Prices" + ], + "summary": "Get token net buy and sell flow", + "description": "Returns how much of a token traders bought from and sold into Ekubo pools in each interval, across every pool on the chain that holds the token, including TWAMM and limit order execution. Buckets use the same boundaries as the token price history and only buckets with at least one swap are returned; an unknown token returns no buckets. A multi-hop route that passes through the token adds the same amount to both bought and sold, so the net flow is exact while the gross amounts include that routing.", + "operationId": "get_GetTokenNetFlow", + "parameters": [ + { + "schema": { + "type": "integer", + "title": "Chain ID", + "description": "The ID of the network that is being fetched", + "format": "int64", + "minimum": 1, + "maximum": 9223372036854776000 + }, + "required": true, + "description": "The ID of the network that is being fetched", + "name": "chainId", + "in": "path" + }, + { + "schema": { + "anyOf": [ + { + "type": "string", + "pattern": "^0x[a-fA-F0-9]+$", + "description": "A hexadecimal number" + }, + { + "type": "string", + "pattern": "^\\d+$", + "description": "A decimal number" + } + ], + "title": "Address", + "description": "An address on the specified blockchain network" + }, + "required": true, + "description": "An address on the specified blockchain network", + "name": "tokenAddress", + "in": "path" + }, + { + "schema": { + "type": "integer", + "minimum": 60, + "maximum": 86400, + "default": 900, + "description": "Bucket size in seconds" + }, + "required": false, + "description": "Bucket size in seconds", + "name": "interval", + "in": "query" + }, + { + "schema": { + "type": "integer", + "minimum": 3600, + "maximum": 2592000, + "default": 86400, + "description": "History duration in seconds ending now, up to 30 days and at most 1000 intervals" + }, + "required": false, + "description": "History duration in seconds ending now, up to 30 days and at most 1000 intervals", + "name": "duration", + "in": "query" + } + ], + "responses": { + "200": { + "description": "The token's bought and sold amounts per interval", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "timestamp": { + "type": "integer", + "description": "Response time, unix seconds" + }, + "start": { + "type": "integer", + "description": "Start of the window, unix seconds" + }, + "end": { + "type": "integer", + "description": "End of the window, unix seconds" + }, + "interval": { + "type": "integer" + }, + "data": { + "type": "array", + "items": { + "type": "object", + "properties": { + "start": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "string", + "format": "date-time" + } + ], + "description": "Bucket start time" + }, + "bought": { + "type": "string", + "description": "Raw token units that left Ekubo pools to traders in the bucket, as a decimal string" + }, + "sold": { + "type": "string", + "description": "Raw token units traders paid into Ekubo pools in the bucket, as a decimal string" + }, + "swap_count": { + "type": "integer", + "description": "Number of swaps in the bucket that touched the token" + } + }, + "required": [ + "start", + "bought", + "sold", + "swap_count" + ] + } + } + }, + "required": [ + "timestamp", + "start", + "end", + "interval", + "data" + ] + } + } + } + }, + "400": { + "description": "The request parameters are invalid", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + }, + "500": { + "description": "An unexpected server error occurred", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + } + } + } + }, "/price/{chainId}/{baseToken}/{quoteToken}/history": { "get": { "tags": [ diff --git a/src/contract/cases/prices.ts b/src/contract/cases/prices.ts index 107e44e..019d9ac 100644 --- a/src/contract/cases/prices.ts +++ b/src/contract/cases/prices.ts @@ -148,6 +148,39 @@ export const pricesCases: ContractCase[] = [ ]); }, })), + ...[ + ["day", 1_800, 86_400], + ["week", 14_400, 604_800], + ["month", 43_200, 2_592_000], + ].map(([range, interval, duration]): ContractCase => ({ + operation: "/tokens/{chainId}/{tokenAddress}/net-flow", + name: `/tokens/{chainId}/{tokenAddress}/net-flow ${range}`, + request: `/tokens/1/${USDC}/net-flow?interval=${interval}&duration=${duration}`, + stub: () => { + stubQuery("getTokenNetFlow", [ + { + start: at(START), + bought: "1250000000", + sold: "400000000", + swap_count: 7, + }, + { + start: at(START + Number(interval)), + bought: "0", + sold: "35000000", + swap_count: 1, + }, + ]); + }, + })), + { + operation: "/tokens/{chainId}/{tokenAddress}/net-flow", + name: "/tokens/{chainId}/{tokenAddress}/net-flow unknown token", + request: "/tokens/1/0x1234/net-flow", + stub: () => { + stubQuery("getTokenNetFlow", []); + }, + }, { operation: "/price/{chainId}/{baseToken}/{quoteToken}/history", request: `/price/1/0x0/${USDC}/history?interval=3600`, diff --git a/src/queries.ts b/src/queries.ts index ab77b15..4f29d5b 100644 --- a/src/queries.ts +++ b/src/queries.ts @@ -402,6 +402,76 @@ export class Queries { `; } + public async getTokenNetFlow({ + chainId, + tokenAddress, + start, + end, + intervalSeconds, + }: { + chainId: bigint; + tokenAddress: bigint; + start: Date; + end: Date; + intervalSeconds: number; + }) { + return this.sql< + { + start: Date; + bought: string; + sold: string; + swap_count: number; + }[] + >` + WITH token_pools AS MATERIALIZED ( + SELECT + ARRAY_AGG(pool_key_id) FILTER ( + WHERE token0 = ${tokenAddress.toString()} + ) AS as_token0, + ARRAY_AGG(pool_key_id) FILTER ( + WHERE token1 = ${tokenAddress.toString()} + ) AS as_token1 + FROM pool_keys + WHERE chain_id = ${chainId} + AND ( + token0 = ${tokenAddress.toString()} + OR token1 = ${tokenAddress.toString()} + ) + ), + -- Deltas are the pool's balance change: positive means the swapper paid + -- the token into the pool. Matching pool_key_id against an array keeps + -- each side a single index-only scan of + -- swaps_pool_key_id_block_time_ohlc_idx (which carries both deltas), one + -- range per pool; a join lets the planner misjudge tokens with + -- thousands of pools and scan every pool's swaps instead. + token_deltas AS ( + SELECT block_time, delta0 AS delta + FROM swaps + WHERE pool_key_id = ANY ((SELECT as_token0 FROM token_pools)::INT8[]) + AND block_time >= ${start} + AND block_time <= ${end} + UNION ALL + SELECT block_time, delta1 AS delta + FROM swaps + WHERE pool_key_id = ANY ((SELECT as_token1 FROM token_pools)::INT8[]) + AND block_time >= ${start} + AND block_time <= ${end} + ) + SELECT + date_bin( + ${intervalSeconds} * INTERVAL '1 second', + block_time, + TIMESTAMPTZ '2000-01-01 00:00:00+00' + ) AS start, + COALESCE(SUM(-delta) FILTER (WHERE delta < 0), 0)::TEXT AS bought, + COALESCE(SUM(delta) FILTER (WHERE delta > 0), 0)::TEXT AS sold, + COUNT(*)::INT4 AS swap_count + FROM token_deltas + GROUP BY 1 + ORDER BY 1 + `; + } + public async getErc20TokensByIds( ids: { chainId: bigint; tokenAddress: bigint }[], ) { diff --git a/src/router.test.ts b/src/router.test.ts index 9530b0b..2a37f3c 100644 --- a/src/router.test.ts +++ b/src/router.test.ts @@ -24,14 +24,14 @@ describe("Hono app integration", () => { ), ); - expect(Object.keys(schema.paths ?? {})).toHaveLength(59); - expect(operations).toHaveLength(59); + expect(Object.keys(schema.paths ?? {})).toHaveLength(60); + expect(operations).toHaveLength(60); expect( operations.reduce( (count, operation) => count + (operation.parameters?.length ?? 0), 0, ), - ).toBe(198); + ).toBe(202); const getToken = operations.find( (operation) => @@ -168,6 +168,17 @@ describe("Hono app integration", () => { ) ).status, ).toBe(400); + + for (const query of [ + "interval=59", + "duration=3599", + "duration=2592001", + "interval=60&duration=60060", + ]) { + expect((await fetchPath(`/tokens/1/0x1/net-flow?${query}`)).status).toBe( + 400, + ); + } }); test("rejects numeric path parameters that do not parse", async () => { diff --git a/src/router.ts b/src/router.ts index dc57938..0842817 100644 --- a/src/router.ts +++ b/src/router.ts @@ -32,6 +32,7 @@ import { GetPoolPriceHistory, GetTokenUsdPriceHistory, } from "./routes/prices"; +import { GetTokenNetFlow } from "./routes/prices/netFlow"; import { BatchListPositionsByAddress, GetPositionNftImage, @@ -92,6 +93,7 @@ const routes: readonly ApiRoute[] = [ GetPairInfoVolume, GetPairInfoPools, GetTokenUsdPriceHistory, + GetTokenNetFlow, GetPairPriceHistory, GetPairOhlcHistory, GetPoolPriceHistory, diff --git a/src/routes/prices/netFlow.pg.test.ts b/src/routes/prices/netFlow.pg.test.ts new file mode 100644 index 0000000..bec57b6 --- /dev/null +++ b/src/routes/prices/netFlow.pg.test.ts @@ -0,0 +1,119 @@ +import { afterAll, beforeAll, describe, expect, test } from "bun:test"; +import postgres from "postgres"; +import { createQueries, type Queries } from "../../queries"; + +const URL = process.env.TEST_PG_CONNECTION_STRING; + +const SCHEMA = "net_flow_test"; +const CHAIN = 1n; +const OTHER_CHAIN = 10n; +const TOKEN = 0x1234n; +const LOW = 0x10n; +const HIGH = 0x9999n; + +const TABLES = ` +DROP SCHEMA IF EXISTS ${SCHEMA} CASCADE; +CREATE SCHEMA ${SCHEMA}; +CREATE TABLE ${SCHEMA}.pool_keys ( + pool_key_id int8 PRIMARY KEY, chain_id int8 NOT NULL, + token0 numeric NOT NULL, token1 numeric NOT NULL); +CREATE TABLE ${SCHEMA}.swaps ( + event_id serial8 PRIMARY KEY, + pool_key_id int8 NOT NULL REFERENCES ${SCHEMA}.pool_keys (pool_key_id), + block_time timestamptz NOT NULL, + delta0 numeric NOT NULL, delta1 numeric NOT NULL); +`; + +describe.skipIf(!URL)("getTokenNetFlow", () => { + let admin: ReturnType; + let queries: Queries; + + beforeAll(async () => { + admin = postgres(URL, { max: 1, onnotice: () => {} }); + await admin.unsafe(TABLES); + await admin.unsafe(` + INSERT INTO ${SCHEMA}.pool_keys VALUES + (1, ${CHAIN}, ${TOKEN}, ${HIGH}), + (2, ${CHAIN}, ${LOW}, ${TOKEN}), + (3, ${CHAIN}, ${LOW}, ${HIGH}), + (4, ${OTHER_CHAIN}, ${TOKEN}, ${HIGH}); + INSERT INTO ${SCHEMA}.swaps (pool_key_id, block_time, delta0, delta1) VALUES + -- token is token0: the trader pays 100 in, then takes 30 out + (1, '2026-10-25T14:05:00Z', 100, -5), + (1, '2026-10-25T14:29:59Z', -30, 2), + -- token is token1: the trader takes 75 out in the same bucket + (2, '2026-10-25T14:10:00Z', 9, -75), + -- next bucket, and a swap that moved none of the token + (2, '2026-10-25T14:30:00Z', 4, 0), + -- pools without the token, or on another chain, are ignored + (3, '2026-10-25T14:10:00Z', 1000, -1000), + (4, '2026-10-25T14:10:00Z', 1000, -1000), + -- outside the window + (1, '2026-10-25T13:59:59Z', 1000, -1000); + `); + queries = await createQueries({ + PG_CONNECTION_STRING: `${URL}${URL?.includes("?") ? "&" : "?"}search_path=${SCHEMA}`, + }); + }); + + afterAll(async () => { + await admin?.unsafe(`DROP SCHEMA IF EXISTS ${SCHEMA} CASCADE`); + await admin?.end(); + }); + + test("splits each pool's token delta into bought and sold per bucket", async () => { + const rows = await queries.getTokenNetFlow({ + chainId: CHAIN, + tokenAddress: TOKEN, + start: new Date("2026-10-25T14:00:00Z"), + end: new Date("2026-10-25T15:00:00Z"), + intervalSeconds: 1_800, + }); + + expect([...rows]).toEqual([ + { + start: new Date("2026-10-25T14:00:00Z"), + bought: "105", + sold: "100", + swap_count: 3, + }, + { + start: new Date("2026-10-25T14:30:00Z"), + bought: "0", + sold: "0", + swap_count: 1, + }, + ]); + }); + + test("aligns buckets to the price history origin", async () => { + const rows = await queries.getTokenNetFlow({ + chainId: CHAIN, + tokenAddress: TOKEN, + start: new Date("2026-10-25T14:00:00Z"), + end: new Date("2026-10-25T15:00:00Z"), + intervalSeconds: 43_200, + }); + + expect([...rows]).toEqual([ + { + start: new Date("2026-10-25T12:00:00Z"), + bought: "105", + sold: "100", + swap_count: 4, + }, + ]); + }); + + test("returns no buckets for a token without pools", async () => { + const rows = await queries.getTokenNetFlow({ + chainId: CHAIN, + tokenAddress: 0xdeadn, + start: new Date("2026-10-25T14:00:00Z"), + end: new Date("2026-10-25T15:00:00Z"), + intervalSeconds: 1_800, + }); + + expect([...rows]).toEqual([]); + }); +}); diff --git a/src/routes/prices/netFlow.test.ts b/src/routes/prices/netFlow.test.ts new file mode 100644 index 0000000..9aede5a --- /dev/null +++ b/src/routes/prices/netFlow.test.ts @@ -0,0 +1,26 @@ +import { describe, expect, test } from "bun:test"; +import { StatusError } from "../../shared/errors"; +import { MAX_NET_FLOW_BUCKETS, netFlowWindow } from "./netFlow"; + +describe("netFlowWindow", () => { + test("ends now, truncated to the second, and spans the duration", () => { + const { start, end } = netFlowWindow(1_761_480_000_999, 1_800, 86_400); + expect(end.getTime()).toBe(1_761_480_000_000); + expect(start.getTime()).toBe(1_761_393_600_000); + }); + + test.each([ + [1_800, 86_400], + [14_400, 604_800], + [43_200, 2_592_000], + ])("accepts the chart's %i/%i request", (interval, duration) => { + expect(() => netFlowWindow(0, interval, duration)).not.toThrow(); + }); + + test("accepts exactly the bucket limit and rejects one more", () => { + expect(() => netFlowWindow(0, 60, 60 * MAX_NET_FLOW_BUCKETS)).not.toThrow(); + expect(() => netFlowWindow(0, 60, 60 * MAX_NET_FLOW_BUCKETS + 1)).toThrow( + StatusError, + ); + }); +}); diff --git a/src/routes/prices/netFlow.ts b/src/routes/prices/netFlow.ts new file mode 100644 index 0000000..6b14c32 --- /dev/null +++ b/src/routes/prices/netFlow.ts @@ -0,0 +1,131 @@ +import { createRoute, z } from "@hono/zod-openapi"; +import { defineRoute } from "../../shared/context"; +import { AddressType, ChainIdType } from "../../shared/validation/address"; +import { createQueries } from "../../queries"; +import { jsonResponse } from "../../shared/openapi"; +import { errorResponses, StatusError } from "../../shared/errors"; + +const DEFAULT_NET_FLOW_INTERVAL_SECONDS = 15 * 60; +const DEFAULT_NET_FLOW_DURATION_SECONDS = 24 * 60 * 60; +const MAX_NET_FLOW_DURATION_SECONDS = 30 * 86_400; +export const MAX_NET_FLOW_BUCKETS = 1_000; + +const TokenNetFlowPointType = z.object({ + start: z.union([z.string(), z.date()]).describe("Bucket start time"), + bought: z + .string() + .describe( + "Raw token units that left Ekubo pools to traders in the bucket, as a decimal string", + ), + sold: z + .string() + .describe( + "Raw token units traders paid into Ekubo pools in the bucket, as a decimal string", + ), + swap_count: z + .number() + .int() + .describe("Number of swaps in the bucket that touched the token"), +}); + +const GetTokenNetFlowResponseType = z.object({ + timestamp: z.number().int().describe("Response time, unix seconds"), + start: z.number().int().describe("Start of the window, unix seconds"), + end: z.number().int().describe("End of the window, unix seconds"), + interval: z.number().int(), + data: z.array(TokenNetFlowPointType), +}); + +export function netFlowWindow( + nowMilliseconds: number, + intervalSeconds: number, + durationSeconds: number, +): { start: Date; end: Date } { + if (durationSeconds / intervalSeconds > MAX_NET_FLOW_BUCKETS) { + throw new StatusError( + 400, + `Interval must produce at most ${MAX_NET_FLOW_BUCKETS} buckets`, + ); + } + + const endSeconds = Math.floor(nowMilliseconds / 1_000); + return { + start: new Date((endSeconds - durationSeconds) * 1_000), + end: new Date(endSeconds * 1_000), + }; +} + +export const GetTokenNetFlow = defineRoute( + createRoute({ + method: "get", + path: "/tokens/{chainId}/{tokenAddress}/net-flow", + tags: ["Prices"], + summary: "Get token net buy and sell flow", + description: + "Returns how much of a token traders bought from and sold into Ekubo pools in each interval, across every pool on the chain that holds the token, including TWAMM and limit order execution. " + + "Buckets use the same boundaries as the token price history and only buckets with at least one swap are returned; an unknown token returns no buckets. " + + "A multi-hop route that passes through the token adds the same amount to both bought and sold, so the net flow is exact while the gross amounts include that routing.", + operationId: "get_GetTokenNetFlow", + request: { + params: z.object({ + chainId: ChainIdType, + tokenAddress: AddressType, + }), + query: z.object({ + interval: z.coerce + .number() + .int() + .min(60) + .max(86_400) + .optional() + .describe("Bucket size in seconds") + .default(DEFAULT_NET_FLOW_INTERVAL_SECONDS), + duration: z.coerce + .number() + .int() + .min(3_600) + .max(MAX_NET_FLOW_DURATION_SECONDS) + .optional() + .describe( + `History duration in seconds ending now, up to 30 days and at most ${MAX_NET_FLOW_BUCKETS} intervals`, + ) + .default(DEFAULT_NET_FLOW_DURATION_SECONDS), + }), + }, + responses: { + 200: jsonResponse( + "The token's bought and sold amounts per interval", + GetTokenNetFlowResponseType, + ), + ...errorResponses, + }, + }), + async (c) => { + const { chainId, tokenAddress } = c.req.valid("param"); + const { interval: intervalSeconds, duration: durationSeconds } = + c.req.valid("query"); + const now = Date.now(); + const { start, end } = netFlowWindow(now, intervalSeconds, durationSeconds); + + const queries = await createQueries(c.env); + const data = await queries.getTokenNetFlow({ + chainId, + tokenAddress: BigInt(tokenAddress), + start, + end, + intervalSeconds, + }); + + const response = { + timestamp: Math.floor(now / 1_000), + start: start.getTime() / 1_000, + end: end.getTime() / 1_000, + interval: intervalSeconds, + data, + } satisfies z.infer; + + return c.json(response, 200, { + "cache-control": `public, max-age=${Math.ceil(intervalSeconds / 4)}, must-revalidate`, + }); + }, +);