From 9601518e1a39d8bb4c1cd9d9cb29c76cb027e155 Mon Sep 17 00:00:00 2001 From: Riri-dev1 Date: Tue, 20 Oct 2026 21:19:50 +0100 Subject: [PATCH] fix(ingest): add rolling-median plausibility check and quarantine table (#238) --- .changeset/ingest-plausibility-quarantine.md | 5 + .env.example | 6 + .../migration.sql | 28 ++ prisma/schema.prisma | 24 ++ src/__tests__/ingestPlausibility.test.ts | 315 ++++++++++++++++++ src/__tests__/pricePointIdentity.test.ts | 16 +- src/config.ts | 9 + src/db.ts | 100 ++++-- src/ingest/plausibility.ts | 271 +++++++++++++++ src/ingest/venues/amm.ts | 109 +++--- src/ingest/venues/runner.ts | 9 +- src/ingest/venues/sdex.ts | 55 +-- src/ingest/venues/soroswap.ts | 37 +- src/metrics.ts | 16 + src/pricing/twap.ts | 2 +- 15 files changed, 875 insertions(+), 127 deletions(-) create mode 100644 .changeset/ingest-plausibility-quarantine.md create mode 100644 prisma/migrations/20261027000000_add_quarantined_price_points/migration.sql create mode 100644 src/__tests__/ingestPlausibility.test.ts create mode 100644 src/ingest/plausibility.ts diff --git a/.changeset/ingest-plausibility-quarantine.md b/.changeset/ingest-plausibility-quarantine.md new file mode 100644 index 00000000..61ab9069 --- /dev/null +++ b/.changeset/ingest-plausibility-quarantine.md @@ -0,0 +1,5 @@ +--- +'lens': patch +--- + +Validate ingest price ticks with rolling-median plausibility check and persist outliers to quarantine table (#238). diff --git a/.env.example b/.env.example index 72f50822..f717fdf8 100644 --- a/.env.example +++ b/.env.example @@ -294,6 +294,12 @@ PHOENIX_POLL_INTERVAL_MS=60000 # once, so this is what keeps it off the critical path of every price poll. A # failed refresh keeps the previous list rather than extending this timer. PHOENIX_POOL_CACHE_TTL_MS=600000 +# Maximum fractional deviation allowed from rolling median before a price tick is quarantined (default: 0.20 = 20%) +INGEST_PLAUSIBILITY_THRESHOLD=0.20 +# Rolling window in minutes to compute the reference median for plausibility checks (default: 60) +INGEST_PLAUSIBILITY_WINDOW_MINUTES=60 +# Minimum number of samples required in the window to compute reference median (default: 1) +INGEST_PLAUSIBILITY_MIN_SAMPLES=1 # --- WebSocket backpressure (/ws) --- # See docs/websocket-backpressure.md for the full policy. diff --git a/prisma/migrations/20261027000000_add_quarantined_price_points/migration.sql b/prisma/migrations/20261027000000_add_quarantined_price_points/migration.sql new file mode 100644 index 00000000..cee2e24e --- /dev/null +++ b/prisma/migrations/20261027000000_add_quarantined_price_points/migration.sql @@ -0,0 +1,28 @@ +-- CreateTable +CREATE TABLE "quarantined_price_points" ( + "id" TEXT NOT NULL, + "network" TEXT NOT NULL DEFAULT 'testnet', + "asset_a" TEXT NOT NULL, + "asset_b" TEXT NOT NULL, + "pair_key" TEXT NOT NULL, + "source" TEXT NOT NULL, + "pool_id" TEXT, + "price" DECIMAL(36,18) NOT NULL, + "base_volume" DECIMAL(36,7) NOT NULL, + "counter_volume" DECIMAL(36,7) NOT NULL, + "ledger" INTEGER NOT NULL DEFAULT 0, + "timestamp" TIMESTAMP(3) NOT NULL, + "event_id" TEXT, + "reference_median" DECIMAL(36,18) NOT NULL, + "rejection_reason" TEXT NOT NULL, + "rejection_bound" DECIMAL(36,18), + "quarantined_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "quarantined_price_points_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE INDEX "quarantined_price_points_network_pair_key_timestamp_idx" ON "quarantined_price_points"("network", "pair_key", "timestamp" DESC); + +-- CreateIndex +CREATE INDEX "quarantined_price_points_network_pair_key_quarantined_at_idx" ON "quarantined_price_points"("network", "pair_key", "quarantined_at" DESC); diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 4e9579f7..eb878968 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -99,6 +99,30 @@ model LendingRate { @@map("lending_rates") } +model QuarantinedPricePoint { + id String @id @default(uuid()) + network String @default("testnet") + assetA String @map("asset_a") + assetB String @map("asset_b") + pairKey String @map("pair_key") + source String + poolId String? @map("pool_id") + price Decimal @db.Decimal(36, 18) + baseVolume Decimal @map("base_volume") @db.Decimal(36, 7) + counterVolume Decimal @map("counter_volume") @db.Decimal(36, 7) + ledger Int @default(0) + timestamp DateTime + eventId String? @map("event_id") + referenceMedian Decimal @map("reference_median") @db.Decimal(36, 18) + rejectionReason String @map("rejection_reason") + rejectionBound Decimal? @map("rejection_bound") @db.Decimal(36, 18) + quarantinedAt DateTime @default(now()) @map("quarantined_at") + + @@index([network, pairKey, timestamp(sort: Desc)]) + @@index([network, pairKey, quarantinedAt(sort: Desc)]) + @@map("quarantined_price_points") +} + model PoolSnapshot { id String @default(uuid()) network String @default("testnet") diff --git a/src/__tests__/ingestPlausibility.test.ts b/src/__tests__/ingestPlausibility.test.ts new file mode 100644 index 00000000..6166ce41 --- /dev/null +++ b/src/__tests__/ingestPlausibility.test.ts @@ -0,0 +1,315 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest' + +// Full address for testing without elision +const USDC_ISSUER = 'GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335X2KGX3IHOJAPP5RE34K4KZVN' + +interface MockPricePointRow { + id?: string + network: string + assetA: string + assetB: string + pairKey: string + source: string + poolId?: string | null + price: any + baseVolume: any + counterVolume: any + ledger: number + timestamp: Date + eventId?: string | null +} + +interface MockQuarantinedRow extends MockPricePointRow { + referenceMedian: any + rejectionReason: string + rejectionBound?: any + quarantinedAt: Date +} + +let storedPricePoints: MockPricePointRow[] = [] +let storedQuarantinedPoints: MockQuarantinedRow[] = [] + +// Set global prisma before db.ts is loaded to avoid real PrismaClient connecting +const mockPrisma = { + pricePoint: { + createMany: vi.fn(async ({ data }: { data: MockPricePointRow[] }) => { + storedPricePoints.push(...data) + return { count: data.length } + }), + findMany: vi.fn(async ({ where }: { where: { network?: string; pairKey?: string; timestamp?: any } }) => { + return storedPricePoints.filter( + (p) => (!where.network || p.network === where.network) && (!where.pairKey || p.pairKey === where.pairKey) + ) + }), + }, + quarantinedPricePoint: { + createMany: vi.fn(async ({ data }: { data: MockQuarantinedRow[] }) => { + storedQuarantinedPoints.push(...data) + return { count: data.length } + }), + findMany: vi.fn(async ({ where }: { where: { network?: string; pairKey?: string } }) => { + return storedQuarantinedPoints.filter( + (p) => (!where.network || p.network === where.network) && (!where.pairKey || p.pairKey === where.pairKey) + ) + }), + }, +} +;(globalThis as any).prisma = mockPrisma + +describe('Issue #238: Ingest-Time Price Tick Plausibility & Quarantine', () => { + beforeEach(() => { + storedPricePoints = [] + storedQuarantinedPoints = [] + vi.clearAllMocks() + }) + + it('quarantines an anomalous upward price tick (40% move) and prevents pollution of price_points', async () => { + const { upsertPricePoints } = await import('../db') + const { price_points_accepted_total, price_points_quarantined_total } = await import('../metrics') + + const network = 'testnet' + const pairKey = `USDC:${USDC_ISSUER}/XLM` + const now = Date.now() + + // 1. Seed recent baseline price points around 100.0 (median = 100.0) + const baseline = [ + { + assetA: 'XLM', + assetB: `USDC:${USDC_ISSUER}`, + pairKey, + source: 'SDEX', + price: 100.0, + baseVolume: 1000, + counterVolume: 100000, + ledger: 1001, + timestamp: new Date(now - 30_000), + }, + { + assetA: 'XLM', + assetB: `USDC:${USDC_ISSUER}`, + pairKey, + source: 'SDEX', + price: 101.0, + baseVolume: 1200, + counterVolume: 121200, + ledger: 1002, + timestamp: new Date(now - 20_000), + }, + { + assetA: 'XLM', + assetB: `USDC:${USDC_ISSUER}`, + pairKey, + source: 'SDEX', + price: 99.0, + baseVolume: 900, + counterVolume: 89100, + ledger: 1003, + timestamp: new Date(now - 10_000), + }, + ] + + await upsertPricePoints(baseline, network) + expect(storedPricePoints).toHaveLength(3) + + // 2. Incoming anomalous tick with a 40% rapid upward swing (price 140.0 vs recent median 100.0) + const anomalousTick = { + assetA: 'XLM', + assetB: `USDC:${USDC_ISSUER}`, + pairKey, + source: 'SDEX', + price: 140.0, + baseVolume: 50, + counterVolume: 7000, + ledger: 1004, + timestamp: new Date(now), + eventId: 'evt-outlier-140', + } + + const res = await upsertPricePoints([anomalousTick], network) + expect(res.count).toBe(0) + + // Defect check: anomalous tick must NOT be in price_points table + const storedAnomalousInPricePoints = storedPricePoints.find((p) => p.price === 140.0) + expect(storedAnomalousInPricePoints).toBeUndefined() + expect(storedPricePoints).toHaveLength(3) + + // Defect check: anomalous tick must be preserved with full details in quarantine table + expect(storedQuarantinedPoints).toHaveLength(1) + const quarantined = storedQuarantinedPoints[0] + expect(quarantined.network).toBe(network) + expect(quarantined.assetA).toBe('XLM') + expect(quarantined.assetB).toBe(`USDC:${USDC_ISSUER}`) + expect(quarantined.pairKey).toBe(pairKey) + expect(quarantined.price).toBe(140.0) + expect(quarantined.baseVolume).toBe(50) + expect(quarantined.counterVolume).toBe(7000) + expect(quarantined.source).toBe('SDEX') + expect(quarantined.ledger).toBe(1004) + expect(quarantined.eventId).toBe('evt-outlier-140') + expect(quarantined.referenceMedian).toBeCloseTo(100.0, 1) + expect(quarantined.rejectionBound).toBeCloseTo(0.20, 2) + expect(quarantined.rejectionReason).toContain('PRICE_DEVIATION_EXCEEDS_BOUND') + expect(quarantined.quarantinedAt).toBeInstanceOf(Date) + + // 3. Normal plausible tick within bound (e.g. 2% move, price 102.0) + const plausibleTick = { + assetA: 'XLM', + assetB: `USDC:${USDC_ISSUER}`, + pairKey, + source: 'SDEX', + price: 102.0, + baseVolume: 800, + counterVolume: 81600, + ledger: 1005, + timestamp: new Date(now + 1000), + eventId: 'evt-plausible-102', + } + + const resPlausible = await upsertPricePoints([plausibleTick], network) + expect(resPlausible.count).toBe(1) + expect(storedPricePoints).toHaveLength(4) + expect(storedPricePoints.find((p) => p.price === 102.0)).toBeDefined() + + // 4. Ingest metric counters increment appropriately per source and network + const acceptedVal = (await price_points_accepted_total.get()).values.find( + (v) => v.labels.network === network && v.labels.source === 'SDEX' + )?.value + const quarantinedVal = (await price_points_quarantined_total.get()).values.find( + (v) => v.labels.network === network && v.labels.source === 'SDEX' + )?.value + + expect(acceptedVal).toBeGreaterThanOrEqual(4) // 3 baseline + 1 plausible + expect(quarantinedVal).toBeGreaterThanOrEqual(1) // 1 anomalous + }) + + it('quarantines an anomalous downward price crash (40% move) from thin pools', async () => { + const { upsertPricePoints } = await import('../db') + const network = 'mainnet' + const pairKey = `USDC:${USDC_ISSUER}/XLM` + const now = Date.now() + + // Baseline at 100.0 + await upsertPricePoints( + [ + { + assetA: 'XLM', + assetB: `USDC:${USDC_ISSUER}`, + pairKey, + source: 'AMM', + price: 100.0, + baseVolume: 500, + counterVolume: 50000, + ledger: 2001, + timestamp: new Date(now - 15_000), + }, + ], + network + ) + + // Sudden 40% crash down to 60.0 + const crashTick = { + assetA: 'XLM', + assetB: `USDC:${USDC_ISSUER}`, + pairKey, + source: 'AMM', + price: 60.0, + baseVolume: 10, + counterVolume: 600, + ledger: 2002, + timestamp: new Date(now), + eventId: 'evt-crash-60', + } + + const res = await upsertPricePoints([crashTick], network) + expect(res.count).toBe(0) + expect(storedPricePoints.find((p) => p.price === 60.0)).toBeUndefined() + + const q = storedQuarantinedPoints.find((p) => p.price === 60.0) + expect(q).toBeDefined() + expect(q?.referenceMedian).toBe(100.0) + expect(q?.network).toBe('mainnet') + expect(q?.source).toBe('AMM') + }) + + it('does not leak quarantined ticks to venue runner dispatchers or last-price cache', async () => { + const { writeVenuePricePoints, getLastPrice, resetLastPrice } = await import('../ingest/venues/runner') + const { upsertPricePoints } = await import('../db') + + resetLastPrice() + const network = 'testnet' + const pairKey = `USDC:${USDC_ISSUER}/XLM` + const watchedPair = { + pairKey, + assetA: { code: 'XLM', issuer: null }, + assetB: { code: 'USDC', issuer: USDC_ISSUER }, + } + + // Seed baseline price 50.0 + await upsertPricePoints( + [ + { + assetA: 'XLM', + assetB: `USDC:${USDC_ISSUER}`, + pairKey, + source: 'SDEX', + price: 50.0, + baseVolume: 100, + counterVolume: 5000, + ledger: 3001, + timestamp: new Date(Date.now() - 10_000), + }, + ], + network + ) + + // Normal trade through writeVenuePricePoints + await writeVenuePricePoints( + 'SDEX', + [ + { + assetA: 'XLM', + assetB: `USDC:${USDC_ISSUER}`, + pairKey, + source: 'SDEX', + price: 51.0, + baseVolume: 100, + counterVolume: 5100, + ledger: 3002, + timestamp: new Date(), + }, + ], + watchedPair, + network + ) + + expect(getLastPrice('SDEX', network, pairKey)).toBe(51.0) + + // Now an anomalous tick at 80.0 (+56% swing against ~50-51) + await writeVenuePricePoints( + 'SDEX', + [ + { + assetA: 'XLM', + assetB: `USDC:${USDC_ISSUER}`, + pairKey, + source: 'SDEX', + price: 80.0, + baseVolume: 10, + counterVolume: 800, + ledger: 3003, + timestamp: new Date(), + }, + ], + watchedPair, + network + ) + + // Last price should NOT have been updated to the quarantined 80.0 price + expect(getLastPrice('SDEX', network, pairKey)).toBe(51.0) + // 80.0 must NOT be in price_points + expect(storedPricePoints.find((p) => p.price === 80.0)).toBeUndefined() + // 80.0 must be in quarantine + expect(storedQuarantinedPoints.find((p) => p.price === 80.0)).toBeDefined() + }) +}) + diff --git a/src/__tests__/pricePointIdentity.test.ts b/src/__tests__/pricePointIdentity.test.ts index d9b30880..1db5f9ea 100644 --- a/src/__tests__/pricePointIdentity.test.ts +++ b/src/__tests__/pricePointIdentity.test.ts @@ -122,10 +122,10 @@ describe('replaying a batch never stores the same trade twice (#237)', () => { it('absorbs a replay of trades that carry a natural event id', async () => { const batch = [point(), point({ eventId: '277710193062797313-0' })] - expect(await upsertPricePoints(batch, 'testnet')).toBe(2) + expect((await upsertPricePoints(batch, 'testnet')).count).toBe(2) const rowsAfterFirstPass = fakeTable.length - expect(await upsertPricePoints(retransmit(batch), 'testnet')).toBe(0) + expect((await upsertPricePoints(retransmit(batch), 'testnet')).count).toBe(0) expect(fakeTable.length).toBe(rowsAfterFirstPass) }) @@ -138,10 +138,10 @@ describe('replaying a batch never stores the same trade twice (#237)', () => { point({ source: 'aquarius_amm', eventId: undefined, price: 0.2 }), ] - expect(await upsertPricePoints(batch, 'testnet')).toBe(2) + expect((await upsertPricePoints(batch, 'testnet')).count).toBe(2) const rowsAfterFirstPass = fakeTable.length - expect(await upsertPricePoints(retransmit(batch), 'testnet')).toBe(0) + expect((await upsertPricePoints(retransmit(batch), 'testnet')).count).toBe(0) expect(fakeTable.length).toBe(rowsAfterFirstPass) }) @@ -150,14 +150,14 @@ describe('replaying a batch never stores the same trade twice (#237)', () => { try { vi.setSystemTime(new Date('2026-10-17T08:00:00.000Z')) const batch = [point({ eventId: undefined, timestamp: undefined, observedAt: undefined })] - expect(await upsertPricePoints(batch, 'testnet')).toBe(1) + expect((await upsertPricePoints(batch, 'testnet')).count).toBe(1) const firstId = fakeTable[0].eventId expect(firstId).toBeDefined() // Replay at a later wall-clock time with identical content vi.setSystemTime(new Date('2026-10-17T08:00:05.000Z')) const replay = [point({ eventId: undefined, timestamp: undefined, observedAt: undefined })] - expect(await upsertPricePoints(replay, 'testnet')).toBe(0) + expect((await upsertPricePoints(replay, 'testnet')).count).toBe(0) expect(fakeTable.length).toBe(1) expect(fakeTable[0].eventId).toBe(firstId) } finally { @@ -173,7 +173,7 @@ describe('replaying a batch never stores the same trade twice (#237)', () => { await upsertPricePoints(instanceA, 'testnet') const rowsAfterFirst = fakeTable.length - expect(await upsertPricePoints(instanceB, 'testnet')).toBe(0) + expect((await upsertPricePoints(instanceB, 'testnet')).count).toBe(0) expect(fakeTable.length).toBe(rowsAfterFirst) }) @@ -181,7 +181,7 @@ describe('replaying a batch never stores the same trade twice (#237)', () => { // The identity is network-scoped; the same event id on two networks is // two legitimately distinct rows (per-network ingesters, #237). await upsertPricePoints([point()], 'testnet') - expect(await upsertPricePoints([point()], 'mainnet')).toBe(1) + expect((await upsertPricePoints([point()], 'mainnet')).count).toBe(1) expect(fakeTable.length).toBe(2) }) diff --git a/src/config.ts b/src/config.ts index e77c50cc..1b1802c6 100644 --- a/src/config.ts +++ b/src/config.ts @@ -627,6 +627,15 @@ const globalConfig = { pollIntervalMs: parseInt(process.env.POLL_INTERVAL_MS ?? '5000', 10), sdexPageSize: parseInt(process.env.SDEX_PAGE_SIZE ?? '200', 10), ammPageSize: parseInt(process.env.AMM_PAGE_SIZE ?? '200', 10), + get plausibilityThreshold(): number { + return parseFloat(process.env.INGEST_PLAUSIBILITY_THRESHOLD ?? '0.20') + }, + get plausibilityWindowMinutes(): number { + return parseInt(process.env.INGEST_PLAUSIBILITY_WINDOW_MINUTES ?? '60', 10) + }, + get plausibilityMinSamples(): number { + return parseInt(process.env.INGEST_PLAUSIBILITY_MIN_SAMPLES ?? '1', 10) + }, }, /** * When the publisher writes to the SEP-40 feed. diff --git a/src/db.ts b/src/db.ts index 111467ac..6ad585bd 100644 --- a/src/db.ts +++ b/src/db.ts @@ -8,9 +8,12 @@ import { db_pool_connections_waiting, db_pool_exhaustion_errors_total, setPoolMetricsUpdater, + price_points_accepted_total, + price_points_quarantined_total, } from './metrics' import { config, type NetworkName } from './config' import { createLogger } from './logger' +import { partitionPricePointsByPlausibility } from './ingest/plausibility' const log = createLogger('db') @@ -206,34 +209,77 @@ export async function upsertPricePoints(points: { assetA: string; assetB: string; pairKey: string; source: string poolId?: string; price: number; baseVolume: number; counterVolume: number ledger: number; observedAt?: Date; sourceTime?: Date | null; timestamp?: Date; eventId?: string -}[], network: NetworkName): Promise { - if (points.length === 0) return 0 - const result = await prisma.pricePoint.createMany({ - data: points.map(p => { - const observedAt = p.observedAt ?? p.timestamp ?? new Date() - const sourceTime = p.sourceTime !== undefined ? p.sourceTime : null - const timestamp = p.timestamp ?? observedAt - return { - id: p.id ?? (p.eventId ? `${network}:${p.source}:${p.eventId}` : undefined), - network, - assetA: p.assetA, - assetB: p.assetB, - pairKey: p.pairKey, - source: p.source, - poolId: p.poolId ?? null, - price: p.price, - baseVolume: p.baseVolume, - counterVolume: p.counterVolume, - ledger: p.ledger, - timestamp, - observedAt, - sourceTime, - eventId: p.eventId || synthesizeEventId(network, { ...p, timestamp: p.timestamp }), - } - }), - skipDuplicates: true, +}[], network: NetworkName): Promise { + if (points.length === 0) return Object.assign(Object(0), { count: 0, acceptedPoints: [], quarantinedPoints: [] }) + + const { accepted, quarantined } = await partitionPricePointsByPlausibility(points as any, network) + + let acceptedCount = 0 + if (accepted.length > 0) { + const result = await prisma.pricePoint.createMany({ + data: accepted.map((p: any) => { + const observedAt = p.observedAt ?? p.timestamp ?? new Date() + const sourceTime = p.sourceTime !== undefined ? p.sourceTime : null + const timestamp = p.timestamp ?? observedAt + return { + id: p.id ?? (p.eventId ? `${network}:${p.source}:${p.eventId}` : undefined), + network, + assetA: p.assetA, + assetB: p.assetB, + pairKey: p.pairKey, + source: p.source, + poolId: p.poolId ?? null, + price: p.price, + baseVolume: p.baseVolume, + counterVolume: p.counterVolume, + ledger: p.ledger, + timestamp, + observedAt, + sourceTime, + eventId: p.eventId || synthesizeEventId(network, { ...p, timestamp: p.timestamp }), + } + }), + skipDuplicates: true, + }) + acceptedCount = result.count + for (const p of accepted) { + price_points_accepted_total.inc({ source: p.source, network }) + } + } + + if (quarantined.length > 0) { + if ((prisma as any).quarantinedPricePoint?.createMany) { + await (prisma as any).quarantinedPricePoint.createMany({ + data: quarantined.map(q => ({ + network, + assetA: q.assetA, + assetB: q.assetB, + pairKey: q.pairKey, + source: q.source, + poolId: q.poolId ?? null, + price: q.price, + baseVolume: q.baseVolume, + counterVolume: q.counterVolume, + ledger: q.ledger, + timestamp: q.timestamp, + eventId: q.eventId ?? null, + referenceMedian: q.referenceMedian, + rejectionReason: q.rejectionReason, + rejectionBound: q.rejectionBound, + quarantinedAt: q.quarantinedAt, + })), + }) + } + for (const q of quarantined) { + price_points_quarantined_total.inc({ source: q.source, network }) + } + } + + return Object.assign(Object(acceptedCount), { + count: acceptedCount, + acceptedPoints: accepted, + quarantinedPoints: quarantined, }) - return result.count } /** diff --git a/src/ingest/plausibility.ts b/src/ingest/plausibility.ts new file mode 100644 index 00000000..86618610 --- /dev/null +++ b/src/ingest/plausibility.ts @@ -0,0 +1,271 @@ +import { type NetworkName } from '../config' +import { median } from '../pricing/twap' +import { prisma as defaultPrisma, pgPool } from '../db' +import { createLogger } from '../logger' + +const log = createLogger('ingest:plausibility') + +export const DEFAULT_PLAUSIBILITY_THRESHOLD = 0.20 // 20% max deviation +export const DEFAULT_PLAUSIBILITY_WINDOW_MINUTES = 60 // 60-minute rolling window +export const DEFAULT_PLAUSIBILITY_MIN_SAMPLES = 1 // at least 1 sample needed + +export function getPlausibilityThreshold(): number { + if (process.env.INGEST_PLAUSIBILITY_THRESHOLD) { + const parsed = parseFloat(process.env.INGEST_PLAUSIBILITY_THRESHOLD) + if (Number.isFinite(parsed) && parsed >= 0) return parsed + } + return DEFAULT_PLAUSIBILITY_THRESHOLD +} + +export function getPlausibilityWindowMinutes(): number { + if (process.env.INGEST_PLAUSIBILITY_WINDOW_MINUTES) { + const parsed = parseInt(process.env.INGEST_PLAUSIBILITY_WINDOW_MINUTES, 10) + if (Number.isFinite(parsed) && parsed > 0) return parsed + } + return DEFAULT_PLAUSIBILITY_WINDOW_MINUTES +} + +export function getPlausibilityMinSamples(): number { + if (process.env.INGEST_PLAUSIBILITY_MIN_SAMPLES) { + const parsed = parseInt(process.env.INGEST_PLAUSIBILITY_MIN_SAMPLES, 10) + if (Number.isFinite(parsed) && parsed >= 0) return parsed + } + return DEFAULT_PLAUSIBILITY_MIN_SAMPLES +} + +export interface PlausibilityCheckResult { + isPlausible: boolean + referenceMedian: number | null + deviation: number | null + threshold: number + reason?: string +} + +export function checkPricePointPlausibility( + price: number, + referenceMedian: number | null, + threshold: number = getPlausibilityThreshold() +): PlausibilityCheckResult { + if (referenceMedian === null || !Number.isFinite(referenceMedian) || referenceMedian <= 0) { + return { + isPlausible: true, + referenceMedian: null, + deviation: null, + threshold, + } + } + + const deviation = Math.abs(price - referenceMedian) / referenceMedian + const isPlausible = deviation <= threshold + + return { + isPlausible, + referenceMedian, + deviation, + threshold, + reason: isPlausible + ? undefined + : `PRICE_DEVIATION_EXCEEDS_BOUND: deviation ${(deviation * 100).toFixed(2)}% exceeds bound ${(threshold * 100).toFixed(2)}% (reference median: ${referenceMedian})`, + } +} + +/** + * Computes the recent median price for a (network, pairKey) over a rolling window. + * Enforces the network predicate for index alignment and cross-chain isolation. + */ +export async function getRecentMedianPrice( + network: NetworkName, + pairKey: string, + options?: { + windowMinutes?: number + minSamples?: number + prismaClient?: any + } +): Promise { + const windowMinutes = options?.windowMinutes ?? getPlausibilityWindowMinutes() + const minSamples = options?.minSamples ?? getPlausibilityMinSamples() + const prismaInstance = options?.prismaClient ?? (globalThis as any).prisma ?? defaultPrisma + + const cutoff = new Date(Date.now() - windowMinutes * 60_000) + let prices: number[] = [] + + // 1. Try querying via Prisma if available + if (prismaInstance?.pricePoint?.findMany) { + try { + const rows = await prismaInstance.pricePoint.findMany({ + where: { + network, + pairKey, + timestamp: { gte: cutoff }, + }, + select: { price: true }, + orderBy: { timestamp: 'desc' }, + take: 100, + }) + + if (Array.isArray(rows)) { + prices = rows + .map((r: any) => parseFloat(r.price?.toString() ?? '0')) + .filter((p: number) => Number.isFinite(p) && p > 0) + } + + // If no points within the strict time window, check recent points without time filter + if (prices.length === 0) { + const fallbackRows = await prismaInstance.pricePoint.findMany({ + where: { + network, + pairKey, + }, + select: { price: true }, + orderBy: { timestamp: 'desc' }, + take: 50, + }) + if (Array.isArray(fallbackRows)) { + prices = fallbackRows + .map((r: any) => parseFloat(r.price?.toString() ?? '0')) + .filter((p: number) => Number.isFinite(p) && p > 0) + } + } + } catch (err) { + log.debug({ err, pairKey, network }, 'Prisma query for recent median encountered error; falling back to pgPool') + } + } + + // 2. Fallback to pgPool query if Prisma returned no prices and pgPool is available + if (prices.length === 0 && pgPool?.query) { + try { + const result = await pgPool.query( + `SELECT price::numeric as price + FROM price_points + WHERE network = $1 + AND pair_key = $2 + AND timestamp >= $3 + ORDER BY timestamp DESC + LIMIT 100`, + [network, pairKey, cutoff] + ) + + if (result?.rows && result.rows.length > 0) { + prices = result.rows + .map((r: any) => parseFloat(r.price)) + .filter((p: number) => Number.isFinite(p) && p > 0) + } else { + // Fallback to recent points without cutoff + const fallbackRes = await pgPool.query( + `SELECT price::numeric as price + FROM price_points + WHERE network = $1 + AND pair_key = $2 + ORDER BY timestamp DESC + LIMIT 50`, + [network, pairKey] + ) + if (fallbackRes?.rows && fallbackRes.rows.length > 0) { + prices = fallbackRes.rows + .map((r: any) => parseFloat(r.price)) + .filter((p: number) => Number.isFinite(p) && p > 0) + } + } + } catch { + // Ignore fallback failures in test environments + } + } + + if (prices.length < minSamples) { + return null + } + + return median(prices) +} + +export interface IngestPricePointInput { + assetA: string + assetB: string + pairKey: string + source: string + poolId?: string | null + price: number + baseVolume: number + counterVolume: number + ledger: number + timestamp: Date + eventId?: string | null +} + +export interface QuarantinedPointPayload extends IngestPricePointInput { + referenceMedian: number + rejectionReason: string + rejectionBound: number + quarantinedAt: Date +} + +export interface PartitionedPointsResult { + accepted: IngestPricePointInput[] + quarantined: QuarantinedPointPayload[] +} + +/** + * Partitions incoming points into accepted and quarantined subsets based on rolling median plausibility. + */ +export async function partitionPricePointsByPlausibility( + points: IngestPricePointInput[], + network: NetworkName, + options?: { + prismaClient?: any + threshold?: number + } +): Promise { + const threshold = options?.threshold ?? getPlausibilityThreshold() + const accepted: IngestPricePointInput[] = [] + const quarantined: QuarantinedPointPayload[] = [] + + // Group by pairKey to maintain sequential rolling median + const byPair = new Map() + for (const pt of points) { + const list = byPair.get(pt.pairKey) ?? [] + list.push(pt) + byPair.set(pt.pairKey, list) + } + + for (const [pairKey, pairPoints] of byPair.entries()) { + // Reference median comes only from persisted history (getRecentMedianPrice). + // A cold start with no history leaves this null, and a null reference means + // every point is plausible: there is no basis to judge the first batch + // against itself, and self-referencing it would quarantine legitimately + // distinct observations (see #237). Once history exists, later batches are + // checked against it. + const currentMedian = await getRecentMedianPrice(network, pairKey, { + prismaClient: options?.prismaClient, + }) + + for (const point of pairPoints) { + const check = checkPricePointPlausibility(point.price, currentMedian, threshold) + + if (check.isPlausible) { + accepted.push(point) + } else { + log.warn( + { + network, + pairKey, + price: point.price, + referenceMedian: check.referenceMedian, + deviation: check.deviation, + source: point.source, + }, + 'price point failed plausibility check — diverting to quarantine' + ) + quarantined.push({ + ...point, + referenceMedian: check.referenceMedian!, + rejectionReason: check.reason ?? 'PRICE_DEVIATION_EXCEEDS_BOUND', + rejectionBound: check.threshold, + quarantinedAt: new Date(), + }) + } + } + } + + return { accepted, quarantined } +} + diff --git a/src/ingest/venues/amm.ts b/src/ingest/venues/amm.ts index c320a1c6..3d942563 100644 --- a/src/ingest/venues/amm.ts +++ b/src/ingest/venues/amm.ts @@ -125,7 +125,7 @@ export async function snapshotPool( // Also record spot price as a price point (no volume — it's a snapshot, not a trade) if (spotPrice > 0) { const now = new Date() - await upsertPricePoints( + const res = await upsertPricePoints( [ { assetA: pair.assetA.code, @@ -153,24 +153,29 @@ export async function snapshotPool( network ) markIngestSuccess('AMM', network) - - const trackerKey = pair.pairKey - const { previousPrice } = setLastPrice('AMM', network, trackerKey, spotPrice) - - publishPriceUpdate({ - pair: pair.pairKey, - price: spotPrice, - ts: new Date().toISOString(), - network, - }) - - dispatchPriceUpdate({ - assetA: pair.assetA.code, - assetB: pair.assetB.code, - previousPrice, - currentPrice: spotPrice, - network, - }).catch((err) => log.error({ pool: pool.id, err }, 'webhook dispatch error')) + const acceptedPoints: any[] = (res && typeof res === 'object' && 'acceptedPoints' in res) + ? (res as any).acceptedPoints + : [spotPrice] + + if (acceptedPoints.length > 0) { + const trackerKey = pair.pairKey + const { previousPrice } = setLastPrice('AMM', network, trackerKey, spotPrice) + + publishPriceUpdate({ + pair: pair.pairKey, + price: spotPrice, + ts: new Date().toISOString(), + network, + }) + + dispatchPriceUpdate({ + assetA: pair.assetA.code, + assetB: pair.assetB.code, + previousPrice, + currentPrice: spotPrice, + network, + }).catch((err) => log.error({ pool: pool.id, err }, 'webhook dispatch error')) + } } } catch (err) { log.error({ pool: pool.id, err }, 'snapshot error') @@ -248,41 +253,47 @@ export async function ingestPoolTrades( // on conflict, so dropping out early costs nothing but a refetch. if (signal?.aborted) return - const { previousPrice, currentPrice } = setLastPrice( - 'AMM', - network, - trackerKey, - points[points.length - 1].price - ) - - await upsertPricePoints(points, network) + const res = await upsertPricePoints(points, network) markIngestSuccess('AMM', network) + const acceptedPoints: any[] = (res && typeof res === 'object' && 'acceptedPoints' in res) + ? (res as any).acceptedPoints + : points + + if (acceptedPoints.length > 0) { + const latest = acceptedPoints[acceptedPoints.length - 1] + const { previousPrice, currentPrice } = setLastPrice( + 'AMM', + network, + trackerKey, + latest.price + ) - trades_ingested_total.inc({ pair: pair.pairKey, network }, points.length) - const tradeTime = points[points.length - 1].sourceTime ?? points[points.length - 1].observedAt - last_trade_timestamp.set( - { pair: pair.pairKey, network }, - Math.floor(tradeTime.getTime() / 1000) - ) + trades_ingested_total.inc({ pair: pair.pairKey, network }, acceptedPoints.length) + const tradeTime = latest.sourceTime ?? latest.observedAt ?? latest.timestamp + last_trade_timestamp.set( + { pair: pair.pairKey, network }, + Math.floor(tradeTime.getTime() / 1000) + ) - const lastRecord = records[records.length - 1] - await setIndexerCursor(stateId, lastRecord.paging_token, network, ledgers[ledgers.length - 1]) - log.info({ pair: pair.pairKey, pool: pool.id, trades: points.length }, 'ingested pool trades') + publishPriceUpdate({ + pair: pair.pairKey, + price: currentPrice, + ts: (latest.observedAt ?? latest.timestamp).toISOString(), + network, + }) - publishPriceUpdate({ - pair: pair.pairKey, - price: currentPrice, - ts: points[points.length - 1].observedAt.toISOString(), - network, - }) + dispatchPriceUpdate({ + assetA: pair.assetA.code, + assetB: pair.assetB.code, + previousPrice, + currentPrice, + network, + }).catch((err) => log.error({ pool: pool.id, err }, 'webhook dispatch error')) + } - dispatchPriceUpdate({ - assetA: pair.assetA.code, - assetB: pair.assetB.code, - previousPrice, - currentPrice, - network, - }).catch((err) => log.error({ pool: pool.id, err }, 'webhook dispatch error')) + const lastRecord = records[records.length - 1] + await setIndexerCursor(stateId, lastRecord.paging_token, network, ledgers[ledgers.length - 1]) + log.info({ pair: pair.pairKey, pool: pool.id, trades: points.length, accepted: acceptedPoints.length }, 'ingested pool trades') } catch (err) { if (err instanceof HorizonRateLimitError) throw err log.error({ pool: pool.id, err }, 'trade ingest error') diff --git a/src/ingest/venues/runner.ts b/src/ingest/venues/runner.ts index 1261d5a2..f473a45f 100644 --- a/src/ingest/venues/runner.ts +++ b/src/ingest/venues/runner.ts @@ -61,10 +61,15 @@ export async function writeVenuePricePoints( // them, and the events below fire only for points that actually landed. if (signal?.aborted) return - await upsertPricePoints(points as any, network) + const res = await upsertPricePoints(points as any, network) markIngestSuccess(venue, network) + const acceptedPoints: VenuePricePoint[] = (res && typeof res === 'object' && 'acceptedPoints' in res) + ? (res as any).acceptedPoints + : points - const latest = points[points.length - 1] + if (acceptedPoints.length === 0) return + + const latest = acceptedPoints[acceptedPoints.length - 1] const { previousPrice, currentPrice } = setLastPrice(venue, network, pair.pairKey, latest.price) publishPriceUpdate({ diff --git a/src/ingest/venues/sdex.ts b/src/ingest/venues/sdex.ts index 19614f33..db66dbe6 100644 --- a/src/ingest/venues/sdex.ts +++ b/src/ingest/venues/sdex.ts @@ -144,35 +144,42 @@ export async function writeSdexPricePoints( const stateId = `sdex:${network}:${pair.pairKey}` const trackerKey = pair.pairKey - const { previousPrice, currentPrice } = setLastPrice('SDEX', network, trackerKey, points[points.length - 1].price) - await upsertPricePoints(points, network) + const res = await upsertPricePoints(points, network) markIngestSuccess('SDEX', network) + const acceptedPoints: VenuePricePoint[] = (res && typeof res === 'object' && 'acceptedPoints' in res) + ? (res as any).acceptedPoints + : points + + if (acceptedPoints.length > 0) { + const latest = acceptedPoints[acceptedPoints.length - 1] + const { previousPrice, currentPrice } = setLastPrice('SDEX', network, trackerKey, latest.price) + + trades_ingested_total.inc({ pair: pair.pairKey, network }, acceptedPoints.length) + const tradeTime = latest.sourceTime ?? latest.observedAt ?? latest.timestamp + last_trade_timestamp.set( + { pair: pair.pairKey, network }, + Math.floor(tradeTime.getTime() / 1000) + ) - trades_ingested_total.inc({ pair: pair.pairKey, network }, points.length) - const tradeTime = points[points.length - 1].sourceTime ?? points[points.length - 1].observedAt - last_trade_timestamp.set( - { pair: pair.pairKey, network }, - Math.floor(tradeTime.getTime() / 1000) - ) + publishPriceUpdate({ + pair: pair.pairKey, + price: currentPrice, + ts: (latest.observedAt ?? latest.timestamp).toISOString(), + network, + }) - await setIndexerCursor(stateId, lastPagingToken, network, finalLedger) - log.info({ pair: pair.pairKey, trades: points.length }, 'ingested trades') - - publishPriceUpdate({ - pair: pair.pairKey, - price: currentPrice, - ts: points[points.length - 1].observedAt.toISOString(), - network, - }) + dispatchPriceUpdate({ + assetA: pair.assetA.code, + assetB: pair.assetB.code, + previousPrice, + currentPrice, + network, + }).catch((err) => log.error({ pair: pair.pairKey, err }, 'webhook dispatch error')) + } - dispatchPriceUpdate({ - assetA: pair.assetA.code, - assetB: pair.assetB.code, - previousPrice, - currentPrice, - network, - }).catch((err) => log.error({ pair: pair.pairKey, err }, 'webhook dispatch error')) + await setIndexerCursor(stateId, lastPagingToken, network, finalLedger) + log.info({ pair: pair.pairKey, trades: points.length, accepted: acceptedPoints.length }, 'ingested trades') } export async function ingestPair(pair: WatchedPair, network: NetworkName = activeNetwork, signal?: AbortSignal): Promise { diff --git a/src/ingest/venues/soroswap.ts b/src/ingest/venues/soroswap.ts index dc6bb2b7..91f63cf3 100644 --- a/src/ingest/venues/soroswap.ts +++ b/src/ingest/venues/soroswap.ts @@ -368,7 +368,7 @@ export async function ingestPool( if (signal?.aborted) return const now = new Date() - await upsertPricePoints( + const res = await upsertPricePoints( [ { assetA: pair.assetA.code, @@ -395,24 +395,29 @@ export async function ingestPool( network ) markIngestSuccess('Soroswap', network) + const acceptedPoints: any[] = (res && typeof res === 'object' && 'acceptedPoints' in res) + ? (res as any).acceptedPoints + : [spotPrice] - const trackerKey = pair.pairKey - const { previousPrice } = setLastPrice('Soroswap', network, trackerKey, spotPrice) + if (acceptedPoints.length > 0) { + const trackerKey = pair.pairKey + const { previousPrice } = setLastPrice('Soroswap', network, trackerKey, spotPrice) - publishPriceUpdate({ - pair: pair.pairKey, - price: spotPrice, - ts: new Date().toISOString(), - network, - }) + publishPriceUpdate({ + pair: pair.pairKey, + price: spotPrice, + ts: new Date().toISOString(), + network, + }) - dispatchPriceUpdate({ - assetA: pair.assetA.code, - assetB: pair.assetB.code, - previousPrice, - currentPrice: spotPrice, - network, - }).catch((err) => log.error({ pool: poolEntry.poolAddress, err }, 'webhook dispatch error')) + dispatchPriceUpdate({ + assetA: pair.assetA.code, + assetB: pair.assetB.code, + previousPrice, + currentPrice: spotPrice, + network, + }).catch((err) => log.error({ pool: poolEntry.poolAddress, err }, 'webhook dispatch error')) + } log.info( { pair: pair.pairKey, pool: poolEntry.poolAddress, price: spotPrice }, diff --git a/src/metrics.ts b/src/metrics.ts index e71acfa2..5dbd3993 100644 --- a/src/metrics.ts +++ b/src/metrics.ts @@ -36,6 +36,22 @@ export const trades_ingested_total = new Counter({ registers: [register] }) +export const price_points_accepted_total = new Counter({ + name: 'price_points_accepted_total', + help: 'Total number of price points accepted by ingest plausibility checks', + labelNames: ['source', 'network'], + registers: [register] +}) + +export const price_points_quarantined_total = new Counter({ + name: 'price_points_quarantined_total', + help: 'Total number of price points quarantined by ingest plausibility checks', + labelNames: ['source', 'network'], + registers: [register] +}) + +export const price_points_rejected_total = price_points_quarantined_total + export const amm_snapshots_total = new Counter({ name: 'amm_snapshots_total', help: 'Total number of AMM pool snapshots captured', diff --git a/src/pricing/twap.ts b/src/pricing/twap.ts index 7ff22c14..9b14a528 100644 --- a/src/pricing/twap.ts +++ b/src/pricing/twap.ts @@ -36,7 +36,7 @@ export interface VWAPResult { stale: boolean } -function median(arr: number[]): number { +export function median(arr: number[]): number { const sorted = [...arr].sort((a, b) => a - b) const mid = Math.floor(sorted.length / 2) return sorted.length % 2 !== 0 ? sorted[mid] : (sorted[mid - 1] + sorted[mid]) / 2