From 330d6ba76270563510c44e38bf45948602d5af40 Mon Sep 17 00:00:00 2001 From: Jossy-ossy Date: Fri, 16 Oct 2026 23:16:42 +0100 Subject: [PATCH] Preserve decimal precision in GraphQL prices --- docs/GRAPHQL_DECIMAL_MIGRATION.md | 32 ++++ src/__tests__/aggregateRefreshNetwork.test.ts | 18 +- src/__tests__/graphqlSubscription.test.ts | 42 ++++- src/__tests__/pricingFixture.test.ts | 2 +- src/__tests__/vwapNetwork.test.ts | 24 ++- src/aggregator/vwap.ts | 154 ++++++++++++------ src/api/graphql.ts | 124 ++++++++++---- src/api/history.ts | 4 +- src/api/rest.ts | 32 ++-- src/jobs/aggregateRefresh.ts | 33 ++-- src/routes/candles.ts | 10 +- src/routes/screener.ts | 22 ++- 12 files changed, 358 insertions(+), 139 deletions(-) create mode 100644 docs/GRAPHQL_DECIMAL_MIGRATION.md diff --git a/docs/GRAPHQL_DECIMAL_MIGRATION.md b/docs/GRAPHQL_DECIMAL_MIGRATION.md new file mode 100644 index 00000000..a0303d87 --- /dev/null +++ b/docs/GRAPHQL_DECIMAL_MIGRATION.md @@ -0,0 +1,32 @@ +# GraphQL decimal field migration + +GraphQL `Float` uses IEEE 754 binary floating point and cannot preserve every +value stored in PostgreSQL `numeric`. The API now exposes parallel `Decimal` +fields serialized as decimal strings. These fields preserve the PostgreSQL +numeric text through the GraphQL response; clients should keep them as strings +or parse them with an arbitrary-precision decimal library rather than convert +them to JavaScript `Number`. + +For example, request `priceExact` beside the deprecated `price` field: + +```graphql +query { + getPrice(assetA: "XLM", assetB: "USDC") { + priceExact + volume24hExact + } +} +``` + +`AggregatedPrice` exposes exact companions for price, per-source prices, +volumes, VWAPs, and 24-hour price change. `PriceBucket` exposes exact +companions for VWAP, volume, and OHLC values. Existing `Float` fields remain +available and return their historical numeric shape, but are marked deprecated +in the schema so clients can migrate without a flag day. They will only be +removed in a future breaking API release; no removal date is set. + +The trade-off is that exact values arrive as strings, which requires explicit +decimal handling in clients. This avoids returning a plausible rounded number +when stored precision matters. Price and volume SQL now retains PostgreSQL +`numeric` calculations; conversion to `Number` happens only for the deprecated +compatibility fields and the existing REST response shapes. diff --git a/src/__tests__/aggregateRefreshNetwork.test.ts b/src/__tests__/aggregateRefreshNetwork.test.ts index 8892cecd..a26f6a72 100644 --- a/src/__tests__/aggregateRefreshNetwork.test.ts +++ b/src/__tests__/aggregateRefreshNetwork.test.ts @@ -3,15 +3,15 @@ import { describe, it, expect, vi, beforeEach } from 'vitest' const { addedJobs, queueNames, workers, mockUpsert, mockSetCachedPrice, - mockCalculateVWAP, mockCalculateOHLCV, mockGetAggregatedPrice, mockGetBestRoute, + mockCalculateVWAPExact, mockCalculateOHLCVExact, mockGetAggregatedPrice, mockGetBestRoute, } = vi.hoisted(() => ({ addedJobs: [] as Array<{ queue: string; name: string; data: any; opts: any }>, queueNames: [] as string[], workers: [] as Array<{ name: string; processor: (job: any) => Promise }>, mockUpsert: vi.fn(), mockSetCachedPrice: vi.fn(), - mockCalculateVWAP: vi.fn(), - mockCalculateOHLCV: vi.fn(), + mockCalculateVWAPExact: vi.fn(), + mockCalculateOHLCVExact: vi.fn(), mockGetAggregatedPrice: vi.fn(), mockGetBestRoute: vi.fn(), })) @@ -51,8 +51,8 @@ vi.mock('../db', () => ({ vi.mock('../redis', () => ({ setCachedPrice: mockSetCachedPrice })) vi.mock('../aggregator/vwap', () => ({ - calculateVWAP: mockCalculateVWAP, - calculateOHLCV: mockCalculateOHLCV, + calculateVWAPExact: mockCalculateVWAPExact, + calculateOHLCVExact: mockCalculateOHLCVExact, getAggregatedPrice: mockGetAggregatedPrice, })) @@ -89,8 +89,8 @@ describe('aggregateRefresh runs once per enabled network', () => { queueNames.length = 0 workers.length = 0 vi.clearAllMocks() - mockCalculateVWAP.mockResolvedValue(0.25) - mockCalculateOHLCV.mockResolvedValue({ open: 0.2, high: 0.3, low: 0.2, close: 0.25, volume: 100, tradeCount: 3 }) + mockCalculateVWAPExact.mockResolvedValue('0.25') + mockCalculateOHLCVExact.mockResolvedValue({ open: '0.2', high: '0.3', low: '0.2', close: '0.25', volume: '100', tradeCount: 3 }) mockGetAggregatedPrice.mockResolvedValue({ price: 0.25 }) mockGetBestRoute.mockResolvedValue({ route: 'SDEX' }) mockSetCachedPrice.mockResolvedValue(undefined) @@ -137,10 +137,10 @@ describe('aggregateRefresh runs once per enabled network', () => { expect(mockGetBestRoute).toHaveBeenCalledWith( mainnetPair.assetA, mainnetPair.assetB, mainnetPair.pairKey, 1000, 'mainnet' ) - for (const call of mockCalculateVWAP.mock.calls) { + for (const call of mockCalculateVWAPExact.mock.calls) { expect(call).toContain('mainnet') } - for (const call of mockCalculateOHLCV.mock.calls) { + for (const call of mockCalculateOHLCVExact.mock.calls) { expect(call).toContain('mainnet') } expect(mockSetCachedPrice).toHaveBeenCalledWith( diff --git a/src/__tests__/graphqlSubscription.test.ts b/src/__tests__/graphqlSubscription.test.ts index 27f3e4ac..8207dce9 100644 --- a/src/__tests__/graphqlSubscription.test.ts +++ b/src/__tests__/graphqlSubscription.test.ts @@ -10,9 +10,14 @@ vi.mock('../redis', () => ({ getCachedPrice: vi.fn() })) vi.mock('../db', () => ({ pgPool: { query: vi.fn() } })) vi.mock('../aggregator/vwap', () => ({ getAggregatedPrice: vi.fn() })) vi.mock('../aggregator/bestRoute', () => ({ getBestRoute: vi.fn() })) -vi.mock('../config', () => ({ config: { pairs: [] } })) +vi.mock('../config', () => ({ + config: { pairs: [{ pairKey: 'USDC/XLM', assetA: { code: 'XLM', issuer: null }, assetB: { code: 'USDC', issuer: 'ISSUER' } }] }, + activeNetwork: 'testnet', +})) import { registerGraphQL } from '../api/graphql' +import { getAggregatedPrice } from '../aggregator/vwap' +import { getBestRoute } from '../aggregator/bestRoute' import { publishPriceUpdate, priceEmitter } from '../events' const SUBPROTOCOL = 'graphql-transport-ws' @@ -181,3 +186,38 @@ describe('GraphQL priceUpdated subscription', () => { ws.close() }) }) + +describe('GraphQL exact decimal fields', () => { + it('returns stored decimal text without converting it through a double', async () => { + vi.mocked(getAggregatedPrice).mockResolvedValue({ + price: 0, priceExact: '0.123456789012345678', + sdexPrice: 0, sdexPriceExact: '0.123456789012345678', + ammPrice: 0, ammPriceExact: '0', + volume24h: 0, volume24hExact: '0', + sdexVolume24h: 0, sdexVolume24hExact: '0', + ammVolume24h: 0, ammVolume24hExact: '0', + vwap1m: 0, vwap1mExact: '0', vwap5m: 0, vwap5mExact: '0', + vwap1h: 0, vwap1hExact: '0', vwap24h: 0, vwap24hExact: '0', + priceChange24h: 0, priceChange24hExact: '0', sources: 0, + confidence: 'unknown', lastTradeAgeSeconds: null, stale: false, + }) + vi.mocked(getBestRoute).mockResolvedValue({ route: 'SDEX' } as any) + + const { app } = await buildServer() + try { + const response = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { query: '{ getPrice(assetA: "XLM", assetB: "USDC") { priceExact sdexPriceExact } }' }, + }) + expect(response.statusCode).toBe(200) + expect(response.json().data.getPrice).toEqual({ + priceExact: '0.123456789012345678', + sdexPriceExact: '0.123456789012345678', + }) + expect(String(Number('0.123456789012345678'))).not.toBe('0.123456789012345678') + } finally { + await app.close() + } + }) +}) diff --git a/src/__tests__/pricingFixture.test.ts b/src/__tests__/pricingFixture.test.ts index 0a35909c..367a6ac0 100644 --- a/src/__tests__/pricingFixture.test.ts +++ b/src/__tests__/pricingFixture.test.ts @@ -22,7 +22,7 @@ function installFixtureQuery() { const sorted = [...rows].sort((a, b) => a.timestamp.getTime() - b.timestamp.getTime()) const prices = sorted.map(row => row.price) return { rows: [{ - open: prices[0], high: Math.max(...prices), low: Math.min(...prices), close: prices[prices.length - 1], + open: String(prices[0]), high: String(Math.max(...prices)), low: String(Math.min(...prices)), close: String(prices[prices.length - 1]), volume: String(sorted.reduce((sum, row) => sum + row.baseVolume, 0)), trade_count: String(sorted.length), }] } } diff --git a/src/__tests__/vwapNetwork.test.ts b/src/__tests__/vwapNetwork.test.ts index f55af520..22be2abb 100644 --- a/src/__tests__/vwapNetwork.test.ts +++ b/src/__tests__/vwapNetwork.test.ts @@ -6,7 +6,7 @@ vi.mock('../db', () => ({ pgPool: { query: mockQuery }, })) -import { calculateVWAP, calculateOHLCV, getPriceChange24h, getAggregatedPrice } from '../aggregator/vwap' +import { calculateVWAP, calculateVWAPExact, calculateOHLCV, getPriceChange24h, getAggregatedPrice } from '../aggregator/vwap' import type { NetworkName } from '../config' // ── A seeded stand-in for the two `price_points`/`pool_snapshots` tables ────── @@ -114,6 +114,7 @@ function installSeededDb() { rows: [{ price_24h_ago: String(sorted[0]?.price ?? 0), price_now: String(sorted[sorted.length - 1]?.price ?? 0), + price_change: String(sorted[0]?.price ? ((sorted[sorted.length - 1]?.price ?? 0) - sorted[0].price) / sorted[0].price * 100 : 0), }], } } @@ -142,11 +143,11 @@ function installSeededDb() { return { rows: [{ sources: String(sources.size) }] } } - if (sql.includes('GROUP BY source')) { + if (sql.includes('SUM(base_volume::numeric) FILTER')) { const rows = rowsInWindow(network, pairKey, windowMinutes) - const bySource = new Map() - for (const r of rows) bySource.set(r.source, (bySource.get(r.source) ?? 0) + r.baseVolume) - return { rows: [...bySource].map(([source, vol]) => ({ source, vol: String(vol) })) } + const sdex = rows.filter(row => row.source === 'SDEX').reduce((sum, row) => sum + row.baseVolume, 0) + const amm = rows.filter(row => row.source === 'AMM').reduce((sum, row) => sum + row.baseVolume, 0) + return { rows: [{ sdex_volume: String(sdex), amm_volume: String(amm), total_volume: String(sdex + amm) }] } } // OHLCV: open/close bracket the window, high/low/volume/count cover it. @@ -155,10 +156,10 @@ function installSeededDb() { const prices = rows.map(r => r.price) return { rows: [{ - open: prices[0], - high: prices.length ? Math.max(...prices) : 0, - low: prices.length ? Math.min(...prices) : 0, - close: prices[prices.length - 1], + open: String(prices[0] ?? 0), + high: String(prices.length ? Math.max(...prices) : 0), + low: String(prices.length ? Math.min(...prices) : 0), + close: String(prices[prices.length - 1] ?? 0), volume: String(rows.reduce((acc, r) => acc + r.baseVolume, 0)), trade_count: String(rows.length), }], @@ -181,6 +182,11 @@ describe('aggregator network scoping', () => { installSeededDb() }) + it('keeps PostgreSQL decimal text intact for the exact VWAP path', async () => { + mockQuery.mockResolvedValueOnce({ rows: [{ vwap: '0.123456789012345678' }] }) + await expect(calculateVWAPExact(PAIR, 60, 'testnet')).resolves.toBe('0.123456789012345678') + }) + it('calculateVWAP returns a different price per network for the same pairKey', async () => { const testnet = await calculateVWAP(PAIR, 60, 'testnet') const mainnet = await calculateVWAP(PAIR, 60, 'mainnet') diff --git a/src/aggregator/vwap.ts b/src/aggregator/vwap.ts index 335ba937..7675a079 100644 --- a/src/aggregator/vwap.ts +++ b/src/aggregator/vwap.ts @@ -18,12 +18,19 @@ export const MEDIUM_CONFIDENCE_MAX_AGE_SECONDS = 300 * refinement: there is no default that is correct by accident, and a caller * that doesn't know which chain it is quoting should refuse rather than guess. */ -export async function calculateVWAP( +function requireDecimalString(value: unknown, field: string): string { + if (typeof value !== 'string' || !/^-?\d+(?:\.\d+)?$/.test(value)) { + throw new TypeError(`Expected PostgreSQL numeric text for ${field}`) + } + return value +} + +export async function calculateVWAPExact( pairKey: string, windowMinutes: number, network: NetworkName, source?: 'SDEX' | 'AMM' -): Promise { +): Promise { const sourceFilter = source ? `AND source = $4` : '' const params: (string | number)[] = [pairKey, network, windowMinutes] if (source) params.push(source) @@ -41,16 +48,25 @@ export async function calculateVWAP( ${sourceFilter}`, params ) - return parseFloat(result.rows[0]?.vwap ?? '0') + return requireDecimalString(result.rows[0]?.vwap ?? '0', 'VWAP') } -export async function calculateOHLCV( +export async function calculateVWAP( + pairKey: string, + windowMinutes: number, + network: NetworkName, + source?: 'SDEX' | 'AMM' +): Promise { + return Number(await calculateVWAPExact(pairKey, windowMinutes, network, source)) +} + +export async function calculateOHLCVExact( pairKey: string, windowMinutes: number, network: NetworkName ): Promise<{ - open: number; high: number; low: number; close: number - volume: number; tradeCount: number + open: string; high: string; low: string; close: string + volume: string; tradeCount: number }> { const result = await pgPool.query( `SELECT @@ -74,51 +90,79 @@ export async function calculateOHLCV( ) const row = result.rows[0] return { - open: parseFloat(row?.open ?? '0'), - high: parseFloat(row?.high ?? '0'), - low: parseFloat(row?.low ?? '0'), - close: parseFloat(row?.close ?? '0'), - volume: parseFloat(row?.volume ?? '0'), + open: requireDecimalString(row?.open ?? '0', 'OHLCV open'), + high: requireDecimalString(row?.high ?? '0', 'OHLCV high'), + low: requireDecimalString(row?.low ?? '0', 'OHLCV low'), + close: requireDecimalString(row?.close ?? '0', 'OHLCV close'), + volume: requireDecimalString(row?.volume ?? '0', 'OHLCV volume'), tradeCount: parseInt(row?.trade_count ?? '0', 10), } } +export async function calculateOHLCV( + pairKey: string, + windowMinutes: number, + network: NetworkName +): Promise<{ + open: number; high: number; low: number; close: number + volume: number; tradeCount: number +}> { + const exact = await calculateOHLCVExact(pairKey, windowMinutes, network) + return { + open: Number(exact.open), + high: Number(exact.high), + low: Number(exact.low), + close: Number(exact.close), + volume: Number(exact.volume), + tradeCount: exact.tradeCount, + } +} + export async function getPriceChange24h(pairKey: string, network: NetworkName): Promise { + return Number(await getPriceChange24hExact(pairKey, network)) +} + +export async function getPriceChange24hExact(pairKey: string, network: NetworkName): Promise { const result = await pgPool.query( - `SELECT - (SELECT price::numeric FROM price_points - WHERE pair_key = $1 AND network = $2 - AND timestamp > NOW() - INTERVAL '24 hours' - ORDER BY timestamp ASC LIMIT 1) AS price_24h_ago, - (SELECT price::numeric FROM price_points - WHERE pair_key = $1 AND network = $2 - AND timestamp > NOW() - INTERVAL '24 hours' - ORDER BY timestamp DESC LIMIT 1) AS price_now`, + `WITH prices AS ( + SELECT + (SELECT price::numeric FROM price_points + WHERE pair_key = $1 AND network = $2 + AND timestamp > NOW() - INTERVAL '24 hours' + ORDER BY timestamp ASC LIMIT 1) AS price_24h_ago, + (SELECT price::numeric FROM price_points + WHERE pair_key = $1 AND network = $2 + AND timestamp > NOW() - INTERVAL '24 hours' + ORDER BY timestamp DESC LIMIT 1) AS price_now + ) + SELECT CASE WHEN COALESCE(price_24h_ago, 0) = 0 THEN 0 + ELSE ((price_now - price_24h_ago) / price_24h_ago) * 100 + END AS price_change + FROM prices`, [pairKey, network] ) - const row = result.rows[0] - const ago = parseFloat(row?.price_24h_ago ?? '0') - const now = parseFloat(row?.price_now ?? '0') - if (ago === 0) return 0 - return ((now - ago) / ago) * 100 + return requireDecimalString(result.rows[0]?.price_change ?? '0', '24-hour price change') } export async function getAggregatedPrice(pairKey: string, network: NetworkName): Promise<{ price: number; sdexPrice: number; ammPrice: number + priceExact: string; sdexPriceExact: string; ammPriceExact: string volume24h: number; sdexVolume24h: number; ammVolume24h: number + volume24hExact: string; sdexVolume24hExact: string; ammVolume24hExact: string vwap1m: number; vwap5m: number; vwap1h: number; vwap24h: number - priceChange24h: number; sources: number + vwap1mExact: string; vwap5mExact: string; vwap1hExact: string; vwap24hExact: string + priceChange24h: number; priceChange24hExact: string; sources: number confidence: 'high' | 'medium' | 'low' | 'unknown' lastTradeAgeSeconds: number | null stale: boolean }> { - const [vwap1m, vwap5m, vwap1h, vwap24h, sdexVwap24h, priceChange24h, ammSpotResult, lastTradeResult] = await Promise.all([ - calculateVWAP(pairKey, 1, network), - calculateVWAP(pairKey, 5, network), - calculateVWAP(pairKey, 60, network), - calculateVWAP(pairKey, 1440, network), - calculateVWAP(pairKey, 1440, network, 'SDEX'), - getPriceChange24h(pairKey, network), + const [vwap1mExact, vwap5mExact, vwap1hExact, vwap24hExact, sdexPriceExact, priceChange24hExact, ammSpotResult, lastTradeResult] = await Promise.all([ + calculateVWAPExact(pairKey, 1, network), + calculateVWAPExact(pairKey, 5, network), + calculateVWAPExact(pairKey, 60, network), + calculateVWAPExact(pairKey, 1440, network), + calculateVWAPExact(pairKey, 1440, network, 'SDEX'), + getPriceChange24hExact(pairKey, network), // AMM price: latest spot_price averaged across pools for this pair, on this // network only. Both halves of the lookup are network-filtered: the outer // scan of pool_snapshots, and the price_points subquery that decides which @@ -148,22 +192,20 @@ export async function getAggregatedPrice(pairKey: string, network: NetworkName): const volResult = await pgPool.query( `SELECT - source, - SUM(base_volume::numeric) AS vol + COALESCE(SUM(base_volume::numeric) FILTER (WHERE source = 'SDEX'), 0) AS sdex_volume, + COALESCE(SUM(base_volume::numeric) FILTER (WHERE source = 'AMM'), 0) AS amm_volume, + COALESCE(SUM(base_volume::numeric) FILTER (WHERE source IN ('SDEX', 'AMM')), 0) AS total_volume FROM price_points WHERE pair_key = $1 AND network = $2 - AND timestamp > NOW() - INTERVAL '24 hours' - GROUP BY source`, + AND timestamp > NOW() - INTERVAL '24 hours'`, [pairKey, network] ) - let sdexVolume24h = 0 - let ammVolume24h = 0 - for (const row of volResult.rows) { - if (row.source === 'SDEX') sdexVolume24h = parseFloat(row.vol) - else if (row.source === 'AMM') ammVolume24h = parseFloat(row.vol) - } + const volumeRow = (volResult.rows[0] ?? {}) as Record + const sdexVolume24hExact = requireDecimalString(volumeRow.sdex_volume ?? '0', 'SDEX volume') + const ammVolume24hExact = requireDecimalString(volumeRow.amm_volume ?? '0', 'AMM volume') + const volume24hExact = requireDecimalString(volumeRow.total_volume ?? '0', '24-hour volume') const sourcesResult = await pgPool.query( `SELECT COUNT(DISTINCT COALESCE(pool_id, 'sdex')) AS sources @@ -174,11 +216,20 @@ export async function getAggregatedPrice(pairKey: string, network: NetworkName): [pairKey, network] ) - const ammPrice = parseFloat(ammSpotResult.rows[0]?.amm_price ?? '0') + const ammPriceExact = requireDecimalString(ammSpotResult.rows[0]?.amm_price ?? '0', 'AMM price') + const ammPrice = Number(ammPriceExact) + const vwap1m = Number(vwap1mExact) + const vwap5m = Number(vwap5mExact) + const vwap1h = Number(vwap1hExact) + const vwap24h = Number(vwap24hExact) + const sdexPrice = Number(sdexPriceExact) + const sdexVolume24h = Number(sdexVolume24hExact) + const ammVolume24h = Number(ammVolume24hExact) + const volume24h = Number(volume24hExact) + const priceChange24h = Number(priceChange24hExact) const lastTrade = lastTradeResult.rows[0]?.last_trade const lastTradeAgeSeconds = lastTrade ? Math.floor((Date.now() - new Date(lastTrade).getTime()) / 1000) : null const sources = parseInt(sourcesResult.rows[0]?.sources ?? '0', 10) - const volume24h = sdexVolume24h + ammVolume24h let confidence: 'high' | 'medium' | 'low' | 'unknown' = 'unknown' @@ -196,16 +247,27 @@ export async function getAggregatedPrice(pairKey: string, network: NetworkName): return { price: vwap1h || vwap24h || ammPrice || 0, - sdexPrice: sdexVwap24h, + priceExact: isNonZeroDecimal(vwap1hExact) ? vwap1hExact : isNonZeroDecimal(vwap24hExact) ? vwap24hExact : ammPriceExact, + sdexPrice, + sdexPriceExact, ammPrice, + ammPriceExact, volume24h, + volume24hExact, sdexVolume24h, + sdexVolume24hExact, ammVolume24h, + ammVolume24hExact, vwap1m, + vwap1mExact, vwap5m, + vwap5mExact, vwap1h, + vwap1hExact, vwap24h, + vwap24hExact, priceChange24h, + priceChange24hExact, sources, confidence, lastTradeAgeSeconds, diff --git a/src/api/graphql.ts b/src/api/graphql.ts index a32a7a7b..bc6ac34b 100644 --- a/src/api/graphql.ts +++ b/src/api/graphql.ts @@ -1,8 +1,7 @@ import type { FastifyInstance } from 'fastify' -import { ErrorWithProps } from 'mercurius' +import { GraphQLScalarType, GraphQLError, Kind, type ExecutionResult } from 'graphql' import { price_requests_total } from '../metrics' -import mercurius, { withFilter, defaultErrorFormatter, type MercuriusContext } from 'mercurius' -import { GraphQLError, type ExecutionResult } from 'graphql' +import mercurius, { ErrorWithProps, withFilter, defaultErrorFormatter, type MercuriusContext } from 'mercurius' import { randomUUID } from 'node:crypto' import { getCachedPrice } from '../redis' import { getAggregatedPrice } from '../aggregator/vwap' @@ -64,22 +63,35 @@ export function formatGraphQLErrors(execution: ExecutionResult, ctx: MercuriusCo } export const schema = ` + scalar Decimal + type AggregatedPrice { assetA: String! assetB: String! pairKey: String! - price: Float! - sdexPrice: Float! - ammPrice: Float! + price: Float! @deprecated(reason: "Use priceExact to preserve decimal precision") + priceExact: Decimal! + sdexPrice: Float! @deprecated(reason: "Use sdexPriceExact to preserve decimal precision") + sdexPriceExact: Decimal! + ammPrice: Float! @deprecated(reason: "Use ammPriceExact to preserve decimal precision") + ammPriceExact: Decimal! bestRoute: String! - volume24h: Float! - sdexVolume24h: Float! - ammVolume24h: Float! - vwap1m: Float! - vwap5m: Float! - vwap1h: Float! - vwap24h: Float! - priceChange24h: Float! + volume24h: Float! @deprecated(reason: "Use volume24hExact to preserve decimal precision") + volume24hExact: Decimal! + sdexVolume24h: Float! @deprecated(reason: "Use sdexVolume24hExact to preserve decimal precision") + sdexVolume24hExact: Decimal! + ammVolume24h: Float! @deprecated(reason: "Use ammVolume24hExact to preserve decimal precision") + ammVolume24hExact: Decimal! + vwap1m: Float! @deprecated(reason: "Use vwap1mExact to preserve decimal precision") + vwap1mExact: Decimal! + vwap5m: Float! @deprecated(reason: "Use vwap5mExact to preserve decimal precision") + vwap5mExact: Decimal! + vwap1h: Float! @deprecated(reason: "Use vwap1hExact to preserve decimal precision") + vwap1hExact: Decimal! + vwap24h: Float! @deprecated(reason: "Use vwap24hExact to preserve decimal precision") + vwap24hExact: Decimal! + priceChange24h: Float! @deprecated(reason: "Use priceChange24hExact to preserve decimal precision") + priceChange24hExact: Decimal! lastUpdated: String! sources: Int! confidence: String! @@ -98,15 +110,23 @@ export const schema = ` type PriceBucket { bucket: String! window: String! - vwap: Float! - sdexVwap: Float - ammVwap: Float - volume: Float! + vwap: Float! @deprecated(reason: "Use vwapExact to preserve decimal precision") + vwapExact: Decimal! + sdexVwap: Float @deprecated(reason: "Use sdexVwapExact to preserve decimal precision") + sdexVwapExact: Decimal + ammVwap: Float @deprecated(reason: "Use ammVwapExact to preserve decimal precision") + ammVwapExact: Decimal + volume: Float! @deprecated(reason: "Use volumeExact to preserve decimal precision") + volumeExact: Decimal! tradeCount: Int! - open: Float - close: Float - high: Float - low: Float + open: Float @deprecated(reason: "Use openExact to preserve decimal precision") + openExact: Decimal + close: Float @deprecated(reason: "Use closeExact to preserve decimal precision") + closeExact: Decimal + high: Float @deprecated(reason: "Use highExact to preserve decimal precision") + highExact: Decimal + low: Float @deprecated(reason: "Use lowExact to preserve decimal precision") + lowExact: Decimal } type PriceUpdate { @@ -219,6 +239,27 @@ function findPair(assetA: string, assetB: string) { const oracleAdapters: OracleAdapter[] = [reflectorOracleAdapter] const resolvers = { + Decimal: new GraphQLScalarType({ + name: 'Decimal', + serialize(value) { + if (typeof value !== 'string' || !/^-?(?:\d+\.?\d*|\.\d+)$/.test(value)) { + throw new TypeError('Decimal values must be PostgreSQL decimal text') + } + return value + }, + parseValue(value) { + if (typeof value !== 'string' || !/^-?(?:\d+\.?\d*|\.\d+)$/.test(value)) { + throw new TypeError('Decimal inputs must be strings') + } + return value + }, + parseLiteral(ast) { + if (ast.kind !== Kind.STRING || !/^-?(?:\d+\.?\d*|\.\d+)$/.test(ast.value)) { + throw new TypeError('Decimal literals must be strings') + } + return ast.value + }, + }), Query: { async getPrice(_: unknown, { assetA, assetB }: { assetA: string; assetB: string }) { price_requests_total.inc() @@ -249,7 +290,15 @@ const resolvers = { const cacheKey = `${activeNetwork}:${pairKey}` const cached = await getCachedPrice(cacheKey) if (cached) { - try { return JSON.parse(cached) } catch { /* fall through */ } + try { + const parsed = JSON.parse(cached) + const exactFields = [ + 'priceExact', 'sdexPriceExact', 'ammPriceExact', 'volume24hExact', + 'sdexVolume24hExact', 'ammVolume24hExact', 'vwap1mExact', 'vwap5mExact', + 'vwap1hExact', 'vwap24hExact', 'priceChange24hExact', + ] + if (exactFields.every(field => typeof parsed[field] === 'string')) return parsed + } catch { /* fall through */ } } const agg = await getAggregatedPrice(pairKey, activeNetwork) @@ -341,9 +390,8 @@ const resolvers = { const limitParam = `$${params.length}` const result = await pgPool.query( - `SELECT bucket, window, vwap::float, sdex_vwap::float, amm_vwap::float, - volume::float, trade_count, open_price::float, close_price::float, - high_price::float, low_price::float + `SELECT bucket, window, vwap, sdex_vwap, amm_vwap, + volume, trade_count, open_price, close_price, high_price, low_price FROM price_aggregates WHERE ${conditions.join(' AND ')} ORDER BY bucket DESC @@ -353,15 +401,23 @@ const resolvers = { return result.rows.map(r => ({ bucket: (r.bucket instanceof Date ? r.bucket : new Date(r.bucket)).toISOString(), window: r.window, - vwap: r.vwap, - sdexVwap: r.sdex_vwap, - ammVwap: r.amm_vwap, - volume: r.volume, + vwap: Number(r.vwap), + vwapExact: r.vwap, + sdexVwap: r.sdex_vwap === null ? null : Number(r.sdex_vwap), + sdexVwapExact: r.sdex_vwap, + ammVwap: r.amm_vwap === null ? null : Number(r.amm_vwap), + ammVwapExact: r.amm_vwap, + volume: Number(r.volume), + volumeExact: r.volume, tradeCount: r.trade_count, - open: r.open_price, - close: r.close_price, - high: r.high_price, - low: r.low_price, + open: r.open_price === null ? null : Number(r.open_price), + openExact: r.open_price, + close: r.close_price === null ? null : Number(r.close_price), + closeExact: r.close_price, + high: r.high_price === null ? null : Number(r.high_price), + highExact: r.high_price, + low: r.low_price === null ? null : Number(r.low_price), + lowExact: r.low_price, })) }, diff --git a/src/api/history.ts b/src/api/history.ts index 1f8a7590..6aebac1d 100644 --- a/src/api/history.ts +++ b/src/api/history.ts @@ -82,8 +82,8 @@ export async function queryHistory( const result = await pgPool.query( `SELECT to_timestamp(floor(EXTRACT(EPOCH FROM ts) / $1) * $1) AS bucket, - (array_agg(price::float ORDER BY ts DESC))[1] AS price, - SUM(volume::float) AS volume + (array_agg(price::numeric ORDER BY ts DESC))[1] AS price, + SUM(volume::numeric) AS volume FROM price_snapshots WHERE pair = $2 AND network = $5 AND ts >= $3 diff --git a/src/api/rest.ts b/src/api/rest.ts index 0b4f62fa..35829126 100644 --- a/src/api/rest.ts +++ b/src/api/rest.ts @@ -177,9 +177,8 @@ export async function registerRESTRoutes(app: FastifyInstance) { } const result = await pgPool.query( - `SELECT bucket, window, vwap::float, sdex_vwap::float, amm_vwap::float, - volume::float, trade_count, open_price::float, close_price::float, - high_price::float, low_price::float + `SELECT bucket, window, vwap, sdex_vwap, amm_vwap, + volume, trade_count, open_price, close_price, high_price, low_price FROM price_aggregates WHERE pair_key = $1 AND window = $2 AND network = $3 ORDER BY bucket DESC @@ -193,15 +192,15 @@ export async function registerRESTRoutes(app: FastifyInstance) { window, buckets: result.rows.map(r => ({ bucket: r.bucket, - vwap: r.vwap, - sdexVwap: r.sdex_vwap, - ammVwap: r.amm_vwap, - volume: r.volume, + vwap: Number(r.vwap), + sdexVwap: r.sdex_vwap === null ? null : Number(r.sdex_vwap), + ammVwap: r.amm_vwap === null ? null : Number(r.amm_vwap), + volume: Number(r.volume), tradeCount: r.trade_count, - open: r.open_price, - close: r.close_price, - high: r.high_price, - low: r.low_price, + open: r.open_price === null ? null : Number(r.open_price), + close: r.close_price === null ? null : Number(r.close_price), + high: r.high_price === null ? null : Number(r.high_price), + low: r.low_price === null ? null : Number(r.low_price), })), } } @@ -228,7 +227,7 @@ export async function registerRESTRoutes(app: FastifyInstance) { WHERE ids.pool_id IS NOT NULL ) SELECT l.pool_id, l.asset_a, l.asset_b, - l.reserve_a::float, l.reserve_b::float, l.spot_price::float, l.fee_bp, l.timestamp + l.reserve_a, l.reserve_b, l.spot_price, l.fee_bp, l.timestamp FROM ids CROSS JOIN LATERAL ( SELECT * FROM pool_snapshots s @@ -239,7 +238,14 @@ export async function registerRESTRoutes(app: FastifyInstance) { WHERE ids.pool_id IS NOT NULL`, [network] ) - return { pools: result.rows } + return { + pools: result.rows.map(pool => ({ + ...pool, + reserve_a: Number(pool.reserve_a), + reserve_b: Number(pool.reserve_b), + spot_price: Number(pool.spot_price), + })), + } }) // GET /price/:assetA/:assetB/depth?amount=1000 diff --git a/src/jobs/aggregateRefresh.ts b/src/jobs/aggregateRefresh.ts index 4b49b9b7..e268fc47 100644 --- a/src/jobs/aggregateRefresh.ts +++ b/src/jobs/aggregateRefresh.ts @@ -1,8 +1,9 @@ import { Queue, Worker } from 'bullmq' +import { Prisma } from '@prisma/client' import { config, getNetworkConfig, type NetworkName } from '../config' import { prisma } from '../db' import { setCachedPrice } from '../redis' -import { calculateVWAP, calculateOHLCV, getAggregatedPrice } from '../aggregator/vwap' +import { calculateVWAPExact, calculateOHLCVExact, getAggregatedPrice } from '../aggregator/vwap' import { getBestRoute } from '../aggregator/bestRoute' import { createLogger } from '../logger' @@ -69,27 +70,33 @@ export function startAggregateWorker(network: NetworkName) { for (const w of windows) { const [vwap, sdexVwap, ammVwap, ohlcv] = await Promise.all([ - calculateVWAP(pairKey, w.minutes, network), - calculateVWAP(pairKey, w.minutes, network, 'SDEX'), - calculateVWAP(pairKey, w.minutes, network, 'AMM'), - calculateOHLCV(pairKey, w.minutes, network), + calculateVWAPExact(pairKey, w.minutes, network), + calculateVWAPExact(pairKey, w.minutes, network, 'SDEX'), + calculateVWAPExact(pairKey, w.minutes, network, 'AMM'), + calculateOHLCVExact(pairKey, w.minutes, network), ]) - if (vwap === 0) continue + if (!/[1-9]/.test(vwap)) continue + + // Prisma Decimal fields use Decimal.js values. Construct them from + // the PostgreSQL numeric text so no JavaScript Number rounding is + // introduced on the history persistence path. + const decimal = (value: string) => new Prisma.Decimal(value) + const optionalDecimal = (value: string) => /[1-9]/.test(value) ? decimal(value) : null await prisma.priceAggregate.upsert({ where: { network_pairKey_window_bucket: { network, pairKey, window: w.key, bucket } }, create: { network, pairKey, window: w.key, bucket, - vwap, sdexVwap: sdexVwap || null, ammVwap: ammVwap || null, - volume: ohlcv.volume, tradeCount: ohlcv.tradeCount, - openPrice: ohlcv.open || null, closePrice: ohlcv.close || null, - highPrice: ohlcv.high || null, lowPrice: ohlcv.low || null, + vwap: decimal(vwap), sdexVwap: optionalDecimal(sdexVwap), ammVwap: optionalDecimal(ammVwap), + volume: decimal(ohlcv.volume), tradeCount: ohlcv.tradeCount, + openPrice: optionalDecimal(ohlcv.open), closePrice: optionalDecimal(ohlcv.close), + highPrice: optionalDecimal(ohlcv.high), lowPrice: optionalDecimal(ohlcv.low), }, update: { - vwap, sdexVwap: sdexVwap || null, ammVwap: ammVwap || null, - volume: ohlcv.volume, tradeCount: ohlcv.tradeCount, - closePrice: ohlcv.close || null, highPrice: ohlcv.high || null, lowPrice: ohlcv.low || null, + vwap: decimal(vwap), sdexVwap: optionalDecimal(sdexVwap), ammVwap: optionalDecimal(ammVwap), + volume: decimal(ohlcv.volume), tradeCount: ohlcv.tradeCount, + closePrice: optionalDecimal(ohlcv.close), highPrice: optionalDecimal(ohlcv.high), lowPrice: optionalDecimal(ohlcv.low), }, }) } diff --git a/src/routes/candles.ts b/src/routes/candles.ts index 1a85f8e2..c7b8cb22 100644 --- a/src/routes/candles.ts +++ b/src/routes/candles.ts @@ -34,11 +34,11 @@ export async function registerCandleRoutes(app: FastifyInstance) { const result = await pgPool.query( `SELECT to_timestamp(floor(EXTRACT(EPOCH FROM timestamp) / $1) * $1) AS time, - (array_agg(price::float ORDER BY timestamp ASC))[1] AS open, - MAX(price::float) AS high, - MIN(price::float) AS low, - (array_agg(price::float ORDER BY timestamp DESC))[1] AS close, - SUM(base_volume::float) AS volume + (array_agg(price::numeric ORDER BY timestamp ASC))[1] AS open, + MAX(price::numeric) AS high, + MIN(price::numeric) AS low, + (array_agg(price::numeric ORDER BY timestamp DESC))[1] AS close, + SUM(base_volume::numeric) AS volume FROM price_points WHERE pair_key = $2 AND timestamp >= $3 diff --git a/src/routes/screener.ts b/src/routes/screener.ts index 0b33fe58..31234f1b 100644 --- a/src/routes/screener.ts +++ b/src/routes/screener.ts @@ -192,10 +192,10 @@ export async function registerScreenerRoutes(app: FastifyInstance) { latest_aggregates AS ( SELECT DISTINCT ON (pair_key) pair_key, - COALESCE(volume::float, 0) AS volume, - COALESCE(close_price::float, 0) AS price, + COALESCE(volume, 0) AS volume, + COALESCE(close_price, 0) AS price, CASE - WHEN open_price > 0 THEN ((close_price::float - open_price::float) / open_price::float) * 100 + WHEN open_price > 0 THEN ((close_price - open_price) / open_price) * 100 ELSE 0 END AS change_24h FROM price_aggregates @@ -208,7 +208,7 @@ export async function registerScreenerRoutes(app: FastifyInstance) { la.volume, la.price, la.change_24h, - COALESCE(pl.liquidity::float, 0) AS liquidity + COALESCE(pl.liquidity, 0) AS liquidity FROM latest_aggregates la LEFT JOIN pair_liquidity pl ON pl.pair_key = la.pair_key ) @@ -230,12 +230,22 @@ export async function registerScreenerRoutes(app: FastifyInstance) { let nextCursor = null if (hasMore && data.length > 0) { const lastRecord = data[data.length - 1] - const cursorPayload = JSON.stringify([lastRecord[sortBy], lastRecord.pair_key]) + const cursorValue = lastRecord[sortBy] + const cursorPayload = JSON.stringify([ + typeof cursorValue === 'string' && Number.isFinite(Number(cursorValue)) ? Number(cursorValue) : cursorValue, + lastRecord.pair_key, + ]) nextCursor = Buffer.from(cursorPayload).toString('base64') } return { - data, + data: data.map(row => ({ + ...row, + volume: Number(row.volume), + price: Number(row.price), + change_24h: Number(row.change_24h), + liquidity: Number(row.liquidity), + })), nextCursor, hasMore, total: null,