From 48a5fc50aad832b638091378b7d2e1f3117254e6 Mon Sep 17 00:00:00 2001 From: phatloi0010 <316136531+phatloi0010@users.noreply.github.com> Date: Tue, 3 Nov 2026 09:30:14 +0700 Subject: [PATCH] feat(pricing): add staleness-aware median pricing with priority source fallback (#81) --- README.md | 2 + clients/typescript/src/schema.d.ts | 65 ++++++ docs/what-lens-knows.md | 4 +- openapi.json | 70 +++++++ openapi.yaml | 47 +++++ src/__tests__/medianPricing.test.ts | 314 ++++++++++++++++++++++++++++ src/api/rest.ts | 88 ++++++++ src/pricing/median.ts | 268 ++++++++++++++++++++++++ 8 files changed, 857 insertions(+), 1 deletion(-) create mode 100644 src/__tests__/medianPricing.test.ts create mode 100644 src/pricing/median.ts diff --git a/README.md b/README.md index ac7ef3e6..51ada868 100644 --- a/README.md +++ b/README.md @@ -101,6 +101,7 @@ npx @modelcontextprotocol/inspector npm run mcp:server | GET | `/v1/price/:assetA/:assetB` | Get aggregated price for a pair | | GET | `/v1/premium/:assetA/:assetB` | Get on-chain vs CEX price premium and spread | | GET | `/v1/price/:assetA/:assetB/route` | Get best swap route for a pair and amount | +| GET | `/v1/price/:assetA/:assetB/median` | Get staleness-aware median price across fresh sources with priority fallback | | GET | `/v1/price/:asset/fiat` | Convert asset price to fiat currency with NGN support (#290) | | GET | `/v1/price/:assetA/:assetB/history` | Get price history buckets | | GET | `/v1/pools` | Get AMM liquidity pool reserves and spot prices | @@ -204,6 +205,7 @@ the committed table differs from a fresh generation. | How do I submit and settle an x402 payment on the Stellar network? | `POST /v1/settle` | `POST /settle` | | How can my application receive push alerts when a price crosses a threshold? | `POST /v1/webhooks` | `POST /webhooks` | | How do I cancel an existing price threshold alert webhook? | `DELETE /v1/webhooks/:id` | `DELETE /webhooks/wh_9f8e7d6c5b4a3a2b1c?network=testnet` | +| What is the staleness-aware median price across venues, and which sources were included or excluded? | `GET /v1/price/:assetA/:assetB/median` | `GET /price/XLM/USDC/median?freshness=300&network=testnet` | Every non-internal route above has a matching entry in [`openapi.yaml`](openapi.yaml); `tests/openapi.test.ts` boots the server and fails if a registered route is diff --git a/clients/typescript/src/schema.d.ts b/clients/typescript/src/schema.d.ts index cfdbc158..a45cc4b9 100644 --- a/clients/typescript/src/schema.d.ts +++ b/clients/typescript/src/schema.d.ts @@ -198,6 +198,23 @@ export interface paths { patch?: never; trace?: never; }; + "/v1/price/{assetA}/{assetB}/median": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + /** Get staleness-aware median price across fresh sources with priority fallback */ + get: operations["getMedianPrice"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/v1/price/{asset}/fiat": { parameters: { query?: never; @@ -2597,6 +2614,54 @@ export interface operations { 429: components["responses"]["TooManyRequests"]; }; }; + getMedianPrice: { + parameters: { + query?: { + /** @description Freshness threshold in seconds (default: 300, valid range: 5 to 86400) */ + freshness?: number; + /** @description Stellar network to answer for — `testnet` or `mainnet`. Defaults to this instance's configured `STELLAR_NETWORK`. An unrecognised value is rejected with 400. DB-backed reads are currently served from whichever network this instance indexes, even when another is requested. */ + network?: components["parameters"]["NetworkQuery"]; + }; + header?: { + /** @description Formatting for decimal quantities (prices, volumes, reserves). Set to `string` to serialize high-precision `Decimal(36,18)` quantities as exact strings, preventing IEEE-754 floating-point precision loss. Defaults to `number`. Can also be negotiated via `Accept: application/json; decimals=strings` or query parameter `?decimals=strings`. */ + "X-Decimal-Format"?: components["parameters"]["DecimalFormatHeader"]; + }; + path: { + /** @example XLM */ + assetA: string; + /** @example USDC */ + assetB: string; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Staleness-aware median price result with included and excluded sources */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": Record; + }; + }; + /** @description Invalid parameters */ + 400: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + /** @description Pair not watched or no price data available */ + 404: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + 429: components["responses"]["TooManyRequests"]; + }; + }; getFiatPrice: { parameters: { query?: { diff --git a/docs/what-lens-knows.md b/docs/what-lens-knows.md index 1ce0e92d..1e959119 100644 --- a/docs/what-lens-knows.md +++ b/docs/what-lens-knows.md @@ -10,7 +10,7 @@ Instead of an arbitrary alphabetized URL list, this page organizes what Lens kno A developer arriving at Lens does not browse by HTTP verb or path parameter. Their exploration follows a natural progression: -1. **Core Pricing & Trade Execution (Steps 1–6)**: +1. **Core Pricing & Trade Execution (Steps 1–53)**: The universal entry question: *"What is an asset trading at right now?"* (`/price/:assetA/:assetB`). Once price is known, the immediate follow-up is execution: *"Where do I get the best return for my trade size?"* (`/price/:assetA/:assetB/route`), *"What does combined order-book depth look like?"* (`/book/:assetA/:assetB`), *"How much slippage will I incur?"* (`/price/:assetA/:assetB/depth`), and how that slippage grows with size (`/impact/:assetA/:assetB`). Pricing in a local currency rather than dollars lives here too (`/price/:asset/fiat`). 2. **Market Overview & Liquidity Pools (Steps 7–17)**: @@ -89,6 +89,7 @@ A developer arriving at Lens does not browse by HTTP verb or path parameter. The | 50 | How do I submit and settle an x402 payment on the Stellar network? | `POST /v1/settle` | `POST /settle` | Automation, Discovery & Payments | | 51 | How can my application receive push alerts when a price crosses a threshold? | `POST /v1/webhooks` | `POST /webhooks` | Automation, Discovery & Payments | | 52 | How do I cancel an existing price threshold alert webhook? | `DELETE /v1/webhooks/:id` | `DELETE /webhooks/wh_9f8e7d6c5b4a3a2b1c?network=testnet` | Automation, Discovery & Payments | +| 53 | What is the staleness-aware median price across venues, and which sources were included or excluded? | `GET /v1/price/:assetA/:assetB/median` | `GET /price/XLM/USDC/median?freshness=300&network=testnet` | Core Pricing & Trade Execution | --- @@ -104,6 +105,7 @@ A developer arriving at Lens does not browse by HTTP verb or path parameter. The | **What is the simulated order-book depth and expected slippage for a trade size?** | `GET /v1/price/:assetA/:assetB/depth` | `GET /price/XLM/USDC/depth?amount=1000&network=testnet` | Get order-book depth and execution slippage for a pair | | **How does slippage grow as my trade size increases?** | `GET /v1/impact/:assetA/:assetB` | `GET /impact/XLM/USDC?network=testnet` | Get an adaptive slippage impact curve for a pair | | **What is this asset worth in my local currency, such as naira?** | `GET /v1/price/:asset/fiat` | `GET /price/XLM/fiat?currency=NGN&network=testnet` | Convert asset price to fiat currency with NGN support (#290) | +| **What is the staleness-aware median price across venues, and which sources were included or excluded?** | `GET /v1/price/:assetA/:assetB/median` | `GET /price/XLM/USDC/median?freshness=300&network=testnet` | Get staleness-aware median price across fresh sources with priority fallback | ### Market Overview & Liquidity Pools diff --git a/openapi.json b/openapi.json index 57def8ee..e7b88be8 100644 --- a/openapi.json +++ b/openapi.json @@ -791,6 +791,76 @@ } } }, + "/v1/price/{assetA}/{assetB}/median": { + "get": { + "operationId": "getMedianPrice", + "summary": "Get staleness-aware median price across fresh sources with priority fallback", + "x-question": "What is the staleness-aware median price across venues, and which sources were included or excluded?", + "x-example": "GET /price/XLM/USDC/median?freshness=300&network=testnet", + "x-order": 53, + "x-category": "Core Pricing & Trade Execution", + "security": [], + "parameters": [ + { + "name": "assetA", + "in": "path", + "required": true, + "schema": { + "type": "string" + }, + "example": "XLM" + }, + { + "name": "assetB", + "in": "path", + "required": true, + "schema": { + "type": "string" + }, + "example": "USDC" + }, + { + "name": "freshness", + "in": "query", + "required": false, + "description": "Freshness threshold in seconds (default: 300, valid range: 5 to 86400)", + "schema": { + "type": "integer", + "default": 300, + "minimum": 5, + "maximum": 86400 + } + }, + { + "$ref": "#/components/parameters/NetworkQuery" + }, + { + "$ref": "#/components/parameters/DecimalFormatHeader" + } + ], + "responses": { + "200": { + "description": "Staleness-aware median price result with included and excluded sources", + "content": { + "application/json": { + "schema": { + "type": "object" + } + } + } + }, + "400": { + "description": "Invalid parameters" + }, + "404": { + "description": "Pair not watched or no price data available" + }, + "429": { + "$ref": "#/components/responses/TooManyRequests" + } + } + } + }, "/v1/price/{asset}/fiat": { "get": { "x-question": "What is this asset worth in my local currency, such as naira?", diff --git a/openapi.yaml b/openapi.yaml index 3cbd81ac..b1b5ab81 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -587,6 +587,53 @@ paths: '429': $ref: '#/components/responses/TooManyRequests' + /v1/price/{assetA}/{assetB}/median: + get: + operationId: getMedianPrice + summary: Get staleness-aware median price across fresh sources with priority fallback + x-question: "What is the staleness-aware median price across venues, and which sources were included or excluded?" + x-example: "GET /price/XLM/USDC/median?freshness=300&network=testnet" + x-order: 53 + x-category: "Core Pricing & Trade Execution" + security: [] + parameters: + - name: assetA + in: path + required: true + schema: + type: string + example: XLM + - name: assetB + in: path + required: true + schema: + type: string + example: USDC + - name: freshness + in: query + required: false + description: 'Freshness threshold in seconds (default: 300, valid range: 5 to 86400)' + schema: + type: integer + default: 300 + minimum: 5 + maximum: 86400 + - $ref: '#/components/parameters/NetworkQuery' + - $ref: '#/components/parameters/DecimalFormatHeader' + responses: + '200': + description: Staleness-aware median price result with included and excluded sources + content: + application/json: + schema: + type: object + '400': + description: Invalid parameters + '404': + description: Pair not watched or no price data available + '429': + $ref: '#/components/responses/TooManyRequests' + /v1/price/{asset}/fiat: get: x-question: "What is this asset worth in my local currency, such as naira?" diff --git a/src/__tests__/medianPricing.test.ts b/src/__tests__/medianPricing.test.ts new file mode 100644 index 00000000..771bc1a9 --- /dev/null +++ b/src/__tests__/medianPricing.test.ts @@ -0,0 +1,314 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest' +import Fastify from 'fastify' + +const { mockQuery, mockGetCachedPrice, mockGetBestRoute, mockWatchedPairs } = vi.hoisted(() => ({ + mockQuery: vi.fn(), + mockGetCachedPrice: vi.fn(), + mockGetBestRoute: vi.fn(), + mockWatchedPairs: [ + { + pairKey: 'USDC/XLM', + assetA: { code: 'XLM', issuer: null }, + assetB: { + code: 'USDC', + issuer: 'GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5', + }, + }, + ], +})) + +vi.mock('../db', () => ({ + pgPool: { query: mockQuery }, +})) + +vi.mock('../redis', () => ({ + getCachedPrice: mockGetCachedPrice, + setCachedPrice: vi.fn(), +})) + +vi.mock('../aggregator/bestRoute', () => ({ + getBestRoute: mockGetBestRoute, +})) + +vi.mock('../config', async (importOriginal) => { + const actual = await importOriginal() + return { + ...actual, + activeNetwork: 'testnet', + getNetworkConfig: (network: 'testnet' | 'mainnet') => ({ + ...actual.getNetworkConfig(network), + pairs: mockWatchedPairs, + }), + } +}) + +import { + calculateMedian, + clampFreshnessThreshold, + computeStalenessAwareMedian, + getStalenessAwareMedianPrice, + type PriceSourceQuote, +} from '../pricing/median' +import { registerRESTRoutes } from '../api/rest' +import { registerNetworkSelector } from '../middleware/network' + +describe('Staleness-aware median pricing with source fallback (#81)', () => { + const fixedNow = 1_700_000_000_000 + + beforeEach(() => { + vi.clearAllMocks() + }) + + describe('calculateMedian & clampFreshnessThreshold', () => { + it('returns null on an empty array instead of 0', () => { + expect(calculateMedian([])).toBeNull() + }) + + it('computes the median for single, odd, and even arrays', () => { + expect(calculateMedian([42])).toBe(42) + expect(calculateMedian([10, 50, 20])).toBe(20) + expect(calculateMedian([10, 20, 30, 40])).toBe(25) + }) + + it('clamps freshness threshold between 5 and 86400 seconds', () => { + expect(clampFreshnessThreshold()).toBe(300) + expect(clampFreshnessThreshold(1)).toBe(5) + expect(clampFreshnessThreshold(100_000)).toBe(86_400) + }) + }) + + describe('computeStalenessAwareMedian', () => { + it('returns null medianPrice and reason no_price_data when quotes array is empty', () => { + const result = computeStalenessAwareMedian([], { now: fixedNow }) + expect(result.medianPrice).toBeNull() + expect(result.sourceCount).toBe(0) + expect(result.fallbackUsed).toBe(false) + expect(result.fallbackSource).toBeNull() + expect(result.fallbackAgeSeconds).toBeNull() + expect(result.reason).toBe('no_price_data') + }) + + it('returns null medianPrice and reason all_sources_invalid when every source has an invalid price or timestamp', () => { + const quotes: PriceSourceQuote[] = [ + { source: 'SDEX', price: 0, timestamp: fixedNow - 10_000 }, + { source: 'AMM', price: -1.5, timestamp: fixedNow - 20_000 }, + { source: 'SOROSWAP', price: 0.12, timestamp: 'not-a-timestamp' }, + ] + + const result = computeStalenessAwareMedian(quotes, { now: fixedNow }) + expect(result.medianPrice).toBeNull() + expect(result.sourceCount).toBe(0) + expect(result.fallbackUsed).toBe(false) + expect(result.reason).toBe('all_sources_invalid') + expect(result.excludedSources).toHaveLength(3) + expect(result.excludedSources[0].reason).toBe('invalid_price') + expect(result.excludedSources[1].reason).toBe('invalid_price') + expect(result.excludedSources[2].reason).toBe('invalid_timestamp') + }) + + it('computes the median strictly from fresh sources and annotates excluded stale sources', () => { + const quotes: PriceSourceQuote[] = [ + { source: 'SDEX', price: 1.05, timestamp: fixedNow - 60_000 }, + { source: 'AMM', price: 1.08, timestamp: fixedNow - 120_000 }, + { source: 'SOROSWAP', price: 1.02, timestamp: fixedNow - 30_000 }, + { source: 'REFLECTOR', price: 2.5, timestamp: fixedNow - 600_000 }, // stale (>300s) + ] + + const result = computeStalenessAwareMedian(quotes, { + freshnessThresholdSeconds: 300, + now: fixedNow, + }) + + expect(result.medianPrice).toBe(1.05) + expect(result.sourceCount).toBe(3) + expect(result.fallbackUsed).toBe(false) + expect(result.fallbackSource).toBeNull() + expect(result.fallbackAgeSeconds).toBeNull() + expect(result.reason).toBeNull() + expect(result.includedSources.map((s) => s.source)).toEqual(['SDEX', 'AMM', 'SOROSWAP']) + expect(result.excludedSources).toEqual([ + { + source: 'REFLECTOR', + price: 2.5, + ageSeconds: 600, + reason: 'stale', + }, + ]) + }) + + it('engages priority fallback chain when all valid sources are stale and reports fallbackAgeSeconds', () => { + const quotes: PriceSourceQuote[] = [ + { source: 'AQUARIUS', price: 1.1, timestamp: fixedNow - 900_000 }, + { source: 'AMM', price: 1.04, timestamp: fixedNow - 500_000 }, + { source: 'SOROSWAP', price: 1.06, timestamp: fixedNow - 400_000 }, + ] + + const result = computeStalenessAwareMedian(quotes, { + freshnessThresholdSeconds: 300, + fallbackPriority: ['SDEX', 'AMM', 'SOROSWAP', 'AQUARIUS'], + now: fixedNow, + }) + + expect(result.fallbackUsed).toBe(true) + expect(result.fallbackSource).toBe('AMM') + expect(result.fallbackAgeSeconds).toBe(500) + expect(result.medianPrice).toBe(1.04) + expect(result.sourceCount).toBe(1) + expect(result.includedSources).toEqual([ + { source: 'AMM', price: 1.04, ageSeconds: 500 }, + ]) + expect(result.excludedSources).toHaveLength(3) + expect(result.reason).toBeNull() + }) + + it('falls back to the freshest valid stale source when none match the priority list', () => { + const quotes: PriceSourceQuote[] = [ + { source: 'CUSTOM_VENUE_A', price: 0.95, timestamp: fixedNow - 800_000 }, + { source: 'CUSTOM_VENUE_B', price: 0.99, timestamp: fixedNow - 450_000 }, + ] + + const result = computeStalenessAwareMedian(quotes, { + freshnessThresholdSeconds: 300, + fallbackPriority: ['SDEX', 'AMM'], + now: fixedNow, + }) + + expect(result.fallbackUsed).toBe(true) + expect(result.fallbackSource).toBe('CUSTOM_VENUE_B') + expect(result.fallbackAgeSeconds).toBe(450) + expect(result.medianPrice).toBe(0.99) + }) + }) + + describe('getStalenessAwareMedianPrice database query', () => { + it('queries price_points scoped by pair_key and network', async () => { + mockQuery.mockResolvedValueOnce({ + rows: [ + { source: 'SDEX', price: '0.12', observed_at: new Date() }, + { source: 'AMM', price: '0.14', observed_at: new Date() }, + ], + }) + + const res = await getStalenessAwareMedianPrice('USDC/XLM', 'testnet', { + freshnessThresholdSeconds: 300, + }) + + expect(mockQuery).toHaveBeenCalledWith( + expect.stringContaining('WHERE pair_key = $1'), + ['USDC/XLM', 'testnet'] + ) + expect(res.medianPrice).toBe(0.13) + expect(res.sourceCount).toBe(2) + expect(res.fallbackUsed).toBe(false) + }) + }) + + describe('GET /price/:assetA/:assetB/median route', () => { + async function buildTestApp() { + const app = Fastify({ logger: false }) + await app.register(registerNetworkSelector) + await registerRESTRoutes(app) + await app.ready() + return app + } + + it('returns 404 when the requested pair is not watched', async () => { + const app = await buildTestApp() + const res = await app.inject({ + method: 'GET', + url: '/price/BTC/ETH/median', + }) + expect(res.statusCode).toBe(404) + expect(res.json().error).toMatch(/not watched/) + await app.close() + }) + + it('returns 400 when freshness parameter is out of [5, 86400] bounds or non-integer', async () => { + const app = await buildTestApp() + const tooLow = await app.inject({ + method: 'GET', + url: '/price/XLM/USDC/median?freshness=4', + }) + expect(tooLow.statusCode).toBe(400) + expect(tooLow.json().error).toBe( + 'freshness must be an integer between 5 and 86400 seconds' + ) + + const tooHigh = await app.inject({ + method: 'GET', + url: '/price/XLM/USDC/median?freshness=100000', + }) + expect(tooHigh.statusCode).toBe(400) + + const nonNumeric = await app.inject({ + method: 'GET', + url: '/price/XLM/USDC/median?freshness=abc', + }) + expect(nonNumeric.statusCode).toBe(400) + await app.close() + }) + + it('returns 404 with reason no_price_data when no price rows exist for a watched pair', async () => { + mockQuery.mockResolvedValueOnce({ rows: [] }) + const app = await buildTestApp() + const res = await app.inject({ + method: 'GET', + url: '/price/XLM/USDC/median', + }) + expect(res.statusCode).toBe(404) + const body = res.json() + expect(body.medianPrice).toBeNull() + expect(body.reason).toBe('no_price_data') + expect(body.sourceCount).toBe(0) + await app.close() + }) + + it('returns 404 with reason all_sources_invalid when all DB rows have invalid prices', async () => { + mockQuery.mockResolvedValueOnce({ + rows: [{ source: 'SDEX', price: '0', observed_at: new Date().toISOString() }], + }) + const app = await buildTestApp() + const res = await app.inject({ + method: 'GET', + url: '/price/XLM/USDC/median', + }) + expect(res.statusCode).toBe(404) + const body = res.json() + expect(body.medianPrice).toBeNull() + expect(body.reason).toBe('all_sources_invalid') + await app.close() + }) + + it('returns 200 with staleness-aware median and included/excluded source annotations', async () => { + const now = Date.now() + mockQuery.mockResolvedValueOnce({ + rows: [ + { source: 'SDEX', price: '0.12', observed_at: new Date(now - 30_000).toISOString() }, + { source: 'AMM', price: '0.14', observed_at: new Date(now - 60_000).toISOString() }, + { + source: 'SOROSWAP', + price: '0.25', + observed_at: new Date(now - 900_000).toISOString(), + }, + ], + }) + const app = await buildTestApp() + const res = await app.inject({ + method: 'GET', + url: '/price/XLM/USDC/median?freshness=300', + }) + expect(res.statusCode).toBe(200) + const body = res.json() + expect(body.medianPrice).toBe(0.13) + expect(body.sourceCount).toBe(2) + expect(body.fallbackUsed).toBe(false) + expect(body.includedSources).toHaveLength(2) + expect(body.excludedSources).toHaveLength(1) + expect(body.excludedSources[0]).toMatchObject({ + source: 'SOROSWAP', + reason: 'stale', + }) + await app.close() + }) + }) +}) diff --git a/src/api/rest.ts b/src/api/rest.ts index 8b70d059..df682dfe 100644 --- a/src/api/rest.ts +++ b/src/api/rest.ts @@ -23,6 +23,11 @@ import { } from './schemas' import { findOverlappingGaps } from '../backfill/gaps' import { getDepth, getImpact } from '../pricing/depth' +import { + getStalenessAwareMedianPrice, + MIN_FRESHNESS_THRESHOLD_SECONDS, + MAX_FRESHNESS_THRESHOLD_SECONDS, +} from '../pricing/median' import { getRetentionPath } from '../jobs/retentionStatus' import { apiError } from '../errors' @@ -269,6 +274,89 @@ export async function registerRESTRoutes(app: FastifyInstance) { } ) + // GET /price/:assetA/:assetB/median + app.get<{ + Params: { assetA: string; assetB: string } + Querystring: { + freshness?: string + freshnessSec?: string + staleThresholdSeconds?: string + decimals?: string + decimalFormat?: string + } + }>( + '/price/:assetA/:assetB/median', + { config: { scope: 'read:prices' } }, + async (req, reply) => { + price_requests_total.inc() + const { assetA, assetB } = req.params + const network = req.network ?? activeNetwork + + const rawFreshness = + req.query?.freshness ?? req.query?.freshnessSec ?? req.query?.staleThresholdSeconds + let freshnessThresholdSeconds: number | undefined + if (rawFreshness !== undefined) { + const parsed = Number(rawFreshness) + if ( + !Number.isInteger(parsed) || + parsed < MIN_FRESHNESS_THRESHOLD_SECONDS || + parsed > MAX_FRESHNESS_THRESHOLD_SECONDS + ) { + return reply + .status(400) + .send( + apiError( + 'invalid_parameter', + `freshness must be an integer between ${MIN_FRESHNESS_THRESHOLD_SECONDS} and ${MAX_FRESHNESS_THRESHOLD_SECONDS} seconds` + ) + ) + } + freshnessThresholdSeconds = parsed + } + + const pair = findPair(assetA, assetB, network) + if (!pair) { + return reply + .status(404) + .send(apiError('unwatched_pair', `Pair ${assetA}/${assetB} not watched on ${network}`)) + } + + const decimalFormat = resolveDecimalFormat(req) + if (decimalFormat === 'string') { + reply.header('x-decimal-format', 'string') + } + + const medianResult = await getStalenessAwareMedianPrice(pair.pairKey, network, { + freshnessThresholdSeconds, + }) + + if (medianResult.medianPrice === null) { + return reply.status(404).send({ + ...medianResult, + error: `No price data available for pair ${pair.pairKey} on ${network}`, + code: 'NOT_FOUND', + assetA: pair.assetA.code, + assetB: pair.assetB.code, + pairKey: pair.pairKey, + network, + }) + } + + return { + ...medianResult, + medianPrice: + decimalFormat === 'string' + ? formatDecimal(medianResult.medianPrice, decimalFormat) + : medianResult.medianPrice, + assetA: pair.assetA.code, + assetB: pair.assetB.code, + pairKey: pair.pairKey, + network, + lastUpdated: new Date().toISOString(), + } + } + ) + // GET /price/:assetA/:assetB/route?amount=1000 app.get<{ Params: { assetA: string; assetB: string } diff --git a/src/pricing/median.ts b/src/pricing/median.ts new file mode 100644 index 00000000..593ad972 --- /dev/null +++ b/src/pricing/median.ts @@ -0,0 +1,268 @@ +import { pgPool } from '../db' +import { activeNetwork, type NetworkName } from '../config' + +export const DEFAULT_FRESHNESS_THRESHOLD_SECONDS = 300 +export const MIN_FRESHNESS_THRESHOLD_SECONDS = 5 +export const MAX_FRESHNESS_THRESHOLD_SECONDS = 86_400 + +export const DEFAULT_FALLBACK_PRIORITY: readonly string[] = [ + 'SDEX', + 'AMM', + 'SOROSWAP', + 'PHOENIX', + 'AQUARIUS', + 'REFLECTOR', + 'DEFINDEX', +] + +export interface PriceSourceQuote { + source: string + price: number + timestamp: Date | string | number +} + +export interface IncludedPriceSource { + source: string + price: number + ageSeconds: number +} + +export type ExcludedSourceReason = 'stale' | 'invalid_price' | 'invalid_timestamp' + +export interface ExcludedPriceSource { + source: string + price: number + ageSeconds: number | null + reason: ExcludedSourceReason +} + +export type MedianEmptyReason = 'no_price_data' | 'all_sources_invalid' | null + +export interface StalenessAwareMedianOptions { + freshnessThresholdSeconds?: number + fallbackPriority?: readonly string[] + now?: Date | number +} + +export interface StalenessAwareMedianResult { + medianPrice: number | null + sourceCount: number + includedSources: IncludedPriceSource[] + excludedSources: ExcludedPriceSource[] + fallbackUsed: boolean + fallbackSource: string | null + fallbackAgeSeconds: number | null + freshnessThresholdSeconds: number + reason: MedianEmptyReason +} + +/** + * Compute the statistical median of a list of finite numbers. + * Returns `null` when no valid numbers are provided rather than `0`, since `0` + * is indistinguishable from a zero-priced quote. + */ +export function calculateMedian(values: number[]): number | null { + const valid = values.filter((v) => Number.isFinite(v)) + if (valid.length === 0) return null + const sorted = [...valid].sort((a, b) => a - b) + const mid = Math.floor(sorted.length / 2) + if (sorted.length % 2 === 1) { + return sorted[mid] + } + return (sorted[mid - 1] + sorted[mid]) / 2 +} + +/** + * Clamp a freshness threshold into `[MIN_FRESHNESS_THRESHOLD_SECONDS, MAX_FRESHNESS_THRESHOLD_SECONDS]`. + */ +export function clampFreshnessThreshold(seconds?: number): number { + if (seconds === undefined || !Number.isFinite(seconds)) { + return DEFAULT_FRESHNESS_THRESHOLD_SECONDS + } + const rounded = Math.round(seconds) + return Math.min( + Math.max(rounded, MIN_FRESHNESS_THRESHOLD_SECONDS), + MAX_FRESHNESS_THRESHOLD_SECONDS + ) +} + +function toEpochMs(ts: Date | string | number): number { + if (ts instanceof Date) return ts.getTime() + if (typeof ts === 'number') return ts + return new Date(ts).getTime() +} + +/** + * Aggregate multiple price sources into a staleness-aware median: + * - Drops sources older than `freshnessThresholdSeconds` (default 300s) or with invalid values + * - Computes the median strictly across fresh sources when any are available + * - Engages a deterministic priority fallback chain when all valid sources are stale + * - Annotates `includedSources`, `excludedSources` (with `reason` and `ageSeconds`), + * and fallback metadata (`fallbackUsed`, `fallbackSource`, `fallbackAgeSeconds`). + */ +export function computeStalenessAwareMedian( + quotes: PriceSourceQuote[], + options: StalenessAwareMedianOptions = {} +): StalenessAwareMedianResult { + const freshnessThresholdSeconds = clampFreshnessThreshold(options.freshnessThresholdSeconds) + const priorityChain = options.fallbackPriority ?? DEFAULT_FALLBACK_PRIORITY + const nowMs = + options.now instanceof Date + ? options.now.getTime() + : typeof options.now === 'number' + ? options.now + : Date.now() + + if (quotes.length === 0) { + return { + medianPrice: null, + sourceCount: 0, + includedSources: [], + excludedSources: [], + fallbackUsed: false, + fallbackSource: null, + fallbackAgeSeconds: null, + freshnessThresholdSeconds, + reason: 'no_price_data', + } + } + + const includedSources: IncludedPriceSource[] = [] + const excludedSources: ExcludedPriceSource[] = [] + const staleCandidates: IncludedPriceSource[] = [] + + for (const quote of quotes) { + const tsMs = toEpochMs(quote.timestamp) + const hasValidTs = Number.isFinite(tsMs) + const ageSeconds = hasValidTs ? Math.max(0, Math.round((nowMs - tsMs) / 1000)) : null + + if (!Number.isFinite(quote.price) || quote.price <= 0) { + excludedSources.push({ + source: quote.source, + price: quote.price, + ageSeconds, + reason: 'invalid_price', + }) + continue + } + + if (!hasValidTs || ageSeconds === null) { + excludedSources.push({ + source: quote.source, + price: quote.price, + ageSeconds: null, + reason: 'invalid_timestamp', + }) + continue + } + + if (ageSeconds <= freshnessThresholdSeconds) { + includedSources.push({ + source: quote.source, + price: quote.price, + ageSeconds, + }) + } else { + excludedSources.push({ + source: quote.source, + price: quote.price, + ageSeconds, + reason: 'stale', + }) + staleCandidates.push({ + source: quote.source, + price: quote.price, + ageSeconds, + }) + } + } + + if (includedSources.length > 0) { + const medianPrice = calculateMedian(includedSources.map((s) => s.price)) + return { + medianPrice, + sourceCount: includedSources.length, + includedSources, + excludedSources, + fallbackUsed: false, + fallbackSource: null, + fallbackAgeSeconds: null, + freshnessThresholdSeconds, + reason: null, + } + } + + if (staleCandidates.length === 0) { + return { + medianPrice: null, + sourceCount: 0, + includedSources: [], + excludedSources, + fallbackUsed: false, + fallbackSource: null, + fallbackAgeSeconds: null, + freshnessThresholdSeconds, + reason: 'all_sources_invalid', + } + } + + // Fallback chain: pick the highest-priority stale source (freshest if multiple + // quotes share the same source tag), or the most recent valid stale source if + // none match the configured priority order. + let chosenFallback: IncludedPriceSource | undefined + for (const prioritySource of priorityChain) { + const matching = staleCandidates + .filter((c) => c.source.toUpperCase() === prioritySource.toUpperCase()) + .sort((a, b) => a.ageSeconds - b.ageSeconds) + if (matching.length > 0) { + chosenFallback = matching[0] + break + } + } + + if (!chosenFallback) { + chosenFallback = [...staleCandidates].sort((a, b) => a.ageSeconds - b.ageSeconds)[0] + } + + return { + medianPrice: chosenFallback.price, + sourceCount: 1, + includedSources: [chosenFallback], + excludedSources, + fallbackUsed: true, + fallbackSource: chosenFallback.source, + fallbackAgeSeconds: chosenFallback.ageSeconds, + freshnessThresholdSeconds, + reason: null, + } +} + +/** + * Query the latest observation per source for `(pairKey, network)` from + * `price_points` and compute the staleness-aware median with fallback. + */ +export async function getStalenessAwareMedianPrice( + pairKey: string, + network: NetworkName = activeNetwork, + options: StalenessAwareMedianOptions = {} +): Promise { + const result = await pgPool.query( + `SELECT DISTINCT ON (source) + source, + price::numeric AS price, + COALESCE(observed_at, timestamp) AS observed_at + FROM price_points + WHERE pair_key = $1 + AND network = $2 + ORDER BY source, COALESCE(observed_at, timestamp) DESC`, + [pairKey, network] + ) + + const quotes: PriceSourceQuote[] = result.rows.map((row: any) => ({ + source: String(row.source), + price: Number(row.price), + timestamp: row.observed_at, + })) + + return computeStalenessAwareMedian(quotes, options) +}