diff --git a/docs/graphql.md b/docs/graphql.md index 021b0d8f..e0e99562 100644 --- a/docs/graphql.md +++ b/docs/graphql.md @@ -14,6 +14,7 @@ Lens exposes a high-performance GraphQL API alongside its REST interface, powere - [`getPriceHistory`](#3-getpricehistory) - [`listPairs`](#4-listpairs) - [`screener`](#5-screener) + - [`volumes` and `spreads`](#6-volumes-and-spreads) 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) @@ -71,6 +72,8 @@ type Query { 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! + volumes(asset: String!, window: String, network: String): VolumeReport! + spreads(asset: String!, window: String, network: String): SpreadReport! } type Subscription { @@ -452,6 +455,65 @@ whether another page exists. --- +### 6. `volumes` and `spreads` + +GraphQL counterparts of REST `GET /volumes/:asset` and `GET /spreads/:asset`. +Each resolver calls the REST route's own implementation (`computeVolumes` in +`src/routes/volumes.ts`, `computeSpreads` in `src/routes/spreads.ts`), so the +figures cannot drift between the two surfaces. + +| Argument | `volumes` | `spreads` | +|---|---|---| +| `asset` | required, e.g. `XLM` | required, e.g. `XLM` | +| `window` | `24h` (default), `7d`, `30d` | `5m` (default), `15m`, `1h`, `24h` | +| `network` | `testnet` / `mainnet`, defaults to the active network | `testnet` / `mainnet`, defaults to the active network | + +An unrecognised `window` or `network` fails with `INVALID_INPUT` and the same +message REST returns with its 400. The network answered for is echoed in +`network`; the two chains are never pooled. + +Per-venue results are a typed list of `VenueQuote`, so a client can select +just the fields it wants per venue: + +- `volumes.venues`: `venue`, `volume` (in the asset's own units, summed across + pairs) and `aggregator`. This is REST's `byVenue` map as a list, in REST's + `venues` order. Every other REST field (`totals`, `netVolume`, + `grossVolume`, `totalVolume`, trade counts, trade-size statistics, `status`, + `provenance`) is returned under the same name. +- `spreads.venues`: one entry per (venue, pair), tightest first, with `pairKey`, + `bid`, `ask`, `mid`, `spreadBps`, `spreadPct`, `observations` and `lastSeen`. + `tightest` is the first of them, or `null` when no venue quoted the asset. + +Fields that belong to the other query are `null` on a `VenueQuote`. + +#### Query Definition +```graphql +query VenueViews($asset: String!, $network: String) { + volumes(asset: $asset, window: "24h", network: $network) { + network + totals { net gross } + venues { venue volume aggregator } + } + spreads(asset: $asset, window: "5m", network: $network) { + network + tightest { venue pairKey spreadBps } + venues { venue pairKey bid ask spreadBps } + } +} +``` + +#### Runnable `curl` Command +```bash +curl -s -X POST http://localhost:3002/graphql \ + -H "Content-Type: application/json" \ + -d '{ + "query": "query VenueViews($asset: String!, $network: String) { volumes(asset: $asset, network: $network) { network totals { net gross } venues { venue volume aggregator } } spreads(asset: $asset, network: $network) { network tightest { venue pairKey spreadBps } } }", + "variables": { "asset": "XLM", "network": "testnet" } + }' +``` + +--- + ## 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 a71f3707..879c645c 100644 --- a/schema.graphql +++ b/schema.graphql @@ -149,6 +149,117 @@ type ScreenerPage { hasMore: Boolean! } +""" +One venue's figures in a `volumes` or `spreads` result. Both queries +return venues as a list of this type so a caller can select only the fields +it wants per venue. + +Which fields are set depends on the query, and mirrors the matching REST +route field for field: + +- `volumes`: `venue`, `volume` (the asset's traded volume on that venue, + in the asset's own units, summed across every pair) and `aggregator`. + The spread fields are null. +- `spreads`: one entry per (venue, pair): `venue`, `pairKey`, `bid`, + `ask`, `mid`, `spreadBps`, `spreadPct`, `observations`, + `lastSeen`. `volume` is null. +""" +type VenueQuote { + venue: String! + + """ + Whether the venue is an aggregator re-reporting primary-market trades (volumes only). + """ + aggregator: Boolean + + """Traded volume in the asset's own units over the window (volumes only).""" + volume: Float + + """ + The pair this quote is for (spreads only): two pairs on one venue are different books. + """ + pairKey: String + + """Lowest print in the window (spreads only).""" + bid: Float + + """Highest print in the window (spreads only).""" + ask: Float + mid: Float + + """(ask - bid) / mid * 10000 (spreads only).""" + spreadBps: Float + + """spreadBps / 100 (spreads only).""" + spreadPct: Float + observations: Int + lastSeen: String +} + +""" +Labelled net (primary markets only) and gross (including aggregators) volume. +""" +type VolumeTotals { + net: Float! + gross: Float! +} + +type VolumeProvenance { + venues: [String!]! + sampleCount: Int! + observedSampleCount: Int! + latestTradeAt: String + dataAgeSeconds: Int +} + +""" +Cross-venue traded volume for an asset: the same figures as REST +GET /volumes/:asset, with the per-venue breakdown (`byVenue` on REST) as a +typed `venues` list. +""" +type VolumeReport { + asset: String! + network: String! + window: String! + startTime: String! + endTime: String! + totals: VolumeTotals! + netVolume: Float! + grossVolume: Float! + + """Alias for netVolume, as on REST.""" + totalVolume: Float! + + """Per-venue volume, ordered by venue name like REST's `venues`.""" + venues: [VenueQuote!]! + tradeCount: Int! + netTradeCount: Int! + grossTradeCount: Int! + medianTradeSize: Float + topTradeVolumeShare: Float + volumeConcentration: Float + concentrationMeasure: String! + + """ok, no_trades or no_positive_volume.""" + status: String! + provenance: VolumeProvenance! +} + +""" +Per-(venue, pair) spreads for an asset: the same figures as REST +GET /spreads/:asset. `venues` is tightest first; `tightest` is null when no +venue quoted the asset in the window. +""" +type SpreadReport { + asset: String! + network: String! + window: String! + startTime: String! + endTime: String! + venues: [VenueQuote!]! + tightest: VenueQuote +} + type Query { """ Aggregated price for a pair. Three outcomes are kept distinct: @@ -196,6 +307,22 @@ type Query { rejected. """ screener(filters: ScreenerFilters, sortBy: String = "volume", order: ScreenerSortOrder = DESC, cursor: String, limit: Int = 20): ScreenerPage! + + """ + Cross-venue traded volume for an asset. Shares its implementation with + REST GET /volumes/:asset: same windows (24h, 7d, 30d; default 24h), same + figures. `network` (testnet or mainnet) defaults to the active network; + any other value fails with INVALID_INPUT, as REST answers 400. + """ + volumes(asset: String!, window: String, network: String): VolumeReport! + + """ + Per-(venue, pair) bid/ask spreads for an asset. Shares its implementation + with REST GET /spreads/:asset: same windows (5m, 15m, 1h, 24h; default 5m), + same figures. `network` (testnet or mainnet) defaults to the active + network; any other value fails with INVALID_INPUT, as REST answers 400. + """ + spreads(asset: String!, window: String, network: String): SpreadReport! } type Subscription { diff --git a/src/__tests__/graphqlVolumesSpreads.test.ts b/src/__tests__/graphqlVolumesSpreads.test.ts new file mode 100644 index 00000000..c253b1ef --- /dev/null +++ b/src/__tests__/graphqlVolumesSpreads.test.ts @@ -0,0 +1,267 @@ +import { describe, it, expect, vi, beforeEach, afterEach } 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 { registerVolumeRoutes } from '../routes/volumes' +import { registerSpreadsRoutes } from '../routes/spreads' + +// #368: the GraphQL `volumes` and `spreads` queries share the REST routes' +// implementation. These tests drive both surfaces of one server with the same +// database rows and assert they answer alike. + +const VOLUMES_QUERY = `query Volumes($asset: String!, $window: String, $network: String) { + volumes(asset: $asset, window: $window, network: $network) { + asset + network + window + startTime + endTime + totals { net gross } + netVolume + grossVolume + totalVolume + venues { venue volume aggregator bid spreadBps } + tradeCount + netTradeCount + grossTradeCount + medianTradeSize + topTradeVolumeShare + volumeConcentration + concentrationMeasure + status + provenance { venues sampleCount observedSampleCount latestTradeAt dataAgeSeconds } + } +}` + +const SPREADS_QUERY = `query Spreads($asset: String!, $window: String, $network: String) { + spreads(asset: $asset, window: $window, network: $network) { + asset + network + window + startTime + endTime + venues { venue pairKey bid ask mid spreadBps spreadPct observations lastSeen volume } + tightest { venue pairKey bid ask mid spreadBps spreadPct observations lastSeen } + } +}` + +/** One aggregate row as the volumes query returns it. */ +const VOLUME_ROW = { + total_volume: '160', + by_venue: { amm: 49.5, aquarius: 10, sdex: 100.5 }, + venues: ['amm', 'aquarius', 'sdex'], + trade_count_by_venue: { amm: 2, aquarius: 1, sdex: 3 }, + trade_count: 6, + observed_sample_count: 7, + median_trade_size: '25', + top_trade_volume_share: '0.5', + volume_concentration: '0.3', + latest_trade_at: new Date('2026-09-01T12:00:00.000Z'), +} + +/** One grouped row as the spreads query returns it. */ +function spreadRow(source: string, pairKey: string, bid: string, ask: string, observations = 10) { + return { + source, + pair_key: pairKey, + bid, + ask, + observations, + last_seen: new Date('2026-09-01T12:00:00.000Z'), + } +} + +const SPREAD_ROWS = [ + spreadRow('sdex', 'XLM/USDC', '90', '110'), + spreadRow('amm', 'XLM/USDC', '99.9', '100.1'), + spreadRow('sdex', 'XLM/EURC', '0.90', '0.92'), + spreadRow('soroswap', 'XLM/USDC', '1', '1', 1), // dropped: single observation +] + +/** Removes the per-request clock fields, which legitimately differ between two calls. */ +function withoutClock>(body: T) { + const { startTime, endTime, provenance, ...rest } = body as Record + return { + ...rest, + ...(provenance ? { provenance: { ...provenance, dataAgeSeconds: undefined } } : {}), + } +} + +async function gql(app: FastifyInstance, query: string, variables: Record) { + const res = await app.inject({ method: 'POST', url: '/graphql', payload: { query, variables } }) + return res.json() +} + +describe('GraphQL volumes and spreads queries (#368)', () => { + let app: FastifyInstance + + beforeEach(async () => { + vi.clearAllMocks() + app = Fastify({ logger: false }) + await registerVolumeRoutes(app) + await registerSpreadsRoutes(app) + await registerGraphQL(app) + await app.ready() + }) + + afterEach(async () => { + await app.close() + }) + + describe('volumes', () => { + beforeEach(() => { + mockQuery.mockResolvedValue({ rows: [VOLUME_ROW] }) + }) + + it('matches REST GET /volumes/:asset for the same input', async () => { + const rest = (await app.inject({ method: 'GET', url: '/volumes/XLM?window=7d' })).json() + const body = await gql(app, VOLUMES_QUERY, { asset: 'XLM', window: '7d' }) + + expect(body.errors).toBeUndefined() + const g = body.data.volumes + + // Every scalar REST reports, GraphQL reports identically. + const { byVenue, venues: restVenueNames, ...restScalars } = withoutClock(rest) as any + const { venues: gqlVenues, ...gqlScalars } = withoutClock(g) as any + expect(gqlScalars).toEqual(restScalars) + + // The per-venue map becomes a typed list, in REST's venue order. + expect(gqlVenues.map((v: any) => v.venue)).toEqual(restVenueNames) + for (const v of gqlVenues) { + expect(v.volume).toBe(byVenue[v.venue]) + } + // Net excludes the aggregator on both surfaces. + expect(g.totals).toEqual({ net: 150, gross: 160 }) + expect(gqlVenues.find((v: any) => v.venue === 'aquarius').aggregator).toBe(true) + expect(gqlVenues.find((v: any) => v.venue === 'sdex').aggregator).toBe(false) + // Spread-only fields are null on a volume quote. + expect(gqlVenues[0].bid).toBeNull() + expect(gqlVenues[0].spreadBps).toBeNull() + }) + + it('runs the same SQL with the same parameters as REST', async () => { + await app.inject({ method: 'GET', url: '/volumes/XLM?network=mainnet' }) + await gql(app, VOLUMES_QUERY, { asset: 'XLM', network: 'mainnet' }) + + const [restSql, restParams] = mockQuery.mock.calls[0] as [string, unknown[]] + const [gqlSql, gqlParams] = mockQuery.mock.calls[1] as [string, unknown[]] + expect(gqlSql).toBe(restSql) + expect(gqlParams.slice(0, 2)).toEqual(['XLM', 'mainnet']) + expect(gqlParams.slice(0, 2)).toEqual(restParams.slice(0, 2)) + }) + + it('defaults to the active network and echoes it', async () => { + const body = await gql(app, VOLUMES_QUERY, { asset: 'XLM' }) + expect(body.data.volumes.network).toBe('testnet') + expect(body.data.volumes.window).toBe('24h') + expect((mockQuery.mock.calls[0] as [string, unknown[]])[1][1]).toBe('testnet') + }) + + it('rejects an unknown network with INVALID_INPUT, as REST answers 400', async () => { + const rest = await app.inject({ method: 'GET', url: '/volumes/XLM?network=devnet' }) + const body = await gql(app, VOLUMES_QUERY, { asset: 'XLM', network: 'devnet' }) + + expect(rest.statusCode).toBe(400) + expect(body.data).toBeNull() + expect(body.errors[0].extensions.code).toBe('INVALID_INPUT') + expect(body.errors[0].message).toBe(rest.json().error) + expect(mockQuery).not.toHaveBeenCalled() + }) + + it('rejects an unknown window with INVALID_INPUT, as REST answers 400', async () => { + const rest = await app.inject({ method: 'GET', url: '/volumes/XLM?window=1y' }) + const body = await gql(app, VOLUMES_QUERY, { asset: 'XLM', window: '1y' }) + + expect(rest.statusCode).toBe(400) + expect(body.errors[0].extensions.code).toBe('INVALID_INPUT') + expect(body.errors[0].message).toBe(rest.json().error) + }) + + it('masks a database failure instead of leaking it', async () => { + mockQuery.mockRejectedValue(new Error('connection refused to 10.0.0.5')) + const body = await gql(app, VOLUMES_QUERY, { asset: 'XLM' }) + expect(body.errors[0].extensions.code).toBe('INTERNAL_SERVER_ERROR') + expect(JSON.stringify(body)).not.toMatch(/connection refused/) + }) + }) + + describe('spreads', () => { + beforeEach(() => { + mockQuery.mockResolvedValue({ rows: SPREAD_ROWS }) + }) + + it('matches REST GET /spreads/:asset for the same input', async () => { + const rest = (await app.inject({ method: 'GET', url: '/spreads/XLM?window=1h' })).json() + const body = await gql(app, SPREADS_QUERY, { asset: 'XLM', window: '1h' }) + + expect(body.errors).toBeUndefined() + const g = body.data.spreads + const { venues, tightest, ...scalars } = withoutClock(g) as any + const { venues: restVenues, tightest: restTightest, ...restScalars } = withoutClock(rest) as any + + expect(scalars).toEqual(restScalars) + // A typed list of venue quotes, same entries and order (tightest first). + expect(venues.map(({ volume, ...v }: any) => v)).toEqual(restVenues) + expect(venues.every((v: any) => v.volume === null)).toBe(true) + expect(tightest).toEqual(restTightest) + expect(tightest.venue).toBe('amm') + // The single-observation venue is dropped on both surfaces. + expect(venues.some((v: any) => v.venue === 'soroswap')).toBe(false) + }) + + it('keeps (venue, pair) grouping: two pairs on one venue stay separate', async () => { + const body = await gql(app, SPREADS_QUERY, { asset: 'XLM' }) + const sdex = body.data.spreads.venues.filter((v: any) => v.venue === 'sdex') + expect(sdex.map((v: any) => v.pairKey).sort()).toEqual(['XLM/EURC', 'XLM/USDC']) + }) + + it('is network-aware: passes the requested network to the query', async () => { + const body = await gql(app, SPREADS_QUERY, { asset: 'XLM', network: 'mainnet' }) + expect(body.data.spreads.network).toBe('mainnet') + expect((mockQuery.mock.calls[0] as [string, unknown[]])[1][1]).toBe('mainnet') + }) + + it('defaults to the active network and the 5m window', async () => { + const body = await gql(app, SPREADS_QUERY, { asset: 'XLM' }) + expect(body.data.spreads.network).toBe('testnet') + expect(body.data.spreads.window).toBe('5m') + }) + + it('returns tightest: null, not a zeroed quote, when nobody quoted the asset', async () => { + mockQuery.mockResolvedValue({ rows: [] }) + const rest = (await app.inject({ method: 'GET', url: '/spreads/XLM' })).json() + const body = await gql(app, SPREADS_QUERY, { asset: 'XLM' }) + expect(rest.tightest).toBeNull() + expect(body.data.spreads.tightest).toBeNull() + expect(body.data.spreads.venues).toEqual([]) + }) + + it('rejects an unknown network or window with INVALID_INPUT, as REST answers 400', async () => { + for (const [qs, variables] of [ + ['network=devnet', { network: 'devnet' }], + ['window=7d', { window: '7d' }], + ] as const) { + const rest = await app.inject({ method: 'GET', url: `/spreads/XLM?${qs}` }) + const body = await gql(app, SPREADS_QUERY, { asset: 'XLM', ...variables }) + expect(rest.statusCode).toBe(400) + expect(body.errors[0].extensions.code).toBe('INVALID_INPUT') + expect(body.errors[0].message).toBe(rest.json().error) + } + expect(mockQuery).not.toHaveBeenCalled() + }) + + it('rejects an empty asset with INVALID_INPUT', async () => { + const body = await gql(app, SPREADS_QUERY, { asset: '' }) + expect(body.errors[0].extensions.code).toBe('INVALID_INPUT') + expect(mockQuery).not.toHaveBeenCalled() + }) + }) +}) diff --git a/src/api/graphql.ts b/src/api/graphql.ts index d258b158..bd98ca66 100644 --- a/src/api/graphql.ts +++ b/src/api/graphql.ts @@ -23,6 +23,9 @@ import { SCREENER_DEFAULT_LIMIT, SCREENER_MAX_LIMIT, } from '../services/screener' +import { computeVolumes, VolumesInputError } from '../routes/volumes' +import { isAggregatorVenue } from '../ingest/venues' +import { computeSpreads, SpreadsInputError } from '../routes/spreads' import { priceEmitter, PRICE_PUBLISHED, type PricePublishedEvent } from '../events' import { HISTORY_INTERVAL_SECONDS, MAX_HISTORY_POINTS } from './history' import { getMaxGraphQLQueryDepth, queryDepthValidationRules } from './queryDepth' @@ -231,6 +234,101 @@ export const schema = ` hasMore: Boolean! } + """ + One venue's figures in a \`volumes\` or \`spreads\` result. Both queries + return venues as a list of this type so a caller can select only the fields + it wants per venue. + + Which fields are set depends on the query, and mirrors the matching REST + route field for field: + + - \`volumes\`: \`venue\`, \`volume\` (the asset's traded volume on that venue, + in the asset's own units, summed across every pair) and \`aggregator\`. + The spread fields are null. + - \`spreads\`: one entry per (venue, pair): \`venue\`, \`pairKey\`, \`bid\`, + \`ask\`, \`mid\`, \`spreadBps\`, \`spreadPct\`, \`observations\`, + \`lastSeen\`. \`volume\` is null. + """ + type VenueQuote { + venue: String! + """Whether the venue is an aggregator re-reporting primary-market trades (volumes only).""" + aggregator: Boolean + """Traded volume in the asset's own units over the window (volumes only).""" + volume: Float + """The pair this quote is for (spreads only): two pairs on one venue are different books.""" + pairKey: String + """Lowest print in the window (spreads only).""" + bid: Float + """Highest print in the window (spreads only).""" + ask: Float + mid: Float + """(ask - bid) / mid * 10000 (spreads only).""" + spreadBps: Float + """spreadBps / 100 (spreads only).""" + spreadPct: Float + observations: Int + lastSeen: String + } + + """Labelled net (primary markets only) and gross (including aggregators) volume.""" + type VolumeTotals { + net: Float! + gross: Float! + } + + type VolumeProvenance { + venues: [String!]! + sampleCount: Int! + observedSampleCount: Int! + latestTradeAt: String + dataAgeSeconds: Int + } + + """ + Cross-venue traded volume for an asset: the same figures as REST + GET /volumes/:asset, with the per-venue breakdown (\`byVenue\` on REST) as a + typed \`venues\` list. + """ + type VolumeReport { + asset: String! + network: String! + window: String! + startTime: String! + endTime: String! + totals: VolumeTotals! + netVolume: Float! + grossVolume: Float! + """Alias for netVolume, as on REST.""" + totalVolume: Float! + """Per-venue volume, ordered by venue name like REST's \`venues\`.""" + venues: [VenueQuote!]! + tradeCount: Int! + netTradeCount: Int! + grossTradeCount: Int! + medianTradeSize: Float + topTradeVolumeShare: Float + volumeConcentration: Float + concentrationMeasure: String! + """ok, no_trades or no_positive_volume.""" + status: String! + provenance: VolumeProvenance! + } + + """ + Per-(venue, pair) spreads for an asset: the same figures as REST + GET /spreads/:asset. \`venues\` is tightest first; \`tightest\` is null when no + venue quoted the asset in the window. + """ + type SpreadReport { + asset: String! + network: String! + window: String! + startTime: String! + endTime: String! + venues: [VenueQuote!]! + tightest: VenueQuote + } + type Query { """ Aggregated price for a pair. Three outcomes are kept distinct: @@ -288,6 +386,20 @@ export const schema = ` cursor: String limit: Int = ${SCREENER_DEFAULT_LIMIT} ): ScreenerPage! + """ + Cross-venue traded volume for an asset. Shares its implementation with + REST GET /volumes/:asset: same windows (24h, 7d, 30d; default 24h), same + figures. \`network\` (testnet or mainnet) defaults to the active network; + any other value fails with INVALID_INPUT, as REST answers 400. + """ + volumes(asset: String!, window: String, network: String): VolumeReport! + """ + Per-(venue, pair) bid/ask spreads for an asset. Shares its implementation + with REST GET /spreads/:asset: same windows (5m, 15m, 1h, 24h; default 5m), + same figures. \`network\` (testnet or mainnet) defaults to the active + network; any other value fails with INVALID_INPUT, as REST answers 400. + """ + spreads(asset: String!, window: String, network: String): SpreadReport! } type Subscription { @@ -321,6 +433,14 @@ function findPair(assetA: string, assetB: string) { const oracleAdapters: OracleAdapter[] = [reflectorOracleAdapter] +/** Maps a shared REST/GraphQL validation failure onto the GraphQL INVALID_INPUT error. */ +function asInvalidInput(err: unknown): never { + if (err instanceof VolumesInputError || err instanceof SpreadsInputError) { + throw new ErrorWithProps(err.message, { code: 'INVALID_INPUT', field: err.field }) + } + throw err +} + const resolvers = { Query: { async getPrice(_: unknown, { assetA, assetB }: { assetA: string; assetB: string }) { @@ -624,6 +744,39 @@ const resolvers = { throw error } }, + + async volumes( + _: unknown, + { asset, window, network }: { asset: string; window?: string | null; network?: string | null } + ) { + // The REST route's own aggregation (src/routes/volumes.ts); this layer + // only reshapes its byVenue map into a typed list of venues. + const report = await computeVolumes(asset, window ?? undefined, network ?? undefined) + .catch(asInvalidInput) + const byVenue = report.byVenue + const names: string[] = report.venues + return { + ...report, + venues: names.map(venue => ({ + venue, + volume: Number(byVenue[venue]) || 0, + aggregator: isAggregatorVenue(venue), + })), + } + }, + + async spreads( + _: unknown, + { asset, window, network }: { asset: string; window?: string | null; network?: string | null } + ) { + // Same guard as the REST handler, which answers an empty asset with a 400. + if (!asset) { + throw new ErrorWithProps('Asset parameter is required', { code: 'INVALID_INPUT', field: 'asset' }) + } + // The REST route's own computation (src/routes/spreads.ts), returned as is. + return computeSpreads(asset, window ?? undefined, network ?? undefined) + .catch(asInvalidInput) + }, }, Subscription: { diff --git a/src/routes/spreads.ts b/src/routes/spreads.ts index 29bf707f..855f30c6 100644 --- a/src/routes/spreads.ts +++ b/src/routes/spreads.ts @@ -20,7 +20,7 @@ const WINDOWS = Object.keys(WINDOW_MINUTES) as SpreadWindow[] /** A venue needs at least this many observations before its spread means anything. */ export const MIN_OBSERVATIONS = 2 -interface VenueSpread { +export interface VenueSpread { venue: string pairKey: string bid: number @@ -72,93 +72,127 @@ export async function registerSpreadsRoutes(app: FastifyInstance) { Querystring: { window?: string; network?: string } }>('/spreads/:asset', { config: { scope: 'read:prices' } }, async (req, reply) => { const { asset } = req.params - const window = req.query.window ?? '5m' if (!asset) { return reply.status(400).send({ error: 'Asset parameter is required' }) } - if (!(window in WINDOW_MINUTES)) { - return reply - .status(400) - .send({ error: `window must be one of: ${WINDOWS.join(', ')}` }) - } - - // Prices from two chains are not comparable, and a spread computed across - // both is not a wide spread — it is a meaningless one. Default to this - // process's own network rather than silently pooling them. - const requested = req.query.network - if (requested !== undefined && requested !== 'testnet' && requested !== 'mainnet') { - return reply.status(400).send({ error: 'network must be one of: testnet, mainnet' }) - } - const network: NetworkName = (requested as NetworkName) ?? activeNetwork - - const minutes = WINDOW_MINUTES[window as SpreadWindow] - const endTime = new Date() - const startTime = new Date(endTime.getTime() - minutes * 60 * 1000) - try { - const { rows } = await pgPool.query( - `SELECT source, - pair_key, - MIN(price)::numeric AS bid, - MAX(price)::numeric AS ask, - COUNT(*)::int AS observations, - MAX(timestamp) AS last_seen - FROM price_points - WHERE (asset_a = $1 OR asset_b = $1) - AND network = $2 - AND timestamp >= $3 - GROUP BY source, pair_key`, - [asset, network, startTime], - ) - - const venues: VenueSpread[] = [] - - for (const row of rows) { - const bid = parseFloat(row.bid) - const ask = parseFloat(row.ask) - const observations = Number(row.observations) || 0 - - // A single print has a zero range by construction, which would report a - // perfect 0 bps spread and beat every real venue for "tightest". One - // observation is not evidence of liquidity. - if (observations < MIN_OBSERVATIONS) continue - if (!Number.isFinite(bid) || !Number.isFinite(ask)) continue - - const mid = (ask + bid) / 2 - if (mid <= 0) continue - - venues.push({ - venue: row.source, - pairKey: row.pair_key, - bid, - ask, - mid, - spreadBps: ((ask - bid) / mid) * 10_000, - spreadPct: ((ask - bid) / mid) * 100, - observations, - lastSeen: new Date(row.last_seen).toISOString(), - }) - } - - venues.sort((a, b) => a.spreadBps - b.spreadBps) - - return { - asset, - network, - window, - startTime: startTime.toISOString(), - endTime: endTime.toISOString(), - // Per venue-and-pair, tightest first. - venues, - // The tightest quote available anywhere in the window, or null when no - // venue quoted this asset. Null rather than a zeroed object: "nobody - // quoted it" and "the spread is zero" must not look alike. - tightest: venues[0] ?? null, - } + return await computeSpreads(asset, req.query.window, req.query.network) } catch (err) { + if (err instanceof SpreadsInputError) { + return reply.status(400).send({ error: err.message }) + } return sendInternalError(reply, req, err, 'Spread aggregation failed') } }) } + +/** A caller-facing validation failure: REST answers it with a 400, GraphQL with INVALID_INPUT. */ +export class SpreadsInputError extends Error { + constructor(readonly field: 'window' | 'network', message: string) { + super(message) + this.name = 'SpreadsInputError' + } +} + +export interface SpreadsResult { + asset: string + network: NetworkName + window: string + startTime: string + endTime: string + venues: VenueSpread[] + tightest: VenueSpread | null +} + +/** + * The spread computation behind `GET /spreads/:asset`, shared with the GraphQL + * `spreads` query so the two cannot drift. See {@link registerSpreadsRoutes} + * for what the numbers mean. Throws {@link SpreadsInputError} for a bad + * `window` or `network`; any other error is an internal failure. + */ +export async function computeSpreads( + asset: string, + windowParam: string | undefined, + networkParam: string | undefined, +): Promise { + const window = windowParam ?? '5m' + + if (!(window in WINDOW_MINUTES)) { + throw new SpreadsInputError('window', `window must be one of: ${WINDOWS.join(', ')}`) + } + + // Prices from two chains are not comparable, and a spread computed across + // both is not a wide spread — it is a meaningless one. Default to this + // process's own network rather than silently pooling them. + const requested = networkParam + if (requested !== undefined && requested !== 'testnet' && requested !== 'mainnet') { + throw new SpreadsInputError('network', 'network must be one of: testnet, mainnet') + } + const network: NetworkName = (requested as NetworkName) ?? activeNetwork + + const minutes = WINDOW_MINUTES[window as SpreadWindow] + const endTime = new Date() + const startTime = new Date(endTime.getTime() - minutes * 60 * 1000) + + const { rows } = await pgPool.query( + `SELECT source, + pair_key, + MIN(price)::numeric AS bid, + MAX(price)::numeric AS ask, + COUNT(*)::int AS observations, + MAX(timestamp) AS last_seen + FROM price_points + WHERE (asset_a = $1 OR asset_b = $1) + AND network = $2 + AND timestamp >= $3 + GROUP BY source, pair_key`, + [asset, network, startTime], + ) + + const venues: VenueSpread[] = [] + + for (const row of rows) { + const bid = parseFloat(row.bid) + const ask = parseFloat(row.ask) + const observations = Number(row.observations) || 0 + + // A single print has a zero range by construction, which would report a + // perfect 0 bps spread and beat every real venue for "tightest". One + // observation is not evidence of liquidity. + if (observations < MIN_OBSERVATIONS) continue + if (!Number.isFinite(bid) || !Number.isFinite(ask)) continue + + const mid = (ask + bid) / 2 + if (mid <= 0) continue + + venues.push({ + venue: row.source, + pairKey: row.pair_key, + bid, + ask, + mid, + spreadBps: ((ask - bid) / mid) * 10_000, + spreadPct: ((ask - bid) / mid) * 100, + observations, + lastSeen: new Date(row.last_seen).toISOString(), + }) + } + + venues.sort((a, b) => a.spreadBps - b.spreadBps) + + return { + asset, + network, + window, + startTime: startTime.toISOString(), + endTime: endTime.toISOString(), + // Per venue-and-pair, tightest first. + venues, + // The tightest quote available anywhere in the window, or null when no + // venue quoted this asset. Null rather than a zeroed object: "nobody + // quoted it" and "the spread is zero" must not look alike. + tightest: venues[0] ?? null, + } +} diff --git a/src/routes/volumes.ts b/src/routes/volumes.ts index 340ba7ba..5715f320 100644 --- a/src/routes/volumes.ts +++ b/src/routes/volumes.ts @@ -41,143 +41,174 @@ export async function registerVolumeRoutes(app: FastifyInstance) { Querystring: { window?: string; network?: string } }>('/volumes/:asset', { config: { scope: 'read:prices' } }, async (req, reply) => { const { asset } = req.params - const window = req.query.window ?? '24h' - if (!(window in WINDOW_HOURS)) { - return reply - .status(400) - .send({ error: `window must be one of: ${WINDOWS.join(', ')}` }) + try { + return await computeVolumes(asset, req.query.window, req.query.network, req.network) + } catch (err) { + if (err instanceof VolumesInputError) { + return reply.status(400).send({ error: err.message }) + } + return sendInternalError(reply, req, err, 'Volume aggregation failed') } + }) +} - const rawRequested: unknown = req.query.network - const requested = typeof rawRequested === 'string' ? rawRequested.trim().toLowerCase() : rawRequested - if (requested !== undefined && requested !== '' && requested !== 'testnet' && requested !== 'mainnet') { - return reply.status(400).send({ error: 'network must be one of: testnet, mainnet' }) - } - const network: NetworkName = req.network - ?? (requested ? requested as NetworkName : activeNetwork) +/** A caller-facing validation failure: REST answers it with a 400, GraphQL with INVALID_INPUT. */ +export class VolumesInputError extends Error { + constructor(readonly field: 'window' | 'network', message: string) { + super(message) + this.name = 'VolumesInputError' + } +} - const hours = WINDOW_HOURS[window as VolumeWindow] - const endTime = new Date() - const startTime = new Date(endTime.getTime() - hours * 60 * 60 * 1000) +export type VolumesResult = Awaited> - try { - // The CTE gives each observation its volume in the requested asset's own - // units. Statistics are aggregated in Postgres so a long window does not - // require sending every trade to the API process. - const { rows } = await pgPool.query( - `WITH asset_trades AS ( - SELECT source, - timestamp, - (CASE - WHEN asset_a = $1 THEN base_volume - WHEN asset_b = $1 THEN counter_volume - END)::numeric AS volume - FROM price_points - WHERE (asset_a = $1 OR asset_b = $1) - AND network = $2 - AND timestamp >= $3 - ), - venue_totals AS ( - SELECT source, - SUM(volume)::float8 AS volume, - (COUNT(*) FILTER (WHERE volume > 0))::int AS trade_count - FROM asset_trades - GROUP BY source - ), - trade_stats AS ( - SELECT (COUNT(*) FILTER (WHERE volume > 0))::int AS trade_count, - COUNT(*)::int AS observed_sample_count, - SUM(volume)::float8 AS total_volume, - percentile_cont(0.5) WITHIN GROUP (ORDER BY volume::float8) - FILTER (WHERE volume > 0) AS median_trade_size, - (MAX(volume) FILTER (WHERE volume > 0))::float8 - / NULLIF((SUM(volume) FILTER (WHERE volume > 0))::float8, 0) AS top_trade_volume_share, - (SUM(volume * volume) FILTER (WHERE volume > 0))::float8 - / NULLIF(POWER((SUM(volume) FILTER (WHERE volume > 0))::float8, 2), 0) AS volume_concentration, - MAX(timestamp) FILTER (WHERE volume > 0) AS latest_trade_at - FROM asset_trades - ), - venue_summary AS ( - SELECT COALESCE(jsonb_object_agg(source, volume) FILTER (WHERE volume > 0), '{}'::jsonb) AS by_venue, - COALESCE(jsonb_agg(source ORDER BY source) FILTER (WHERE volume > 0), '[]'::jsonb) AS venues, - COALESCE(jsonb_object_agg(source, trade_count) FILTER (WHERE volume > 0), '{}'::jsonb) AS trade_count_by_venue - FROM venue_totals - ) - SELECT trade_stats.*, venue_summary.by_venue, venue_summary.venues, venue_summary.trade_count_by_venue - FROM trade_stats CROSS JOIN venue_summary`, - [asset, network, startTime], - ) +/** + * The aggregation behind `GET /volumes/:asset`, shared with the GraphQL + * `volumes` query so the two cannot drift. See {@link registerVolumeRoutes} + * for what the numbers mean. + * + * `requestNetwork` is the network the request-level selector already resolved + * (`req.network` on REST); when given it wins over `networkParam`, exactly as + * the route has always behaved. Throws {@link VolumesInputError} for a bad + * `window` or `network`; any other error is an internal failure. + */ +export async function computeVolumes( + asset: string, + windowParam: string | undefined, + networkParam: unknown, + requestNetwork?: NetworkName, +) { + const window = windowParam ?? '24h' - const row = rows[0] ?? {} - const tradeCount = Number(row.trade_count) || 0 - const observedSampleCount = Number(row.observed_sample_count) || 0 - const totalVolume = Number(row.total_volume) || 0 - const latestTradeAt = row.latest_trade_at == null - ? null - : new Date(row.latest_trade_at as string | Date).toISOString() - const dataAgeSeconds = latestTradeAt === null - ? null - : Math.max(0, Math.floor((endTime.getTime() - new Date(latestTradeAt).getTime()) / 1000)) - const status = observedSampleCount === 0 - ? 'no_trades' - : tradeCount === 0 || totalVolume <= 0 - ? 'no_positive_volume' - : 'ok' + if (!(window in WINDOW_HOURS)) { + throw new VolumesInputError('window', `window must be one of: ${WINDOWS.join(', ')}`) + } - // Net excludes aggregator venues, which re-report trades that already - // settled on a primary market; gross is everything observed. The - // distribution statistics above (median, HHI, status) stay on gross. - const byVenue: Record = row.by_venue ?? {} - const tradesByVenue: Record = row.trade_count_by_venue ?? {} - let aggregatorVolume = 0 - let aggregatorTradeCount = 0 - for (const [source, volume] of Object.entries(byVenue)) { - if (!isAggregatorVenue(source)) continue - aggregatorVolume += Number(volume) || 0 - aggregatorTradeCount += Number(tradesByVenue[source]) || 0 - } - const grossVolume = totalVolume - const grossTradeCount = tradeCount - const netVolume = Math.max(0, grossVolume - aggregatorVolume) - const netTradeCount = Math.max(0, grossTradeCount - aggregatorTradeCount) + const rawRequested: unknown = networkParam + const requested = typeof rawRequested === 'string' ? rawRequested.trim().toLowerCase() : rawRequested + if (requested !== undefined && requested !== '' && requested !== 'testnet' && requested !== 'mainnet') { + throw new VolumesInputError('network', 'network must be one of: testnet, mainnet') + } + const network: NetworkName = requestNetwork + ?? (requested ? requested as NetworkName : activeNetwork) - return { - asset, - network, - window, - startTime: startTime.toISOString(), - endTime: endTime.toISOString(), - // Labelled net and gross totals to eliminate aggregator double-counting. - totals: { - net: netVolume, - gross: grossVolume, - }, - netVolume, - grossVolume, - // Backward-compatible alias for net volume. - totalVolume: netVolume, - // Per-venue breakdown that the totals are summed from. - byVenue, - venues: row.venues ?? [], - tradeCount: netTradeCount, - netTradeCount, - grossTradeCount, - medianTradeSize: row.median_trade_size == null ? null : Number(row.median_trade_size), - topTradeVolumeShare: row.top_trade_volume_share == null ? null : Number(row.top_trade_volume_share), - volumeConcentration: row.volume_concentration == null ? null : Number(row.volume_concentration), - concentrationMeasure: 'hhi', - status, - provenance: { - venues: row.venues ?? [], - sampleCount: grossTradeCount, - observedSampleCount, - latestTradeAt, - dataAgeSeconds, - }, - } - } catch (err) { - return sendInternalError(reply, req, err, 'Volume aggregation failed') - } - }) + const hours = WINDOW_HOURS[window as VolumeWindow] + const endTime = new Date() + const startTime = new Date(endTime.getTime() - hours * 60 * 60 * 1000) + + // The CTE gives each observation its volume in the requested asset's own + // units. Statistics are aggregated in Postgres so a long window does not + // require sending every trade to the API process. + const { rows } = await pgPool.query( + `WITH asset_trades AS ( + SELECT source, + timestamp, + (CASE + WHEN asset_a = $1 THEN base_volume + WHEN asset_b = $1 THEN counter_volume + END)::numeric AS volume + FROM price_points + WHERE (asset_a = $1 OR asset_b = $1) + AND network = $2 + AND timestamp >= $3 + ), + venue_totals AS ( + SELECT source, + SUM(volume)::float8 AS volume, + (COUNT(*) FILTER (WHERE volume > 0))::int AS trade_count + FROM asset_trades + GROUP BY source + ), + trade_stats AS ( + SELECT (COUNT(*) FILTER (WHERE volume > 0))::int AS trade_count, + COUNT(*)::int AS observed_sample_count, + SUM(volume)::float8 AS total_volume, + percentile_cont(0.5) WITHIN GROUP (ORDER BY volume::float8) + FILTER (WHERE volume > 0) AS median_trade_size, + (MAX(volume) FILTER (WHERE volume > 0))::float8 + / NULLIF((SUM(volume) FILTER (WHERE volume > 0))::float8, 0) AS top_trade_volume_share, + (SUM(volume * volume) FILTER (WHERE volume > 0))::float8 + / NULLIF(POWER((SUM(volume) FILTER (WHERE volume > 0))::float8, 2), 0) AS volume_concentration, + MAX(timestamp) FILTER (WHERE volume > 0) AS latest_trade_at + FROM asset_trades + ), + venue_summary AS ( + SELECT COALESCE(jsonb_object_agg(source, volume) FILTER (WHERE volume > 0), '{}'::jsonb) AS by_venue, + COALESCE(jsonb_agg(source ORDER BY source) FILTER (WHERE volume > 0), '[]'::jsonb) AS venues, + COALESCE(jsonb_object_agg(source, trade_count) FILTER (WHERE volume > 0), '{}'::jsonb) AS trade_count_by_venue + FROM venue_totals + ) + SELECT trade_stats.*, venue_summary.by_venue, venue_summary.venues, venue_summary.trade_count_by_venue + FROM trade_stats CROSS JOIN venue_summary`, + [asset, network, startTime], + ) + + const row = rows[0] ?? {} + const tradeCount = Number(row.trade_count) || 0 + const observedSampleCount = Number(row.observed_sample_count) || 0 + const totalVolume = Number(row.total_volume) || 0 + const latestTradeAt = row.latest_trade_at == null + ? null + : new Date(row.latest_trade_at as string | Date).toISOString() + const dataAgeSeconds = latestTradeAt === null + ? null + : Math.max(0, Math.floor((endTime.getTime() - new Date(latestTradeAt).getTime()) / 1000)) + const status = observedSampleCount === 0 + ? 'no_trades' + : tradeCount === 0 || totalVolume <= 0 + ? 'no_positive_volume' + : 'ok' + + // Net excludes aggregator venues, which re-report trades that already + // settled on a primary market; gross is everything observed. The + // distribution statistics above (median, HHI, status) stay on gross. + const byVenue: Record = row.by_venue ?? {} + const tradesByVenue: Record = row.trade_count_by_venue ?? {} + let aggregatorVolume = 0 + let aggregatorTradeCount = 0 + for (const [source, volume] of Object.entries(byVenue)) { + if (!isAggregatorVenue(source)) continue + aggregatorVolume += Number(volume) || 0 + aggregatorTradeCount += Number(tradesByVenue[source]) || 0 + } + const grossVolume = totalVolume + const grossTradeCount = tradeCount + const netVolume = Math.max(0, grossVolume - aggregatorVolume) + const netTradeCount = Math.max(0, grossTradeCount - aggregatorTradeCount) + + return { + asset, + network, + window, + startTime: startTime.toISOString(), + endTime: endTime.toISOString(), + // Labelled net and gross totals to eliminate aggregator double-counting. + totals: { + net: netVolume, + gross: grossVolume, + }, + netVolume, + grossVolume, + // Backward-compatible alias for net volume. + totalVolume: netVolume, + // Per-venue breakdown that the totals are summed from. + byVenue, + venues: row.venues ?? [], + tradeCount: netTradeCount, + netTradeCount, + grossTradeCount, + medianTradeSize: row.median_trade_size == null ? null : Number(row.median_trade_size), + topTradeVolumeShare: row.top_trade_volume_share == null ? null : Number(row.top_trade_volume_share), + volumeConcentration: row.volume_concentration == null ? null : Number(row.volume_concentration), + concentrationMeasure: 'hhi', + status, + provenance: { + venues: row.venues ?? [], + sampleCount: grossTradeCount, + observedSampleCount, + latestTradeAt, + dataAgeSeconds, + }, + } }