Add GraphQL screener query using shared REST filtering and pagination logic

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 |

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.

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 {

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<string, unknown> = {}) {
+  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<string, unknown> = {}) {
+  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()
+  })
+})