diff --git a/docs/data-quality.md b/docs/data-quality.md index f41810ca..d1b1e325 100644 --- a/docs/data-quality.md +++ b/docs/data-quality.md @@ -62,7 +62,7 @@ TWAP and VWAP computations apply statistical filtering to reject manipulation sp - **TWAP Windows**: Window ranges from [`1`](../src/routes/price.ts#L8) to [`1440`](../src/routes/price.ts#L9) minutes (default [`60`](../src/routes/price.ts#L10) minutes). Sample intervals range from [`1`](../src/routes/price.ts#L11) to [`3600`](../src/routes/price.ts#L12) seconds (default [`60`](../src/routes/price.ts#L13) seconds). - **VWAP Windows**: Window ranges from [`1`](../src/routes/price.ts#L15) to [`1440`](../src/routes/price.ts#L16) minutes (default [`60`](../src/routes/price.ts#L17) minutes). - **Price History Limits**: Bucket count is capped at [`1000`](../src/api/rest.ts#L25) buckets (default [`100`](../src/api/rest.ts#L24) buckets). -- **Screener Pair Limits**: Pair count defaults to [`20`](../src/routes/screener.ts#L22) and is clamped between [`1`](../src/routes/screener.ts#L33) and [`100`](../src/routes/screener.ts#L23). +- **Screener Pair Limits**: Pair count defaults to [`20`](../src/services/screener.ts#L16) and is clamped between [`1`](../src/services/screener.ts#L109) and [`100`](../src/services/screener.ts#L17). - **Spread Observations**: Venues must quote at least [`2`](../src/routes/spreads.ts#L20) observations in the window to report a spread. - **Violation Behavior**: Inputs outside allowed ranges trigger HTTP [`400`](../src/routes/price.ts#L55) Bad Request with validation details. @@ -144,8 +144,8 @@ When a data quality guarantee is breached, Lens behaves deterministically: | `vwap_window_default` | [`60`](../src/routes/price.ts#L17) | minutes | [`src/routes/price.ts:17`](../src/routes/price.ts#L17) | Default VWAP window | | `history_default_limit` | [`100`](../src/api/rest.ts#L24) | buckets | [`src/api/rest.ts:24`](../src/api/rest.ts#L24) | Default bucket limit for historical aggregates | | `history_max_limit` | [`1000`](../src/api/rest.ts#L25) | buckets | [`src/api/rest.ts:25`](../src/api/rest.ts#L25) | Upper ceiling for historical aggregates | -| `screener_default_limit` | [`20`](../src/routes/screener.ts#L22) | pairs | [`src/routes/screener.ts:22`](../src/routes/screener.ts#L22) | Default limit of pairs returned by screener | -| `screener_max_limit` | [`100`](../src/routes/screener.ts#L23) | pairs | [`src/routes/screener.ts:23`](../src/routes/screener.ts#L23) | Maximum clamped limit of pairs returned by screener | +| `screener_default_limit` | [`20`](../src/services/screener.ts#L16) | pairs | [`src/services/screener.ts:16`](../src/services/screener.ts#L16) | Default limit of pairs returned by screener | +| `screener_max_limit` | [`100`](../src/services/screener.ts#L17) | pairs | [`src/services/screener.ts:17`](../src/services/screener.ts#L17) | Maximum clamped limit of pairs returned by screener | | `spread_min_observations` | [`2`](../src/routes/spreads.ts#L20) | observations | [`src/routes/spreads.ts:20`](../src/routes/spreads.ts#L20) | Minimum observations required before a venue spread is reported | | `registered_venues_count` | [`6`](../src/ingest/venues/index.ts#L23) | adapters | [`src/ingest/venues/index.ts:23`](../src/ingest/venues/index.ts#L23) | Total number of registered venue adapters | | `database_price_precision` | [`36`](../prisma/schema.prisma#L20) | digits | [`prisma/schema.prisma:20`](../prisma/schema.prisma#L20) | Total numeric precision in PostgreSQL decimal columns | diff --git a/docs/graphql.md b/docs/graphql.md index b1c2ea12..021b0d8f 100644 --- a/docs/graphql.md +++ b/docs/graphql.md @@ -13,6 +13,7 @@ Lens exposes a high-performance GraphQL API alongside its REST interface, powere - [`etBestRoute`](#2-getbestroute) - [`getPriceHistory`](#3-getpricehistory) - [`listPairs`](#4-listpairs) + - [`screener`](#5-screener) 4. [Live Price Subscriptions (`graphql-transport-ws`)](#live-price-subscriptions-graphql-transport-ws) - [Protocol Handshake Walkthrough](#protocol-handshake-walkthrough) - [Network Filtering Semantics](#network-filtering-semantics) @@ -69,6 +70,7 @@ type Query { getBestRoute(assetA: String!, assetB: String!, amount: Float!): RouteInfo getPriceHistory(assetA: String!, assetB: String!, window: String!, from: String, to: String, limit: Int): [PriceBucket] listPairs: [String]! + screener(filters: ScreenerFilters, sortBy: String = "volume", order: ScreenerSortOrder = DESC, cursor: String, limit: Int = 20): ScreenerPage! } type Subscription { @@ -364,6 +366,92 @@ curl -s -X POST http://localhost:3002/graphql \ --- +### 5. `screener` + +Screens tracked pairs by volume, 24-hour change, price and AMM liquidity, +with cursor pagination. It shares its implementation with the REST +`GET /screener` route: same filters, same sort fields, same cursor semantics +and the same `[1, 100]` limit clamp (never rejected). + +#### Arguments + +| Argument | Type | Default | Description | +|---|---|---|---| +| `filters` | `ScreenerFilters` | none | Numeric filters (`volume`, `change24hMin`, `change24hMax`, `priceMin`, `priceMax`, `liquidity`); omitted fields do not filter. Each numeric filter is a lower/upper bound as its name says. | +| `sortBy` | `String` | `"volume"` | One of `volume`, `change_24h`, `price`, `liquidity`. | +| `order` | `ScreenerSortOrder` | `DESC` | `ASC` or `DESC`. | +| `cursor` | `String` | none | Opaque `nextCursor` from the previous page. | +| `limit` | `Int` | `20` | Page size, clamped to `[1, 100]`. | + +> [!IMPORTANT] +> **`market_cap` is rejected, not ignored.** Lens has no circulating supply +data to compute a market cap from. Both the `marketCap` filter and +> `sortBy: "market_cap"` fail the query with an `UNSUPPORTED_PARAMETER` / +> `INVALID_INPUT` error carrying the same explanation REST gives — never a +> silent null. Use `liquidity` instead. + +#### Query Definition +```graphql +query ScreenPairs($filters: ScreenerFilters, $sortBy: String, $limit: Int) { + screener(filters: $filters, sortBy: $sortBy, limit: $limit) { + network + hasMore + nextCursor + rows { + pairKey + volume + price + change24h + liquidity + liquidityUnit + } + } +} +``` + +#### Runnable `curl` Command +[```bash +curl -s -X POST http://localhost:3002/graphql \ + -H "Content-Type: application/json" \ + -d '{ + "query": "query ScreenPairs($filters: ScreenerFilters, $sortBy: String, $limit: Int) { screener(filters: $filters, sortBy: $sortBy, limit: $limit) { network hasMore nextCursor rows { pairKey volume price change24h liquidity liquidityUnit } } }", + "variables": { + "filters": { "volume": 1000, "change24hMin": -10 }, + "sortBy": "liquidity", + "limit": 5 + } + }' +```] + +#### Example Response +```json +{ + "data": { + "screener": { + "network": "testnet", + "hasMore": true, + "nextCursor": "eyJib2x1bWUiOjUwMDAwfQ==", + "rows": [ + { + "pairKey": "XLM/USDC", + "volume": 1425000.5, + "price": 0.1245, + "change24h": 1.25, + "liquidity": 83000.4, + "liquidityUnit": "USDC" + } + ] + } + } +} +``` + +To fetch the next page, pass the previous response's `nextCursor` back as the +`cursor` argument. `nextCursor` is `null` on the last page; `hasMore` reports +whether another page exists. + +--- + ## Live Price Subscriptions (`graphql-transport-ws`) Live price updates are streamed over WebSocket using the [`graphql-transport-ws`](https://github.com/enisdenjo/graphql-ws) subprotocol. The client must negotiate this subprotocol during the WebSocket handshake. diff --git a/schema.graphql b/schema.graphql index 350540eb..a71f3707 100644 --- a/schema.graphql +++ b/schema.graphql @@ -99,6 +99,56 @@ type OracleComparison { oracles: [OracleSource!]! } +"""Sort order for the `screener` query.""" +enum ScreenerSortOrder { + ASC + DESC +} + +""" +Filters for the `screener` query — the same filters REST GET /screener +accepts. Every field is optional; an omitted field does not filter. + +`marketCap` is declared only so it can be rejected with an explanation, +mirroring the REST route: Lens has no circulating supply data to compute a +market cap from. Passing it fails the query with UNSUPPORTED_PARAMETER +rather than being silently ignored. +""" +input ScreenerFilters { + volume: Float + change24hMin: Float + change24hMax: Float + priceMin: Float + priceMax: Float + liquidity: Float + marketCap: Float +} + +""" +One row of screener results. `liquidityUnit` is the quote asset of +`pairKey`; liquidity is summed AMM reserve value denominated in it. +""" +type ScreenerRow { + pairKey: String! + volume: Float! + price: Float! + change24h: Float! + liquidity: Float! + liquidityUnit: String! +} + +""" +One page of screener results, cursor-paginated exactly like REST +GET /screener: pass `nextCursor` back as `cursor` for the following page; +it is null on the last page. `hasMore` reports whether another page exists. +""" +type ScreenerPage { + network: String! + rows: [ScreenerRow!]! + nextCursor: String + hasMore: Boolean! +} + type Query { """ Aggregated price for a pair. Three outcomes are kept distinct: @@ -132,6 +182,20 @@ type Query { The network argument defaults to the active network. """ oracle(asset: String!, network: String): OracleComparison + + """ + The pair screener: filter, sort and cursor-paginate tracked pairs. Shares + its implementation, filters, sort fields and cursor pagination with REST + GET /screener, and resolves against the active network. + + - `sortBy` accepts the REST route's fields: volume, change_24h, price, + liquidity. Sorting on `market_cap` fails with the same explanation REST + gives (UNSUPPORTED_PARAMETER is reserved for the `marketCap` filter): + Lens has no circulating supply data to compute it from. + - `limit` is clamped to [1, 100] like REST, never + rejected. + """ + screener(filters: ScreenerFilters, sortBy: String = "volume", order: ScreenerSortOrder = DESC, cursor: String, limit: Int = 20): ScreenerPage! } type Subscription { diff --git a/src/__tests__/graphqlScreener.test.ts b/src/__tests__/graphqlScreener.test.ts new file mode 100644 index 00000000..87706081 --- /dev/null +++ b/src/__tests__/graphqlScreener.test.ts @@ -0,0 +1,184 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest' +import Fastify, { type FastifyInstance } from 'fastify' + +const { mockQuery } = vi.hoisted(() => ({ mockQuery: vi.fn() })) + +vi.mock('../db', () => ({ pgPool: { query: mockQuery }, prisma: {} })) +vi.mock('../redis', () => ({ getCachedPrice: vi.fn() })) +vi.mock('../aggregator/vwap', () => ({ getAggregatedPrice: vi.fn() })) +vi.mock('../aggregator/bestRoute', () => ({ getBestRoute: vi.fn() })) +vi.mock('../config', () => ({ config: { pairs: [] }, activeNetwork: 'testnet' })) + +import { registerGraphQL } from '../api/graphql' +import { MARKET_CAP_UNSUPPORTED } from '../services/screener' + +const SCREENER_QUERY = `query Screener( + $filters: ScreenerFilters + $sortBy: String + $order: ScreenerSortOrder + $cursor: String + $limit: Int +) { + screener(filters: $filters, sortBy: $sortBy, order: $order, cursor: $cursor, limit: $limit) { + network + hasMore + nextCursor + rows { + pairKey + volume + price + change24h + liquidity + liquidityUnit + } + } +}` + +function makeRow(overrides: Record = {}) { + return { + pair_key: 'XLM/USDC', + volume: 1000, + price: 0.12, + change_24h: 1.5, + liquidity: 50000, + ...overrides, + } +} + +function lastCall() { + const [sql, params] = mockQuery.mock.calls[mockQuery.mock.calls.length - 1] as [string, unknown[]] + return { sql, params } +} + +async function screener(app: FastifyInstance, variables: Record = {}) { + const res = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { query: SCREENER_QUERY, variables }, + }) + return res.json() +} + +describe('GraphQL screener query', () => { + let app: FastifyInstance + + beforeEach(async () => { + vi.clearAllMocks() + mockQuery.mockResolvedValue({ rows: [] }) + app = Fastify({ logger: false }) + await registerGraphQL(app) + await app.ready() + }) + + it('turns a filter into a bound WHERE condition, like the REST route', async () => { + mockQuery.mockResolvedValue({ rows: [makeRow()] }) + const json = await screener(app, { filters: { volume: 500 } }) + + expect(json.errors).toBeUndefined() + const { sql, params } = lastCall() + expect(sql).toMatch(/volume >= \$2/) + // $1 is the network; the filter follows; the page size (+1 probe row) is last. + expect(params[0]).toBe('testnet') + expect(params[1]).toBe(500) + expect(params.at(-1)).toBe(21) + + const page = json.data.screener + expect(page.network).toBe('testnet') + expect(page.rows).toEqual([ + { + pairKey: 'XLM/USDC', + volume: 1000, + price: 0.12, + change24h: 1.5, + liquidity: 50000, + liquidityUnit: 'USDC', + }, + ]) + }) + + it('sorts by the requested field and direction', async () => { + const json = await screener(app, { sortBy: 'liquidity', order: 'ASC' }) + + expect(json.errors).toBeUndefined() + expect(lastCall().sql).toMatch(/ORDER BY liquidity ASC, pair_key ASC/) + }) + + it('defaults to sorting by volume descending', async () => { + const json = await screener(app) + + expect(json.errors).toBeUndefined() + expect(lastCall().sql).toMatch(/ORDER BY volume DESC, pair_key DESC/) + }) + + it('clamps limit to 100 instead of rejecting it', async () => { + const json = await screener(app, { limit: 500 }) + + expect(json.errors).toBeUndefined() + expect(lastCall().params.at(-1)).toBe(101) + }) + + it('hits the page boundary: keeps limit rows, reports hasMore and a cursor for the last kept row', async () => { + mockQuery.mockResolvedValue({ + rows: [ + makeRow({ pair_key: 'A/X', volume: 30 }), + makeRow({ pair_key: 'B/X', volume: 20 }), + makeRow({ pair_key: 'C/X', volume: 10 }), + ], + }) + const json = await screener(app, { limit: 2 }) + + expect(json.errors).toBeUndefined() + const page = json.data.screener + expect(page.rows.map((r: { pairKey: string }) => r.pairKey)).toEqual(['A/X', 'B/X']) + expect(page.hasMore).toBe(true) + expect(JSON.parse(Buffer.from(page.nextCursor, 'base64').toString())).toEqual([20, 'B/X']) + expect(lastCall().params.at(-1)).toBe(3) + }) + + it('applies a cursor as a keyset condition on the sort field', async () => { + const cursor = Buffer.from(JSON.stringify([20, 'B/X'])).toString('base64') + const json = await screener(app, { cursor }) + + expect(json.errors).toBeUndefined() + const { sql, params } = lastCall() + expect(sql).toMatch(/\(volume, pair_key\) < \(\$2, \$3\)/) + expect(params.slice(1, 3)).toEqual([20, 'B/X']) + }) + + it('rejects a malformed cursor', async () => { + const json = await screener(app, { cursor: 'not-base64-json' }) + + expect(json.errors).toHaveLength(1) + expect(json.errors[0].message).toBe('Invalid cursor format') + expect(json.errors[0].extensions.code).toBe('INVALID_INPUT') + expect(mockQuery).not.toHaveBeenCalled() + }) + + it('rejects the marketCap filter with the same explanation REST gives', async () => { + const json = await screener(app, { filters: { marketCap: 1000000 } }) + + expect(json.errors).toHaveLength(1) + // Same constant the REST route rejects with — not a silent null. + expect(json.errors[0].message).toBe(MARKET_CAP_UNSUPPORTED) + expect(json.errors[0].extensions.code).toBe('UNSUPPORTED_PARAMETER') + expect(mockQuery).not.toHaveBeenCalled() + }) + + it('rejects sortBy: "market_cap" with the same explanation', async () => { + const json = await screener(app, { sortBy: 'market_cap' }) + + expect(json.errors).toHaveLength(1) + expect(json.errors[0].message).toBe(MARKET_CAP_UNSUPPORTED) + expect(json.errors[0].extensions.code).toBe('INVALID_INPUT') + expect(mockQuery).not.toHaveBeenCalled() + }) + + it('rejects an unknown sortBy', async () => { + const json = await screener(app, { sortBy: 'price;DROP' }) + + expect(json.errors).toHaveLength(1) + expect(json.errors[0].message).toMatch(/sortBy must be one of/) + expect(json.errors[0].extensions.code).toBe('INVALID_INPUT') + expect(mockQuery).not.toHaveBeenCalled() + }) +}) diff --git a/src/api/graphql.ts b/src/api/graphql.ts index d87f20b3..d258b158 100644 --- a/src/api/graphql.ts +++ b/src/api/graphql.ts @@ -13,6 +13,16 @@ import { parseAssetStr, getActivePairs } from '../pairsRegistry' import { reflectorOracleAdapter } from '../ingest/oracles/reflector' import type { OracleAdapter } from '../ingest/oracles/adapter' import { compareOracles } from '../services/oracleComparison' +import { + ALLOWED_SORT_FIELDS, + MARKET_CAP_CODE, + MARKET_CAP_UNSUPPORTED, + runScreener, + ScreenerCursorError, + ScreenerSortError, + SCREENER_DEFAULT_LIMIT, + SCREENER_MAX_LIMIT, +} from '../services/screener' import { priceEmitter, PRICE_PUBLISHED, type PricePublishedEvent } from '../events' import { HISTORY_INTERVAL_SECONDS, MAX_HISTORY_POINTS } from './history' import { getMaxGraphQLQueryDepth, queryDepthValidationRules } from './queryDepth' @@ -28,8 +38,14 @@ const WINDOW_SECONDS: Record = { const PRICE_TOPIC = 'PRICE_UPDATED' // Error codes that resolvers raise on purpose; their messages are written for // the caller and are safe to return as-is. Anything else is treated as an -// unexpected internal failure and masked. -const USER_FACING_CODES = new Set(['INVALID_INPUT', 'PAIR_NOT_FOUND', 'UNAUTHORIZED']) +// unexpected internal failure and masked. UNSUPPORTED_PARAMETER carries the +// screener's market_cap rejection (shared with the REST route). +const USER_FACING_CODES = new Set([ + 'INVALID_INPUT', + 'PAIR_NOT_FOUND', + 'UNAUTHORIZED', + 'UNSUPPORTED_PARAMETER', +]) const INTERNAL_ERROR_MESSAGE = 'Internal server error' function isUserFacing(err: GraphQLError): boolean { @@ -163,6 +179,58 @@ export const schema = ` oracles: [OracleSource!]! } + """ + Sort order for the \`screener\` query. + """ + enum ScreenerSortOrder { + ASC + DESC + } + + """ + Filters for the \`screener\` query — the same filters REST GET /screener + accepts. Every field is optional; an omitted field does not filter. + + \`marketCap\` is declared only so it can be rejected with an explanation, + mirroring the REST route: Lens has no circulating supply data to compute a + market cap from. Passing it fails the query with UNSUPPORTED_PARAMETER + rather than being silently ignored. + """ + input ScreenerFilters { + volume: Float + change24hMin: Float + change24hMax: Float + priceMin: Float + priceMax: Float + liquidity: Float + marketCap: Float + } + + """ + One row of screener results. \`liquidityUnit\` is the quote asset of + \`pairKey\`; liquidity is summed AMM reserve value denominated in it. + """ + type ScreenerRow { + pairKey: String! + volume: Float! + price: Float! + change24h: Float! + liquidity: Float! + liquidityUnit: String! + } + + """ + One page of screener results, cursor-paginated exactly like REST + GET /screener: pass \`nextCursor\` back as \`cursor\` for the following page; + it is null on the last page. \`hasMore\` reports whether another page exists. + """ + type ScreenerPage { + network: String! + rows: [ScreenerRow!]! + nextCursor: String + hasMore: Boolean! + } + type Query { """ Aggregated price for a pair. Three outcomes are kept distinct: @@ -201,6 +269,25 @@ export const schema = ` The network argument defaults to the active network. """ oracle(asset: String!, network: String): OracleComparison + """ + The pair screener: filter, sort and cursor-paginate tracked pairs. Shares + its implementation, filters, sort fields and cursor pagination with REST + GET /screener, and resolves against the active network. + + - \`sortBy\` accepts the REST route's fields: volume, change_24h, price, + liquidity. Sorting on \`market_cap\` fails with the same explanation REST + gives (UNSUPPORTED_PARAMETER is reserved for the \`marketCap\` filter): + Lens has no circulating supply data to compute it from. + - \`limit\` is clamped to [1, ${SCREENER_MAX_LIMIT}] like REST, never + rejected. + """ + screener( + filters: ScreenerFilters + sortBy: String = "volume" + order: ScreenerSortOrder = DESC + cursor: String + limit: Int = ${SCREENER_DEFAULT_LIMIT} + ): ScreenerPage! } type Subscription { @@ -444,6 +531,99 @@ const resolvers = { oracles: comparison.oracles.map(o => ({ ...o, status: o.status.toUpperCase() })), } }, + + async screener( + _: unknown, + { + filters, + sortBy, + order, + cursor, + limit, + }: { + filters?: { + volume?: number | null + change24hMin?: number | null + change24hMax?: number | null + priceMin?: number | null + priceMax?: number | null + liquidity?: number | null + marketCap?: number | null + } | null + sortBy?: string | null + order?: 'ASC' | 'DESC' | null + cursor?: string | null + limit?: number | null + } + ) { + // market_cap is rejected with the same explanation REST gives, rather + // than being quietly ignored: the `marketCap` filter exists only to be + // refused, and `sortBy: "market_cap"` gets the same treatment below. + if (filters?.marketCap != null) { + throw new ErrorWithProps(MARKET_CAP_UNSUPPORTED, { + code: MARKET_CAP_CODE, + field: 'marketCap', + }) + } + // Mirrors REST: sortBy is a validated string parameter, so an invalid + // value (market_cap included) is an input error carrying the + // market_cap explanation when that is the reason. + const sortField = sortBy ?? 'volume' + if (sortField === 'market_cap') { + throw new ErrorWithProps(MARKET_CAP_UNSUPPORTED, { + code: 'INVALID_INPUT', + field: 'sortBy', + }) + } + if (!(ALLOWED_SORT_FIELDS as readonly string[]).includes(sortField)) { + throw new ErrorWithProps(`sortBy must be one of: ${ALLOWED_SORT_FIELDS.join(', ')}`, { + code: 'INVALID_INPUT', + field: 'sortBy', + }) + } + try { + // Filtering, sorting, cursor pagination and the limit clamp are the + // REST route's implementation (src/services/screener.ts); this layer + // only maps GraphQL naming (camelCase) onto it. The network is the + // active one, matching every other GraphQL query here. + const page = await runScreener({ + network: activeNetwork, + filters: { + volume: filters?.volume ?? undefined, + change_24h_min: filters?.change24hMin ?? undefined, + change_24h_max: filters?.change24hMax ?? undefined, + price_min: filters?.priceMin ?? undefined, + price_max: filters?.priceMax ?? undefined, + liquidity: filters?.liquidity ?? undefined, + }, + sortBy: sortField, + order: order?.toLowerCase() ?? 'desc', + cursor, + limit: limit ?? SCREENER_DEFAULT_LIMIT, + }) + return { + network: page.network, + rows: page.data.map(row => ({ + pairKey: row.pair_key, + volume: row.volume, + price: row.price, + change24h: row.change_24h, + liquidity: row.liquidity, + liquidityUnit: row.liquidity_unit, + })), + nextCursor: page.nextCursor, + hasMore: page.hasMore, + } + } catch (error) { + if (error instanceof ScreenerCursorError || error instanceof ScreenerSortError) { + throw new ErrorWithProps(error.message, { + code: 'INVALID_INPUT', + field: error instanceof ScreenerCursorError ? 'cursor' : 'sortBy', + }) + } + throw error + } + }, }, Subscription: { diff --git a/src/routes/screener.ts b/src/routes/screener.ts index f14a4c53..813be4c2 100644 --- a/src/routes/screener.ts +++ b/src/routes/screener.ts @@ -1,49 +1,34 @@ import type { FastifyInstance } from 'fastify' import { z } from 'zod' -import { pgPool } from '../db' import { activeNetwork } from '../config' import { formatValidationError, VALIDATION_ERROR_CODE } from '../validation' +import { + ALLOWED_SORT_FIELDS, + MARKET_CAP_CODE, + MARKET_CAP_UNSUPPORTED, + runScreener, + ScreenerCursorError, + SCREENER_DEFAULT_LIMIT, +} from '../services/screener' import '../middleware/network' // declares req.network on the FastifyRequest type -// Lens has no circulating-supply data, so a market cap cannot be computed. It -// used to be reported as a copy of `liquidity`; a caller who sorts or filters on -// it must be told it is gone rather than have the parameter silently ignored. -const MARKET_CAP_UNSUPPORTED = - 'market_cap is not supported: Lens has no circulating supply data to compute it from. Use liquidity instead' - -/** - * Liquidity Definition for /screener: - * - Sources/Venues: Stellar AMM liquidity pools (source = 'AMM' in price_points, - * mapped to pool_snapshots for the pair). - * - Valuation & Formula: Summed across all active AMM pools for the pair of - * `(reserve_a * spot_price + reserve_b)`. Since spot_price is quoted in - * `asset_b` per unit of `asset_a` (i.e. reserve_b / reserve_a), `reserve_a * spot_price` - * evaluates to the value of reserve_a denominated in asset_b. - * - Unit: Denominated in the quote asset (counter asset, `asset_b`, the second asset - * in `pair_key` "BASE/QUOTE"). Named explicitly as `liquidity_unit` in each row. - * - Prices: Spot price from the latest snapshot of each liquidity pool on the requested network. - */ - -/** Machine-readable code for the market_cap rejection (not a validation error). */ -const MARKET_CAP_CODE = 'UNSUPPORTED_PARAMETER' - -const ALLOWED_SORT_FIELDS = ['volume', 'change_24h', 'price', 'liquidity'] as const +// The query execution (filters, sorting, cursor pagination, the liquidity +// definition and the [1, 100] limit clamp) lives in src/services/screener.ts +// and is shared with the GraphQL `screener` query; this route owns only the +// HTTP concerns: query-string validation and error shaping. const numFilter = (name: string) => z.coerce.number({ error: `${name} must be a valid number` }).optional() -export const SCREENER_DEFAULT_LIMIT = 20 -export const SCREENER_MAX_LIMIT = 100 - const ScreenerQuerySchema = z .object({ - // Historical behavior: limit is clamped to [1, 100], never rejected. - // `GET /screener?limit=500` returns 100 rows, `limit=0` behaves as 1. + // Historical behavior: limit is clamped to [1, 100] inside runScreener, + // never rejected. `GET /screener?limit=500` returns 100 rows, `limit=0` + // behaves as 1. limit: z.coerce .number({ error: 'limit must be a valid integer' }) .int({ error: 'limit must be a valid integer' }) - .default(SCREENER_DEFAULT_LIMIT) - .transform((v) => Math.min(SCREENER_MAX_LIMIT, Math.max(1, v))), + .default(SCREENER_DEFAULT_LIMIT), sortBy: z.string().default('volume'), order: z .preprocess( @@ -122,142 +107,28 @@ export async function registerScreenerRoutes(app: FastifyInstance) { ) } const { limit, sortBy, order, cursor } = parsed.data - const filters = { - volume: parsed.data.volume, - change_24h_min: parsed.data.change_24h_min, - change_24h_max: parsed.data.change_24h_max, - price_min: parsed.data.price_min, - price_max: parsed.data.price_max, - liquidity: parsed.data.liquidity, - } - - // 3. Build query parameters and WHERE conditions - const conditions: string[] = [] - // Every CTE below is scoped to one network: testnet and mainnet rows share - // these tables, and pooling them would rank a pair on the other chain's data. - const params: any[] = [req.network ?? activeNetwork] - let paramIdx = 2 - - if (filters.volume !== undefined) { - conditions.push(`volume >= $${paramIdx++}`) - params.push(filters.volume) - } - if (filters.change_24h_min !== undefined) { - conditions.push(`change_24h >= $${paramIdx++}`) - params.push(filters.change_24h_min) - } - if (filters.change_24h_max !== undefined) { - conditions.push(`change_24h <= $${paramIdx++}`) - params.push(filters.change_24h_max) - } - if (filters.price_min !== undefined) { - conditions.push(`price >= $${paramIdx++}`) - params.push(filters.price_min) - } - if (filters.price_max !== undefined) { - conditions.push(`price <= $${paramIdx++}`) - params.push(filters.price_max) - } - if (filters.liquidity !== undefined) { - conditions.push(`liquidity >= $${paramIdx++}`) - params.push(filters.liquidity) - } - // 4. Cursor decoding and pagination condition - if (cursor) { - try { - const decoded = Buffer.from(cursor, 'base64').toString('utf-8') - const [lastVal, lastId] = JSON.parse(decoded) - - if (lastVal !== null && lastVal !== undefined && lastId) { - const op = order === 'desc' ? '<' : '>' - conditions.push(`(${sortBy}, pair_key) ${op} ($${paramIdx++}, $${paramIdx++})`) - params.push(lastVal, lastId) - } - } catch (e) { - return reply.status(400).send({ error: 'Invalid cursor format', code: 'VALIDATION_ERROR' }) - } - } - - const whereClause = conditions.length > 0 ? `WHERE ${conditions.join(' AND ')}` : '' - - // 5. Construct the full SQL query using CTEs - const sql = ` - WITH pair_pools AS ( - SELECT DISTINCT pair_key, pool_id - FROM price_points - WHERE network = $1 AND source = 'AMM' AND pool_id IS NOT NULL - ), - latest_snapshots AS ( - SELECT DISTINCT ON (ps.pool_id) - ps.pool_id, - (ps.reserve_a * ps.spot_price + ps.reserve_b) AS pool_liq - FROM pool_snapshots ps - WHERE ps.network = $1 - ORDER BY ps.pool_id, ps.timestamp DESC - ), - pair_liquidity AS ( - SELECT pp.pair_key, SUM(ls.pool_liq) AS liquidity - FROM pair_pools pp - JOIN latest_snapshots ls ON ls.pool_id = pp.pool_id - GROUP BY pp.pair_key - ), - latest_aggregates AS ( - SELECT DISTINCT ON (pair_key) - pair_key, - COALESCE(volume::float, 0) AS volume, - COALESCE(close_price::float, 0) AS price, - CASE - WHEN open_price > 0 THEN ((close_price::float - open_price::float) / open_price::float) * 100 - ELSE 0 - END AS change_24h - FROM price_aggregates - WHERE network = $1 AND window = '24h' - ORDER BY pair_key, bucket DESC - ), - screener_data AS ( - SELECT - la.pair_key, - la.volume, - la.price, - la.change_24h, - COALESCE(pl.liquidity::float, 0) AS liquidity - FROM latest_aggregates la - LEFT JOIN pair_liquidity pl ON pl.pair_key = la.pair_key - ) - SELECT * FROM screener_data - ${whereClause} - ORDER BY ${sortBy} ${order.toUpperCase()}, pair_key ${order.toUpperCase()} - LIMIT $${paramIdx} - ` - - // 6. Execute query - params.push(limit + 1) // Fetch one extra to determine hasMore - const result = await pgPool.query(sql, params) - const rows = result.rows - - const hasMore = rows.length > limit - const data = hasMore ? rows.slice(0, limit) : rows - - // 7. Generate nextCursor - let nextCursor = null - if (hasMore && data.length > 0) { - const lastRecord = data[data.length - 1] - const cursorPayload = JSON.stringify([lastRecord[sortBy], lastRecord.pair_key]) - nextCursor = Buffer.from(cursorPayload).toString('base64') - } - - return { + return await runScreener({ network: req.network ?? activeNetwork, - data: data.map((row) => ({ - ...row, - liquidity_unit: row.pair_key.split('/')[1] ?? 'quote', - })), - nextCursor, - hasMore, - total: null, - } + filters: { + volume: parsed.data.volume, + change_24h_min: parsed.data.change_24h_min, + change_24h_max: parsed.data.change_24h_max, + price_min: parsed.data.price_min, + price_max: parsed.data.price_max, + liquidity: parsed.data.liquidity, + }, + sortBy, + order, + cursor, + limit, + }) } catch (error) { + // A malformed cursor is the caller's mistake and gets the same 400 body + // the route has always returned for it. + if (error instanceof ScreenerCursorError) { + return reply.status(400).send({ error: error.message, code: error.code }) + } log.error({ err: error }, 'screener query failed') return reply.status(500).send({ error: 'Internal server error' }) } diff --git a/src/services/screener.ts b/src/services/screener.ts new file mode 100644 index 00000000..c8dd284d --- /dev/null +++ b/src/services/screener.ts @@ -0,0 +1,245 @@ +import { pgPool } from '../db' + +// Lens has no circulating-supply data, so a market cap cannot be computed. It +// used to be reported as a copy of `liquidity`; a caller who sorts or filters on +// it must be told it is gone rather than have the parameter silently ignored. +// Both the REST route and the GraphQL `screener` query reject it with this +// exact explanation. +export const MARKET_CAP_UNSUPPORTED = + 'market_cap is not supported: Lens has no circulating supply data to compute it from. Use liquidity instead' + +/** Machine-readable code for the market_cap filter rejection (not a validation error). */ +export const MARKET_CAP_CODE = 'UNSUPPORTED_PARAMETER' + +export const ALLOWED_SORT_FIELDS = ['volume', 'change_24h', 'price', 'liquidity'] as const + +export const SCREENER_DEFAULT_LIMIT = 20 +export const SCREENER_MAX_LIMIT = 100 + +export type ScreenerSortField = (typeof ALLOWED_SORT_FIELDS)[number] +export type ScreenerOrder = 'asc' | 'desc' + +/** + * Numeric filters, one per REST query parameter. Each is a lower bound unless + * the name says otherwise; omitted fields do not filter. + */ +export interface ScreenerFilters { + volume?: number + change_24h_min?: number + change_24h_max?: number + price_min?: number + price_max?: number + liquidity?: number +} + +/** One screener row as produced by the SQL (snake_case, straight from the DB). */ +export interface ScreenerRow { + pair_key: string + volume: number + price: number + change_24h: number + liquidity: number +} + +export interface ScreenerPage { + network: string + data: Array + nextCursor: string | null + hasMore: boolean + total: null +} + +/** + * The cursor is not decodable base64 JSON. Each surface maps this to its own + * error shape: the REST route answers 400, the GraphQL resolver raises + * INVALID_INPUT — both with the same message. + */ +export class ScreenerCursorError extends Error { + readonly code = 'VALIDATION_ERROR' as const + constructor() { + super('Invalid cursor format') + this.name = 'ScreenerCursorError' + } +} + +/** + * Raised at the SQL boundary when sortBy/order are not in the allowlist. + * Both callers validate these earlier and produce a friendlier 4xx; this class + * exists so an unvalidated value can never be interpolated into SQL. + */ +export class ScreenerSortError extends Error { + constructor(message: string) { + super(message) + this.name = 'ScreenerSortError' + } +} + +/** + * Liquidity Definition for the screener (REST GET /screener and GraphQL + * `screener`): + * - Sources/Venues: Stellar AMM liquidity pools (source = 'AMM' in price_points, + * mapped to pool_snapshots for the pair). + * - Valuation & Formula: Summed across all active AMM pools for the pair of + * `(reserve_a * spot_price + reserve_b)`. Since spot_price is quoted in + * `asset_b` per unit of `asset_a` (i.e. reserve_b / reserve_a), `reserve_a * spot_price` + * evaluates to the value of reserve_a denominated in asset_b. + * - Unit: Denominated in the quote asset (counter asset, `asset_b`, the second asset + * in `pair_key` "BASE/QUOTE"). Named explicitly as `liquidity_unit` in each row. + * - Prices: Spot price from the latest snapshot of each liquidity pool on the requested network. + */ + +/** + * The one screener query implementation, shared by REST and GraphQL. + * + * `limit` is clamped to [1, SCREENER_MAX_LIMIT], never rejected — historical + * REST behaviour (`?limit=500` returns 100 rows, `limit=0` behaves as 1) that + * GraphQL inherits. `sortBy` and `order` are allowlisted here as the last line + * of defence before SQL interpolation; callers validate them earlier to give + * callers a proper 4xx. + */ +export async function runScreener(opts: { + network: string + filters?: ScreenerFilters + sortBy?: string + order?: string + cursor?: string | null + limit?: number +}): Promise { + const { network, filters, cursor } = opts + const limit = Math.min(SCREENER_MAX_LIMIT, Math.max(1, opts.limit ?? SCREENER_DEFAULT_LIMIT)) + const sortBy = opts.sortBy ?? 'volume' + if (!(ALLOWED_SORT_FIELDS as readonly string[]).includes(sortBy)) { + throw new ScreenerSortError(`sortBy must be one of: ${ALLOWED_SORT_FIELDS.join(', ')}`) + } + const order = (opts.order ?? 'desc').toLowerCase() + if (order !== 'asc' && order !== 'desc') { + throw new ScreenerSortError('order must be one of: asc, desc') + } + + // Build query parameters and WHERE conditions + const conditions: string[] = [] + // Every CTE below is scoped to one network: testnet and mainnet rows share + // these tables, and pooling them would rank a pair on the other chain's data. + const params: any[] = [network] + let paramIdx = 2 + + if (filters?.volume !== undefined) { + conditions.push(`volume >= $${paramIdx++}`) + params.push(filters.volume) + } + if (filters?.change_24h_min !== undefined) { + conditions.push(`change_24h >= $${paramIdx++}`) + params.push(filters.change_24h_min) + } + if (filters?.change_24h_max !== undefined) { + conditions.push(`change_24h <= $${paramIdx++}`) + params.push(filters.change_24h_max) + } + if (filters?.price_min !== undefined) { + conditions.push(`price >= $${paramIdx++}`) + params.push(filters.price_min) + } + if (filters?.price_max !== undefined) { + conditions.push(`price <= $${paramIdx++}`) + params.push(filters.price_max) + } + if (filters?.liquidity !== undefined) { + conditions.push(`liquidity >= $${paramIdx++}`) + params.push(filters.liquidity) + } + + // Cursor decoding and pagination condition + if (cursor) { + try { + const decoded = Buffer.from(cursor, 'base64').toString('utf-8') + const [lastVal, lastId] = JSON.parse(decoded) + + if (lastVal !== null && lastVal !== undefined && lastId) { + const op = order === 'desc' ? '<' : '>' + conditions.push(`(${sortBy}, pair_key) ${op} ($${paramIdx++}, $${paramIdx++})`) + params.push(lastVal, lastId) + } + } catch { + throw new ScreenerCursorError() + } + } + + const whereClause = conditions.length > 0 ? `WHERE ${conditions.join(' AND ')}` : '' + + // Construct the full SQL query using CTEs + const sql = ` + WITH pair_pools AS ( + SELECT DISTINCT pair_key, pool_id + FROM price_points + WHERE network = $1 AND source = 'AMM' AND pool_id IS NOT NULL + ), + latest_snapshots AS ( + SELECT DISTINCT ON (ps.pool_id) + ps.pool_id, + (ps.reserve_a * ps.spot_price + ps.reserve_b) AS pool_liq + FROM pool_snapshots ps + WHERE ps.network = $1 + ORDER BY ps.pool_id, ps.timestamp DESC + ), + pair_liquidity AS ( + SELECT pp.pair_key, SUM(ls.pool_liq) AS liquidity + FROM pair_pools pp + JOIN latest_snapshots ls ON ls.pool_id = pp.pool_id + GROUP BY pp.pair_key + ), + latest_aggregates AS ( + SELECT DISTINCT ON (pair_key) + pair_key, + COALESCE(volume::float, 0) AS volume, + COALESCE(close_price::float, 0) AS price, + CASE + WHEN open_price > 0 THEN ((close_price::float - open_price::float) / open_price::float) * 100 + ELSE 0 + END AS change_24h + FROM price_aggregates + WHERE network = $1 AND window = '24h' + ORDER BY pair_key, bucket DESC + ), + screener_data AS ( + SELECT + la.pair_key, + la.volume, + la.price, + la.change_24h, + COALESCE(pl.liquidity::float, 0) AS liquidity + FROM latest_aggregates la + LEFT JOIN pair_liquidity pl ON pl.pair_key = la.pair_key + ) + SELECT * FROM screener_data + ${whereClause} + ORDER BY ${sortBy} ${order.toUpperCase()}, pair_key ${order.toUpperCase()} + LIMIT $${paramIdx} + ` + + // Execute query + params.push(limit + 1) // Fetch one extra to determine hasMore + const result = await pgPool.query(sql, params) + const rows = result.rows as ScreenerRow[] + + const hasMore = rows.length > limit + const data = hasMore ? rows.slice(0, limit) : rows + + // Generate nextCursor + let nextCursor = null + if (hasMore && data.length > 0) { + const lastRecord = data[data.length - 1] + const cursorPayload = JSON.stringify([lastRecord[sortBy as ScreenerSortField], lastRecord.pair_key]) + nextCursor = Buffer.from(cursorPayload).toString('base64') + } + + return { + network, + data: data.map((row) => ({ + ...row, + liquidity_unit: row.pair_key.split('/')[1] ?? 'quote', + })), + nextCursor, + hasMore, + total: null, + } +} diff --git a/tests/dataQuality.test.ts b/tests/dataQuality.test.ts index 094dca66..c2853555 100644 --- a/tests/dataQuality.test.ts +++ b/tests/dataQuality.test.ts @@ -26,7 +26,7 @@ import { VWAP_WINDOW_DEFAULT_MINUTES, } from '../src/routes/price' import { HISTORY_DEFAULT_LIMIT, HISTORY_MAX_LIMIT } from '../src/api/rest' -import { SCREENER_DEFAULT_LIMIT, SCREENER_MAX_LIMIT } from '../src/routes/screener' +import { SCREENER_DEFAULT_LIMIT, SCREENER_MAX_LIMIT } from '../src/services/screener' import { MIN_OBSERVATIONS as SPREAD_MIN_OBSERVATIONS } from '../src/routes/spreads' import { SNAPSHOT_RETENTION_DAYS, SNAPSHOT_PRUNE_BATCH_SIZE } from '../src/jobs/snapshotRetention' import { DEFAULT_RETENTION_DAYS as PARTITION_RETENTION_DAYS } from '../src/partitioning/pricePoints'