add per-token net buy/sell flow endpoint

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": [

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`,

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 }[],
   ) {

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 () => {

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,