diff --git a/.changeset/graphql-exact-decimal-prices.md b/.changeset/graphql-exact-decimal-prices.md new file mode 100644 index 00000000..1dad38e8 --- /dev/null +++ b/.changeset/graphql-exact-decimal-prices.md @@ -0,0 +1,5 @@ +--- +'lens': minor +--- + +Expose exact decimal strings (`priceExact`, `volume24hExact`, `vwap1hExact`, and the other `*Exact` fields) on the GraphQL `AggregatedPrice` and `PriceBucket` types, so PostgreSQL numeric precision survives the GraphQL response. The matching `Float` fields are deprecated and unchanged. 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/schema.graphql b/schema.graphql index a71f3707..05742af7 100644 --- a/schema.graphql +++ b/schema.graphql @@ -1,3 +1,5 @@ +scalar Decimal + type Asset { code: String! issuer: String @@ -16,18 +18,29 @@ 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! @@ -47,15 +60,23 @@ type RouteInfo { 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 { diff --git a/src/__tests__/aggregateRefreshNetwork.test.ts b/src/__tests__/aggregateRefreshNetwork.test.ts index 8892cecd..b7afc950 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) @@ -133,14 +133,14 @@ describe('aggregateRefresh runs once per enabled network', () => { await processor({ data: { pairKey: mainnetPair.pairKey, pair: mainnetPair, network: 'mainnet' } }) - expect(mockGetAggregatedPrice).toHaveBeenCalledWith(mainnetPair.pairKey, 'mainnet') + expect(mockGetAggregatedPrice).toHaveBeenCalledWith(mainnetPair.pairKey, 'mainnet', { decimalFormat: 'string' }) 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__/graphqlQueryDepth.test.ts b/src/__tests__/graphqlQueryDepth.test.ts index 717321de..1673b28d 100644 --- a/src/__tests__/graphqlQueryDepth.test.ts +++ b/src/__tests__/graphqlQueryDepth.test.ts @@ -49,7 +49,11 @@ describe('GraphQL query depth limit', () => { beforeEach(async () => { process.env.GRAPHQL_MAX_QUERY_DEPTH = '2' vi.clearAllMocks() - vi.mocked(getCachedPrice).mockResolvedValue(JSON.stringify({ price: 0.1234 })) + vi.mocked(getCachedPrice).mockResolvedValue(JSON.stringify({ + price: 0.1234, priceExact: '0.1234', sdexPriceExact: '0.1234', ammPriceExact: '0', + volume24hExact: '0', sdexVolume24hExact: '0', ammVolume24hExact: '0', vwap1mExact: '0', + vwap5mExact: '0', vwap1hExact: '0', vwap24hExact: '0', priceChange24hExact: '0', + })) app = await buildServer() }) diff --git a/src/__tests__/graphqlSubscription.test.ts b/src/__tests__/graphqlSubscription.test.ts index b2061917..283008ff 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,33 @@ 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.123456789012345678', sdexPrice: '0.123456789012345678', ammPrice: '0', + volume24h: '0', sdexVolume24h: '0', ammVolume24h: '0', + vwap1m: '0', vwap5m: '0', vwap1h: '0', vwap24h: '0', + priceChange24h: '0', sources: 0, + confidence: 'unknown', lastTradeAgeSeconds: null, stale: false, + } as any) + 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..397ff4eb 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 ────── @@ -181,6 +181,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/priceFields.ts b/src/aggregator/priceFields.ts new file mode 100644 index 00000000..1d5588db --- /dev/null +++ b/src/aggregator/priceFields.ts @@ -0,0 +1,37 @@ +import type { AggregatedPriceString } from './vwap' + +/** + * Shapes a string-format aggregate for the GraphQL AggregatedPrice type: each + * `*Exact` field is the PostgreSQL numeric text, and the deprecated Float field + * of the same name is derived from it for existing clients. + */ +export function withExactPriceFields(agg: AggregatedPriceString) { + return { + price: Number(agg.price), + priceExact: agg.price, + sdexPrice: Number(agg.sdexPrice), + sdexPriceExact: agg.sdexPrice, + ammPrice: Number(agg.ammPrice), + ammPriceExact: agg.ammPrice, + volume24h: Number(agg.volume24h), + volume24hExact: agg.volume24h, + sdexVolume24h: Number(agg.sdexVolume24h), + sdexVolume24hExact: agg.sdexVolume24h, + ammVolume24h: Number(agg.ammVolume24h), + ammVolume24hExact: agg.ammVolume24h, + vwap1m: Number(agg.vwap1m), + vwap1mExact: agg.vwap1m, + vwap5m: Number(agg.vwap5m), + vwap5mExact: agg.vwap5m, + vwap1h: Number(agg.vwap1h), + vwap1hExact: agg.vwap1h, + vwap24h: Number(agg.vwap24h), + vwap24hExact: agg.vwap24h, + priceChange24h: Number(agg.priceChange24h), + priceChange24hExact: agg.priceChange24h, + sources: agg.sources, + confidence: agg.confidence, + lastTradeAgeSeconds: agg.lastTradeAgeSeconds, + stale: agg.stale, + } +} diff --git a/src/aggregator/vwap.ts b/src/aggregator/vwap.ts index 6ef4e777..37142173 100644 --- a/src/aggregator/vwap.ts +++ b/src/aggregator/vwap.ts @@ -56,6 +56,17 @@ export async function calculateVWAP( return detailed.num } +/** VWAP as PostgreSQL numeric text, with no Number round-trip. */ +export async function calculateVWAPExact( + pairKey: string, + windowMinutes: number, + network: NetworkName, + source?: 'SDEX' | 'AMM' +): Promise { + const detailed = await calculateVWAPDetailed(pairKey, windowMinutes, network, source) + return detailed.raw +} + export async function calculateOHLCV( pairKey: string, windowMinutes: number, @@ -63,6 +74,26 @@ export async function calculateOHLCV( ): Promise<{ open: number; high: number; low: number; close: number volume: number; tradeCount: number +}> { + const exact = await calculateOHLCVExact(pairKey, windowMinutes, network) + return { + open: parseFloat(exact.open), + high: parseFloat(exact.high), + low: parseFloat(exact.low), + close: parseFloat(exact.close), + volume: parseFloat(exact.volume), + tradeCount: exact.tradeCount, + } +} + +/** OHLCV with prices and volume as PostgreSQL numeric text. */ +export async function calculateOHLCVExact( + pairKey: string, + windowMinutes: number, + network: NetworkName +): Promise<{ + open: string; high: string; low: string; close: string + volume: string; tradeCount: number }> { const result = await pgPool.query( `SELECT @@ -86,11 +117,11 @@ 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: String(row?.open ?? '0'), + high: String(row?.high ?? '0'), + low: String(row?.low ?? '0'), + close: String(row?.close ?? '0'), + volume: String(row?.volume ?? '0'), tradeCount: parseInt(row?.trade_count ?? '0', 10), } } @@ -398,3 +429,4 @@ export async function getAggregatedPrice( } } + diff --git a/src/api/graphql.ts b/src/api/graphql.ts index d258b158..a84c0a27 100644 --- a/src/api/graphql.ts +++ b/src/api/graphql.ts @@ -1,11 +1,11 @@ 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' +import { withExactPriceFields } from '../aggregator/priceFields' import { getBestRoute } from '../aggregator/bestRoute' import { pgPool } from '../db' import { config, activeNetwork, type NetworkName } from '../config' @@ -80,6 +80,8 @@ export function formatGraphQLErrors(execution: ExecutionResult, ctx: MercuriusCo } export const schema = ` + scalar Decimal + type Asset { code: String! issuer: String @@ -98,18 +100,29 @@ export const schema = ` 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! @@ -129,15 +142,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 { @@ -322,6 +343,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() @@ -352,10 +394,20 @@ 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) + const agg = withExactPriceFields( + await getAggregatedPrice(pairKey, activeNetwork, { decimalFormat: 'string' }) + ) const route = await getBestRoute(pair.assetA, pair.assetB, pairKey, 1000, activeNetwork) return { assetA, assetB, pairKey, ...agg, @@ -444,9 +496,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 @@ -456,15 +507,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 70d8f61e..42c695d7 100644 --- a/src/api/history.ts +++ b/src/api/history.ts @@ -86,8 +86,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/jobs/aggregateRefresh.ts b/src/jobs/aggregateRefresh.ts index 4b49b9b7..9806ef4a 100644 --- a/src/jobs/aggregateRefresh.ts +++ b/src/jobs/aggregateRefresh.ts @@ -1,8 +1,10 @@ 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 { withExactPriceFields } from '../aggregator/priceFields' import { getBestRoute } from '../aggregator/bestRoute' import { createLogger } from '../logger' @@ -35,7 +37,7 @@ export function startAggregateWorker(network: NetworkName) { async (job) => { const { pairKey, pair } = job.data try { - const agg = await getAggregatedPrice(pairKey, network) + const agg = await getAggregatedPrice(pairKey, network, { decimalFormat: 'string' }) const route = await getBestRoute(pair.assetA, pair.assetB, pairKey, 1000, network) const result = { @@ -46,7 +48,7 @@ export function startAggregateWorker(network: NetworkName) { // the payload has to carry the same network the route stamps — a // cache hit is served verbatim, network field included. network, - ...agg, + ...withExactPriceFields(agg), bestRoute: route.route, lastUpdated: new Date(), } @@ -69,27 +71,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 0ef20b6f..863ffbb9 100644 --- a/src/routes/candles.ts +++ b/src/routes/candles.ts @@ -60,11 +60,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 network = $5