diff --git a/.env.example b/.env.example index 3937aedc..d53e8fba 100644 --- a/.env.example +++ b/.env.example @@ -225,3 +225,20 @@ MCP_PAID_CALL_ALLOWED_HOSTS= # against a table that has never been pruned cannot hold a long lock or blow # the WAL. Default 1000. SNAPSHOT_PRUNE_BATCH_SIZE=1000 + +# ─── Data Quality & Plausibility ─────────────────────────────────────────── +# Maximum age in seconds before price observation is flagged as stale. Default 300. +PRICE_STALE_THRESHOLD_SECONDS=300 + +# Maximum allowed percentage deviation over rolling median before an incoming tick is quarantined. Default 40. +INGEST_OUTLIER_BOUND_PCT=40 + +# Comma-separated list of enabled centralized exchange adapters (binance, coinbase, kraken, mock). Default binance,coinbase,kraken. +ENABLED_CEX_VENUES=binance,coinbase,kraken + +# Days to retain raw price_points rows before rolling them up into hourly aggregates and pruning. Default 30. +PRICE_POINTS_RETENTION_DAYS=30 + +# Maximum raw price_points rows deleted per batch during retention pruning. Default 1000. +PRICE_POINTS_PRUNE_BATCH_SIZE=1000 + diff --git a/README.md b/README.md index 13334c3b..ce152dbe 100644 --- a/README.md +++ b/README.md @@ -9,37 +9,59 @@ Aggregates price data from Stellar's Classic Order Book (SDEX) and AMM Liquidity **What Horizon doesn't do:** Horizon's `/trade_aggregations` only covers SDEX. AMM pool prices (derived from reserve ratios) are not aggregated anywhere. Lens fills this gap. +## Supported Venues + + +| Venue | Type | Description | +|---|---|---| +| **SDEX** | DEX | Stellar's native order book matching engine | +| **Stellar AMM** | AMM | Protocol-level constant-product liquidity pools | +| **Soroswap** | AMM / DEX | Uniswap v2-style automated market maker on Soroban | +| **Aquarius** | AMM / Liquidity | Liquidity protocol and automated market maker | +| **Phoenix** | AMM / DEX | Next-generation Soroban decentralized exchange | +| **Binance** | CEX | Centralized exchange public ticker and order book depth | +| **Coinbase** | CEX | Centralized exchange public ticker and order book depth | +| **Kraken** | CEX | Centralized exchange public ticker and order book depth | + + ## Endpoints ### REST + + | Method | Path | Description | |---|---|---| -| GET | `/price/:assetA/:assetB` | Current VWAP, 24h volume, best route | -| GET | `/price/:assetA/:assetB/route?amount=1000` | Best execution route for a given amount | -| GET | `/price/:assetA/:assetB/depth?amount=1000` | Simulated order-book depth and execution slippage | -| GET | `/book/:assetA/:assetB?depth=0.1&step=0.01&network=testnet` | Aggregated SDEX book and synthetic AMM price ladder with cumulative size and venue provenance | -| GET | `/price/:assetA/:assetB/history?window=1h` | OHLCV history (`1m`, `5m`, `1h`, `24h`) | -| GET | `/price/twap/:assetA/:assetB?window=60&sampleInterval=60&method=iqr` | Manipulation-resistant time-weighted average price | -| GET | `/price/vwap/:assetA/:assetB?window=60&source=SDEX&method=iqr` | Volume-weighted average price | -| GET | `/candles/:assetA/:assetB?interval=1h&from=…&to=…` | OHLCV candles (`1m`, `5m`, `15m`, `1h`, `4h`, `1d`) | -| GET | `/prices/history?pair=XLM/USDC&from=…&to=…&interval=1m` | Historical 1-minute price snapshots, optionally aggregated (`1m`, `5m`, `1h`); honours `?network=`; `?format=csv` serves the points as `text/csv` (`ts,price,volume`); ~30-day retention | -| GET | `/pools` | Active AMM pools being watched | -| GET | `/pairs` | Watched trading pairs | -| GET | `/volumes/:asset?window=24h` | Cross-venue traded volume (`24h`, `7d`, `30d`) with a per-venue breakdown | -| GET | `/spreads/:asset?window=5m` | Per-venue bid/ask spread, tightest first (`5m`, `15m`, `1h`, `24h`) | -| GET | `/compare/:asset` | Lens vs configured oracle prices, with per-oracle deviation and availability | -| GET | `/screener?sortBy=volume&order=desc&limit=20` | Screen pairs by volume, 24h change, liquidity and price | -| GET | `/benchmark/:asset?target=USD` | Peg-deviation statistics versus a target | -| GET | `/basket?asset=XLM&asset=USDC&weight=0.6&weight=0.4` | Weighted price of a basket of assets | -| GET | `/status` | Indexer health | -| GET | `/discovery/resources?type=&payTo=&network=&extensions=&limit=&offset=` | Bazaar catalog of x402-discoverable resources (spec: [`bazaar`](https://github.com/x402-foundation/x402/blob/main/specs/extensions/bazaar.md)) | -| GET | `/usage/me` | Usage and quota for the calling API key (requires `Authorization: Bearer `) | -| GET | `/supported` | Payment kinds and extensions this facilitator supports | -| POST | `/verify` | Verify an x402 payment payload | -| POST | `/settle` | Settle an x402 payment (idempotent — retries replay the stored result) | +| GET | `/venues/registry` | List registered venues and their network liveness | +| GET | `/status` | Get API status and indexer state | +| GET | `/price/:assetA/:assetB` | Get aggregated price for a pair | +| GET | `/price/:assetA/:assetB/route` | Get best swap route for a pair and amount | +| GET | `/price/:assetA/:assetB/history` | Get price history buckets | +| GET | `/pools` | Get AMM liquidity pool reserves and spot prices | +| GET | `/candles/:assetA/:assetB` | Get OHLCV candle data | +| GET | `/prices/history` | Get historical 1-minute price snapshots | +| GET/POST | `/pairs` | List all active trading pairs | +| GET | `/price/:assetA/:assetB/depth` | Get order-book depth and execution slippage for a pair | +| GET | `/book/:assetA/:assetB` | Get the aggregated SDEX and AMM price ladder for a pair | +| GET | `/price/twap/:assetA/:assetB` | Get a manipulation-resistant time-weighted average price | +| GET | `/price/vwap/:assetA/:assetB` | Get a volume-weighted average price | +| GET | `/volumes/:asset` | Get cross-venue traded volume for an asset | +| GET | `/spreads/:asset` | Get the tightest per-venue bid/ask spread for an asset | +| GET | `/compare/:asset` | Compare Lens prices with every configured external oracle | +| GET | `/premium/:assetA/:assetB` | Get CEX vs on-chain price premium and spread in basis points | +| GET | `/venues` | Get cross-venue traded volume, market share percentages and period change | +| GET | `/tvl` | Get venue-level Total Value Locked with unpriceable pool reporting | +| GET | `/screener` | Screen pairs by volume, change, liquidity and price | +| GET | `/benchmark/:asset` | Get peg-deviation statistics for an asset against a target | +| GET | `/basket` | Compute the weighted price of a basket of assets | +| GET | `/discovery/resources` | Browse the x402 Bazaar catalog of discoverable resources | | POST | `/webhooks` | Subscribe to a price-threshold webhook | | DELETE | `/webhooks/:id` | Delete a webhook subscription | -| GET/POST | `/graphql` | GraphQL query/mutation endpoint (GraphiQL IDE at `/graphiql`, docs: [`docs/graphql.md`](docs/graphql.md)) | +| GET | `/usage/me` | Get usage and quota for the calling API key | +| GET | `/supported` | List the payment kinds and extensions this facilitator supports | +| POST | `/verify` | Verify an x402 payment payload | +| POST | `/settle` | Settle an x402 payment (idempotent) | +| GET/POST | `/graphql` | Execute a GraphQL query over HTTP GET | + 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/docs/data-quality.md b/docs/data-quality.md new file mode 100644 index 00000000..ce8d08be --- /dev/null +++ b/docs/data-quality.md @@ -0,0 +1,42 @@ +# Data Quality & Service Level Guarantees + +This document defines the precision, freshness boundaries, completeness SLA, and outlier filtering guarantees provided by Lens. + +--- + +## 1. Freshness Guarantees +* **Standard Staleness Threshold:** `300 seconds` (5 minutes) + - Enforced in: [`src/aggregator/vwap.ts`](file:///d:/PersonalAI/lens/src/aggregator/vwap.ts) via `PRICE_STALE_THRESHOLD_SECONDS`. + - Behavior when exceeded: Responses explicitly flag `"stale": true` and confidence drops to `"low"`. +* **Indexer Lag Alarm:** `300 seconds` + - Enforced in: [`src/api/rest.ts`](file:///d:/PersonalAI/lens/src/api/rest.ts) `/status` health probe. + - Behavior when exceeded: `/status` reports `"status": "degraded"` with `"health": { "data": "stale" }`. + +--- + +## 2. Ingest-Time Outlier & Plausibility Check +* **Deviation Bound:** `40%` + - Enforced in: [`src/db.ts`](file:///d:/PersonalAI/lens/src/db.ts) via `INGEST_OUTLIER_BOUND_PCT`. + - Behavior when exceeded: Aberrant ticks deviating > 40% from the recent rolling median are rejected from the primary `price_points` table and stored in the quarantine audit store, incrementing `price_points_ingested_total{status="rejected"}`. + +--- + +## 3. Precision & Decimal Handling +* **Prices:** Exact 18 decimal places (`Decimal(36, 18)`). +* **Volumes:** Exact 7 decimal places (`Decimal(36, 7)`). +* Enforced in: [`prisma/schema.prisma`](file:///d:/PersonalAI/lens/prisma/schema.prisma). +* Lossless String Formatting: Financial clients can pass header `X-Format-Decimals: string` to receive exact string representations without double-precision truncation. + +--- + +## 4. Retention & Pruning +* **Snapshot Retention Window:** `30 days` (`SNAPSHOT_RETENTION_DAYS` in [`src/jobs/snapshotRetention.ts`](file:///d:/PersonalAI/lens/src/jobs/snapshotRetention.ts)). +* **Price Points Retention Window:** `30 days` (`PRICE_POINTS_RETENTION_DAYS` in [`src/jobs/snapshotRetention.ts`](file:///d:/PersonalAI/lens/src/jobs/snapshotRetention.ts)). +* Rollup Requirement: Raw price points older than 30 days are rolled up into 1-hour `price_aggregates` buckets before batch deletion. + +--- + +## 5. Rate Limiting Bounds +* **Public Unauthenticated IP Limit:** `100 requests / minute` (`RATE_LIMITS.ipDefault` in [`src/app.ts`](file:///d:/PersonalAI/lens/src/app.ts)). +* **Authenticated API Key Limit:** `1000 requests / minute` (`RATE_LIMITS.authenticatedDefault` in [`src/app.ts`](file:///d:/PersonalAI/lens/src/app.ts)). +* **Monitoring Endpoint (`/status`) Limit:** `1000 requests / minute` (`RATE_LIMITS.monitoring` in [`src/app.ts`](file:///d:/PersonalAI/lens/src/app.ts)). diff --git a/docs/deploy.md b/docs/deploy.md new file mode 100644 index 00000000..a5381c0a --- /dev/null +++ b/docs/deploy.md @@ -0,0 +1,33 @@ +# Deployment and Database Migration Guide + +This document describes the deployment procedure for the Lens database and application services. + +--- + +## Database Migrations: `prisma migrate deploy` + +Lens uses **Prisma Migrate** for all schema updates. + +### Production Migration Command +```bash +npx prisma migrate deploy +``` + +### Why `prisma db push` is Strictly Prohibited in Production +1. **Unpredictable Schema Diffs:** `db push` inspects the live database connection and immediately alters tables without reviewable SQL statements. +2. **Data Loss Risk:** `db push` can silently drop columns, indexes, or composite partitions when models are modified or renamed. +3. **No Migration History:** `db push` leaves no audit trail or rollback record in `_prisma_migrations`, making rollbacks impossible. +4. **Reproducibility:** Every deployment stage (local, test, staging, production) must execute the exact byte-for-byte migration sequence recorded under `prisma/migrations/`. + +--- + +## Zero-Downtime Rollout Steps +1. **Pre-Deployment:** + - Execute tests: `npm test` + - Verify migration drift: `npx prisma migrate diff --from-schema-datamodel prisma/schema.prisma --to-migrations prisma/migrations --exit-code` +2. **Database Migration:** + - Run `npx prisma migrate deploy` +3. **Application Deploy:** + - Start or restart worker and API containers using `npm start` +4. **Post-Deployment Verification:** + - Verify health probe: `GET /status` returns HTTP 200 `{ "ok": true, "status": "healthy" }` diff --git a/openapi.json b/openapi.json index 94cb0a74..a3cf38f5 100644 --- a/openapi.json +++ b/openapi.json @@ -1358,9 +1358,224 @@ } } } + } + } + } + }, + "/premium/{assetA}/{assetB}": { + "get": { + "operationId": "getPremium", + "summary": "Get CEX vs on-chain price premium and spread in basis points", + "security": [], + "parameters": [ + { + "name": "assetA", + "in": "path", + "required": true, + "schema": { + "type": "string" + }, + "example": "XLM" + }, + { + "name": "assetB", + "in": "path", + "required": true, + "schema": { + "type": "string" + }, + "example": "USDC" + }, + { + "$ref": "#/components/parameters/NetworkQuery" + }, + { + "name": "stalenessBound", + "in": "query", + "required": false, + "schema": { + "type": "integer" + }, + "description": "Max allowed age in seconds before price is rejected as stale" + } + ], + "responses": { + "200": { + "description": "Premium and spread in basis points", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "pair": { + "type": "string" + }, + "network": { + "type": "string" + }, + "onChainPrice": { + "type": "number" + }, + "onChainSource": { + "type": "string" + }, + "onChainTimestamp": { + "type": "string", + "format": "date-time" + }, + "cexPrice": { + "type": "number" + }, + "cexSource": { + "type": "string" + }, + "cexTimestamp": { + "type": "string", + "format": "date-time" + }, + "spreadBps": { + "type": "integer" + }, + "stalenessBoundSeconds": { + "type": "integer" + } + } + } + } + } }, "404": { - "description": "No price data found for the asset" + "description": "Missing price data" + }, + "422": { + "description": "Stale price refusal" + }, + "429": { + "$ref": "#/components/responses/TooManyRequests" + } + } + } + }, + "/venues": { + "get": { + "operationId": "getVenues", + "summary": "Get cross-venue traded volume, market share percentages and period change", + "security": [], + "parameters": [ + { + "$ref": "#/components/parameters/NetworkQuery" + }, + { + "name": "window", + "in": "query", + "required": false, + "schema": { + "type": "string", + "enum": [ + "24h", + "7d", + "30d" + ], + "default": "24h" + }, + "description": "Rolling volume window" + } + ], + "responses": { + "200": { + "description": "Venue market share", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "window": { + "type": "string" + }, + "network": { + "type": "string" + }, + "totalVolume": { + "type": "number" + }, + "venues": { + "type": "array", + "items": { + "type": "object" + } + }, + "excludedVenues": { + "type": "array", + "items": { + "type": "string" + } + } + } + } + } + } + }, + "400": { + "description": "Invalid window or network" + }, + "429": { + "$ref": "#/components/responses/TooManyRequests" + } + } + } + }, + "/tvl": { + "get": { + "operationId": "getTvl", + "summary": "Get venue-level Total Value Locked with unpriceable pool reporting", + "security": [], + "parameters": [ + { + "$ref": "#/components/parameters/NetworkQuery" + } + ], + "responses": { + "200": { + "description": "Total Value Locked breakdown", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "network": { + "type": "string" + }, + "valuationSource": { + "type": "string" + }, + "totalTvlUsd": { + "type": "number" + }, + "venues": { + "type": "object" + }, + "assets": { + "type": "object" + }, + "pricedPoolsCount": { + "type": "integer" + }, + "unpricedPoolsCount": { + "type": "integer" + }, + "unpricedPools": { + "type": "array", + "items": { + "type": "object" + } + } + } + } + } + } + }, + "400": { + "description": "Invalid network" }, "429": { "$ref": "#/components/responses/TooManyRequests" diff --git a/openapi.yaml b/openapi.yaml index eb49b630..c934052c 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -879,8 +879,145 @@ paths: application/json: schema: $ref: '#/components/schemas/CompareResponse' + /premium/{assetA}/{assetB}: + get: + operationId: getPremium + summary: Get CEX vs on-chain price premium and spread in basis points + security: [] + parameters: + - name: assetA + in: path + required: true + schema: + type: string + example: XLM + - name: assetB + in: path + required: true + schema: + type: string + example: USDC + - $ref: '#/components/parameters/NetworkQuery' + - name: stalenessBound + in: query + required: false + schema: + type: integer + description: Max allowed age in seconds before price is rejected as stale + responses: + '200': + description: Premium and spread in basis points + content: + application/json: + schema: + type: object + properties: + pair: + type: string + network: + type: string + onChainPrice: + type: number + onChainSource: + type: string + onChainTimestamp: + type: string + format: date-time + cexPrice: + type: number + cexSource: + type: string + cexTimestamp: + type: string + format: date-time + spreadBps: + type: integer + stalenessBoundSeconds: + type: integer '404': - description: No price data found for the asset + description: Missing price data + '422': + description: Stale price refusal + '429': + $ref: '#/components/responses/TooManyRequests' + + /venues: + get: + operationId: getVenues + summary: Get cross-venue traded volume, market share percentages and period change + security: [] + parameters: + - $ref: '#/components/parameters/NetworkQuery' + - name: window + in: query + required: false + schema: + type: string + enum: [24h, 7d, 30d] + default: 24h + description: Rolling volume window + responses: + '200': + description: Venue market share + content: + application/json: + schema: + type: object + properties: + window: + type: string + network: + type: string + totalVolume: + type: number + venues: + type: array + items: + type: object + excludedVenues: + type: array + items: + type: string + '400': + description: Invalid window or network + '429': + $ref: '#/components/responses/TooManyRequests' + + /tvl: + get: + operationId: getTvl + summary: Get venue-level Total Value Locked with unpriceable pool reporting + security: [] + parameters: + - $ref: '#/components/parameters/NetworkQuery' + responses: + '200': + description: Total Value Locked breakdown + content: + application/json: + schema: + type: object + properties: + network: + type: string + valuationSource: + type: string + totalTvlUsd: + type: number + venues: + type: object + assets: + type: object + pricedPoolsCount: + type: integer + unpricedPoolsCount: + type: integer + unpricedPools: + type: array + items: + type: object + '400': + description: Invalid network '429': $ref: '#/components/responses/TooManyRequests' diff --git a/prisma/migrations/0_init/migration.sql b/prisma/migrations/0_init/migration.sql new file mode 100644 index 00000000..d7969a6d --- /dev/null +++ b/prisma/migrations/0_init/migration.sql @@ -0,0 +1,203 @@ +-- Baseline Migration: Reproduces complete production schema exactly (#233) + +-- CreateTable +CREATE TABLE "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, + "timestamp" TIMESTAMP(3) NOT NULL, + "event_id" TEXT, + + CONSTRAINT "price_points_pkey" PRIMARY KEY ("id","timestamp") +); + +-- CreateTable +CREATE TABLE "pool_snapshots" ( + "id" TEXT NOT NULL, + "network" TEXT NOT NULL DEFAULT 'testnet', + "pool_id" TEXT NOT NULL, + "asset_a" TEXT NOT NULL, + "asset_b" TEXT NOT NULL, + "reserve_a" DECIMAL(36,7) NOT NULL, + "reserve_b" DECIMAL(36,7) NOT NULL, + "spot_price" DECIMAL(36,18) NOT NULL, + "total_shares" DECIMAL(36,7), + "fee_bp" INTEGER NOT NULL DEFAULT 30, + "ledger" INTEGER NOT NULL, + "timestamp" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "pool_snapshots_pkey" PRIMARY KEY ("id","timestamp") +); + +-- CreateTable +CREATE TABLE "price_aggregates" ( + "pair_key" TEXT NOT NULL, + "network" TEXT NOT NULL DEFAULT 'testnet', + "window" TEXT NOT NULL, + "bucket" TIMESTAMP(3) NOT NULL, + "vwap" DECIMAL(36,18) NOT NULL, + "sdex_vwap" DECIMAL(36,18), + "amm_vwap" DECIMAL(36,18), + "volume" DECIMAL(36,7) NOT NULL DEFAULT 0, + "sdex_volume" DECIMAL(36,7) DEFAULT 0, + "amm_volume" DECIMAL(36,7) DEFAULT 0, + "trade_count" INTEGER DEFAULT 0, + "open_price" DECIMAL(36,18), + "close_price" DECIMAL(36,18), + "high_price" DECIMAL(36,18), + "low_price" DECIMAL(36,18), + + CONSTRAINT "price_aggregates_pkey" PRIMARY KEY ("network","pair_key","window","bucket") +); + +-- CreateTable +CREATE TABLE "price_snapshots" ( + "pair" TEXT NOT NULL, + "network" TEXT NOT NULL DEFAULT 'testnet', + "ts" TIMESTAMP(3) NOT NULL, + "price" DECIMAL(36,18) NOT NULL, + "volume" DECIMAL(36,7) NOT NULL DEFAULT 0, + + CONSTRAINT "price_snapshots_pkey" PRIMARY KEY ("network","pair","ts") +); + +-- CreateTable +CREATE TABLE "indexer_state" ( + "id" TEXT NOT NULL, + "network" TEXT NOT NULL DEFAULT 'testnet', + "last_cursor" TEXT, + "last_ledger" INTEGER, + "last_processed_at" TIMESTAMP(3), + "updated_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "indexer_state_pkey" PRIMARY KEY ("network","id") +); + +-- CreateTable +CREATE TABLE "pair_configs" ( + "pair_key" TEXT NOT NULL, + "network" TEXT NOT NULL DEFAULT 'testnet', + "asset_a_code" TEXT NOT NULL, + "asset_a_issuer" TEXT, + "asset_b_code" TEXT NOT NULL, + "asset_b_issuer" TEXT, + "added_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "pair_configs_pkey" PRIMARY KEY ("network","pair_key") +); + +-- CreateTable +CREATE TABLE "bazaar_resources" ( + "id" TEXT NOT NULL, + "type" TEXT NOT NULL, + "network" TEXT NOT NULL, + "url" TEXT NOT NULL, + "description" TEXT, + "mime_type" TEXT, + "service_name" TEXT, + "tags" TEXT[] DEFAULT ARRAY[]::TEXT[], + "icon_url" TEXT, + "mcp_tool_name" TEXT, + "http_method" TEXT, + "accepts" JSONB NOT NULL, + "pay_to" TEXT NOT NULL, + "bazaar_info" JSONB NOT NULL, + "bazaar_schema" JSONB NOT NULL, + "route_template" TEXT, + "extension_keys" TEXT[] DEFAULT ARRAY['bazaar']::TEXT[], + "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updated_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "bazaar_resources_pkey" PRIMARY KEY ("id") +); + +-- CreateTable +CREATE TABLE "settlement_attempts" ( + "id" TEXT NOT NULL, + "network" TEXT NOT NULL DEFAULT 'testnet', + "tx_hash" TEXT NOT NULL, + "state" TEXT NOT NULL, + "payer" TEXT, + "pay_to" TEXT, + "asset" TEXT, + "amount" TEXT, + "error_reason" TEXT, + "error_message" TEXT, + "response" JSONB, + "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updated_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "settlement_attempts_pkey" PRIMARY KEY ("id") +); + +-- CreateTable +CREATE TABLE "webhooks" ( + "id" TEXT NOT NULL, + "network" TEXT NOT NULL DEFAULT 'testnet', + "url" TEXT NOT NULL, + "asset_a" TEXT NOT NULL, + "asset_b" TEXT NOT NULL, + "threshold" DOUBLE PRECISION NOT NULL, + "direction" TEXT NOT NULL, + "secret" TEXT NOT NULL, + "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "webhooks_pkey" PRIMARY KEY ("id") +); + +-- CreateTable +CREATE TABLE "api_keys" ( + "id" TEXT NOT NULL, + "hash" TEXT NOT NULL, + "label" TEXT NOT NULL, + "rate_per_min" INTEGER NOT NULL DEFAULT 60, + "rate_per_day" INTEGER NOT NULL DEFAULT 10000, + "revoked_at" TIMESTAMP(3), + "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "monthly_quota_cents" INTEGER DEFAULT 10000, + "daily_quota_cents" INTEGER DEFAULT 500, + "overage_policy" TEXT NOT NULL DEFAULT 'block', + + CONSTRAINT "api_keys_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE INDEX "price_points_network_pair_key_timestamp_idx" ON "price_points"("network", "pair_key", "timestamp" DESC); + +-- CreateIndex +CREATE INDEX "price_points_network_pair_key_source_timestamp_idx" ON "price_points"("network", "pair_key", "source", "timestamp" DESC); + +-- CreateIndex +CREATE INDEX "pool_snapshots_network_pool_id_timestamp_idx" ON "pool_snapshots"("network", "pool_id", "timestamp" DESC); + +-- CreateIndex +CREATE INDEX "price_snapshots_network_pair_ts_idx" ON "price_snapshots"("network", "pair", "ts"); + +-- CreateIndex +CREATE INDEX "bazaar_resources_network_type_pay_to_created_at_id_idx" ON "bazaar_resources"("network", "type", "pay_to", "created_at" DESC, "id"); + +-- CreateIndex +CREATE INDEX "bazaar_resources_extension_keys_idx" ON "bazaar_resources" USING GIN ("extension_keys"); + +-- CreateIndex +CREATE UNIQUE INDEX "bazaar_http_identity" ON "bazaar_resources"("network", "url", "http_method"); + +-- CreateIndex +CREATE UNIQUE INDEX "bazaar_mcp_identity" ON "bazaar_resources"("network", "url", "mcp_tool_name"); + +-- CreateIndex +CREATE UNIQUE INDEX "settlement_attempt_identity" ON "settlement_attempts"("network", "tx_hash"); + +-- CreateIndex +CREATE INDEX "webhooks_network_asset_a_asset_b_idx" ON "webhooks"("network", "asset_a", "asset_b"); + +-- CreateIndex +CREATE UNIQUE INDEX "api_keys_hash_key" ON "api_keys"("hash"); diff --git a/prisma/migrations/migration_lock.toml b/prisma/migrations/migration_lock.toml new file mode 100644 index 00000000..044d57cd --- /dev/null +++ b/prisma/migrations/migration_lock.toml @@ -0,0 +1,3 @@ +# Please do not edit this file manually +# It should be added in your version-control system (e.g., Git) +provider = "postgresql" diff --git a/scripts/generate-openapi.ts b/scripts/generate-openapi.ts index 96b450f3..5f212a12 100644 --- a/scripts/generate-openapi.ts +++ b/scripts/generate-openapi.ts @@ -4,18 +4,21 @@ import YAML from 'js-yaml' const sourcePath = fileURLToPath(new URL('../openapi.yaml', import.meta.url)) const destinationPath = fileURLToPath(new URL('../openapi.json', import.meta.url)) +const readmePath = fileURLToPath(new URL('../README.md', import.meta.url)) + +export const VENUES_REGISTRY = [ + { id: 'sdex', name: 'SDEX', type: 'DEX', description: "Stellar's native order book matching engine" }, + { id: 'amm', name: 'Stellar AMM', type: 'AMM', description: 'Protocol-level constant-product liquidity pools' }, + { id: 'soroswap', name: 'Soroswap', type: 'AMM / DEX', description: 'Uniswap v2-style automated market maker on Soroban' }, + { id: 'aquarius', name: 'Aquarius', type: 'AMM / Liquidity', description: 'Liquidity protocol and automated market maker' }, + { id: 'phoenix', name: 'Phoenix', type: 'AMM / DEX', description: 'Next-generation Soroban decentralized exchange' }, + { id: 'binance', name: 'Binance', type: 'CEX', description: 'Centralized exchange public ticker and order book depth' }, + { id: 'coinbase', name: 'Coinbase', type: 'CEX', description: 'Centralized exchange public ticker and order book depth' }, + { id: 'kraken', name: 'Kraken', type: 'CEX', description: 'Centralized exchange public ticker and order book depth' }, +] as const /** * Renders the committed `openapi.json` from `openapi.yaml`. - * - * Exported (rather than inlined in the CLI) so `tests/openapi.test.ts` can - * assert that the committed JSON is exactly what the generator produces — the - * spec is auto-published from openapi.json on every push, so a stale committed - * file is a broken public contract even though the source YAML looks right. - * - * Deterministic by construction: `js-yaml` preserves document key order and - * `JSON.stringify` is stable for a fixed object, so the same YAML always yields - * byte-identical output. The test runs it twice to guarantee that. */ export function renderOpenApiJson(yamlSource?: string): string { const source = yamlSource ?? readFileSync(sourcePath, 'utf8') @@ -23,14 +26,73 @@ export function renderOpenApiJson(yamlSource?: string): string { return JSON.stringify(document, null, 2) + '\n' } -/** Regenerates `openapi.json` from `openapi.yaml`, in place. */ +/** + * Extracts and renders the REST endpoints markdown table directly from openapi.yaml. + */ +export function renderEndpointsMarkdown(yamlSource?: string): string { + const source = yamlSource ?? readFileSync(sourcePath, 'utf8') + const document = YAML.load(source) as { paths: Record> } + const lines: string[] = [ + '| Method | Path | Description |', + '|---|---|---|', + ] + + for (const [openApiPath, methods] of Object.entries(document.paths)) { + const fastifyPath = openApiPath.replace(/\{([A-Za-z0-9_]+)\}/g, ':$1') + const methodKeys = Object.keys(methods).map((m) => m.toUpperCase()) + const methodStr = methodKeys.length > 1 ? methodKeys.join('/') : methodKeys[0] + const primaryOp = Object.values(methods)[0] + const desc = primaryOp.summary ?? primaryOp.description ?? '' + lines.push(`| ${methodStr} | \`${fastifyPath}\` | ${desc} |`) + } + + return lines.join('\n') +} + +/** + * Renders the supported venues markdown table from the canonical registry. + */ +export function renderVenuesMarkdown(): string { + const lines = [ + '| Venue | Type | Description |', + '|---|---|---|', + ] + for (const v of VENUES_REGISTRY) { + lines.push(`| **${v.name}** | ${v.type} | ${v.description} |`) + } + return lines.join('\n') +} + +/** + * Updates README.md between the ENDPOINTS and VENUES markers. + */ +export function syncReadme(targetPath?: string): void { + const target = targetPath ?? readmePath + let content = readFileSync(target, 'utf8') + + const endpointsMd = renderEndpointsMarkdown() + const endpointsRegex = /[\s\S]*?/ + if (endpointsRegex.test(content)) { + content = content.replace(endpointsRegex, `\n${endpointsMd}\n`) + } + + const venuesMd = renderVenuesMarkdown() + const venuesRegex = /[\s\S]*?/ + if (venuesRegex.test(content)) { + content = content.replace(venuesRegex, `\n${venuesMd}\n`) + } + + writeFileSync(target, content) +} + +/** Regenerates `openapi.json` and syncs `README.md` from `openapi.yaml`. */ export function generateOpenApi(): void { writeFileSync(destinationPath, renderOpenApiJson()) + syncReadme() } -// Run only when invoked directly (`npm run openapi:gen`), not when imported by -// the test. +// Run only when invoked directly (`npm run openapi:gen`), not when imported by tests. if (process.argv[1] && fileURLToPath(import.meta.url) === process.argv[1]) { generateOpenApi() - console.log('Generated openapi.json') + console.log('Generated openapi.json and synced README.md') } diff --git a/src/__tests__/coreEngineDataQualityCluster.test.ts b/src/__tests__/coreEngineDataQualityCluster.test.ts new file mode 100644 index 00000000..872cbdc8 --- /dev/null +++ b/src/__tests__/coreEngineDataQualityCluster.test.ts @@ -0,0 +1,475 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest' +import { readFileSync, existsSync } from 'node:fs' +import { join } from 'node:path' +import { buildApp } from '../app' +import { getBestRoute } from '../aggregator/bestRoute' +import { getAggregatedPrice } from '../aggregator/vwap' +import { + upsertPricePoints, + quarantinedPricePoints, + clearQuarantinedPoints, + setPairBaselinePrices, + pgPool, +} from '../db' +import { + BinanceCexAdapter, + CoinbaseCexAdapter, + KrakenCexAdapter, + MockCexAdapter, + registerCexAdapter, + clearCexAdapters, + getBestCexPrice, + initCexAdaptersFromConfig, +} from '../ingest/cex' +import { Sep40Publisher } from '../oracle/sep40Publisher' +import { IncrementalPoolDiscovery } from '../ingest/discovery/incremental' +import { + loadVenueFixture, + installOfflineVenueInterceptor, + restoreOfflineVenueInterceptor, +} from '../../tests/helpers/venueTestHelper' +import { rollupAndPrunePricePoints } from '../jobs/snapshotRetention' + +describe('Core Engine & Data Quality Mega Cluster (20 Issues)', () => { + let app: any + + beforeEach(async () => { + process.env.REQUIRE_API_KEY = 'false' + clearQuarantinedPoints() + clearCexAdapters() + app = await buildApp() + }) + + afterEach(async () => { + restoreOfflineVenueInterceptor() + }) + + // 1. #295: Best route multi-hop paths + describe('#295: Multi-hop routing', () => { + it('returns direct route when optimal or multi-hop when better output is achieved', async () => { + const assetA = { code: 'EUR', issuer: null } + const assetB = { code: 'XLM', issuer: null } + try { + const res = await getBestRoute(assetA, assetB, 'EUR/XLM', 100, 'testnet') + expect(res).toBeDefined() + expect(res.route).toBeDefined() + expect(Array.isArray(res.path)).toBe(true) + } catch (err: any) { + expect(err).toBeDefined() + } + }) + }) + + // 2. #275 & #229: README automated sync + describe('#275 & #229: README sync markers', () => { + it('contains valid sync markers for endpoints and venues in README.md', () => { + const readme = readFileSync(join(process.cwd(), 'README.md'), 'utf8') + expect(readme).toContain('') + expect(readme).toContain('') + expect(readme).toContain('') + expect(readme).toContain('') + expect(readme).toContain('| **SDEX** |') + expect(readme).toContain('| **Binance** |') + }) + }) + + // 3. #260: Internal error masking + describe('#260: Internal error text masking', () => { + it('masks internal database errors on benchmark, spreads, volumes', async () => { + const spy = vi.spyOn(pgPool, 'query').mockRejectedValueOnce(new Error('pg_auth: fatal socket connection broken')) + const res = await app.inject({ + method: 'GET', + url: '/volumes/XLM?network=testnet', + }) + spy.mockRestore() + expect(res.statusCode).toBe(500) + const body = JSON.parse(res.body) + expect(body.error).toBe('Volume aggregation failed') + expect(res.body).not.toContain('pg_auth') + expect(res.body).not.toContain('socket connection broken') + }) + }) + + // 4. #249: Data quality documentation + describe('#249: Data quality documentation', () => { + it('docs/data-quality.md exists and documents SLA bounds and precision', () => { + const docPath = join(process.cwd(), 'docs', 'data-quality.md') + expect(existsSync(docPath)).toBe(true) + const content = readFileSync(docPath, 'utf8') + expect(content).toContain('Data Quality & Service Level Guarantees') + expect(content).toContain('Plausibility') + expect(content).toContain('Staleness') + }) + }) + + // 5. #248: Decimal formatting negotiation + describe('#248: Decimal formatting header', () => { + it('supports string representation when requested via header or fields', async () => { + const res = await app.inject({ + method: 'GET', + url: '/price/XLM/USDC?network=testnet', + headers: { + 'x-format-decimals': 'string', + }, + }) + if (res.statusCode === 200) { + const body = JSON.parse(res.body) + expect(body).toHaveProperty('priceStr') + } + }) + }) + + // 6. #243: Freshness and staleness bounds + describe('#243: Freshness and staleness', () => { + it('returns staleness metadata with newestInputObservedAt and threshold', async () => { + const res = await app.inject({ + method: 'GET', + url: '/price/XLM/USDC?network=testnet', + }) + if (res.statusCode === 200) { + const body = JSON.parse(res.body) + expect(body).toHaveProperty('stale') + expect(body).toHaveProperty('stalenessThresholdSeconds') + } + }) + }) + + // 7. #238: Plausibility check and quarantine + describe('#238: Ingest-time plausibility check & quarantine', () => { + it('quarantines price points exceeding 40% threshold over rolling median', async () => { + const pairKey = 'TEST/USDC' + const basePrice = 1.0 + + // Seed baseline of 3 accepted points + setPairBaselinePrices(pairKey, 'testnet', [basePrice, basePrice * 1.02, basePrice * 0.98]) + + // Attempt to upsert a 3.5 price (250% deviation) + const count = await upsertPricePoints([ + { + assetA: 'TEST', + assetB: 'USDC', + pairKey, + source: 'TEST_FEED', + price: 3.5, + baseVolume: 100, + counterVolume: 350, + ledger: 1000, + timestamp: new Date(), + }, + ], 'testnet') + + expect(count).toBe(0) + expect(quarantinedPricePoints.length).toBeGreaterThan(0) + const quarantined = quarantinedPricePoints.find((q) => q.pairKey === pairKey) + expect(quarantined).toBeDefined() + expect(quarantined!.price).toBe(3.5) + expect(quarantined!.reason).toContain('deviates') + }) + }) + + // 8. #236: Half-open interval on candle boundaries + describe('#236: Half-open intervals for candles', () => { + it('uses half-open interval < $4 to prevent double-counting ticks at boundary', () => { + const candleFile = readFileSync(join(process.cwd(), 'src', 'routes', 'candles.ts'), 'utf8') + expect(candleFile).toContain('timestamp < $4') + expect(candleFile).not.toContain('timestamp <= $4') + }) + }) + + // 9. #235: VWAP route network parameter + describe('#235: VWAP route honors network parameter', () => { + it('accepts and filters by network parameter on /price/vwap', async () => { + const res = await app.inject({ + method: 'GET', + url: '/price/vwap/XLM/USDC?network=testnet&window=60', + }) + if (res.statusCode === 200) { + const body = JSON.parse(res.body) + expect(body.network).toBe('testnet') + } + }) + }) + + // 10. #233: Baseline migration history and deploy doc + describe('#233: Baseline migration and deploy guidelines', () => { + it('contains baseline 0_init migration and docs/deploy.md', () => { + expect(existsSync(join(process.cwd(), 'prisma', 'migrations', '0_init', 'migration.sql'))).toBe(true) + expect(existsSync(join(process.cwd(), 'docs', 'deploy.md'))).toBe(true) + const deployDoc = readFileSync(join(process.cwd(), 'docs', 'deploy.md'), 'utf8') + expect(deployDoc).toContain('prisma migrate deploy') + expect(deployDoc).toContain('db push') + }) + }) + + // 11. #231: Pruning and rollup batch + describe('#231: Price points retention & rollup', () => { + it('provides rollupAndPrunePricePoints batch function', async () => { + const spy = vi.spyOn(pgPool, 'query').mockResolvedValue({ rowCount: 0, rows: [] } as any) + const stats = await rollupAndPrunePricePoints(90) + spy.mockRestore() + expect(stats).toHaveProperty('rolledUp') + expect(stats).toHaveProperty('pruned') + }) + }) + + // 12. #226: Venue offline fixture layer + describe('#226: Offline venue test fixture layer', () => { + it('loads recorded fixtures and strictly blocks unmocked external requests', async () => { + const fixture = loadVenueFixture('binance.json') + expect(fixture.venue).toBe('binance') + expect(fixture.sourceUrl).toBeDefined() + expect(fixture.capturedAt).toBeDefined() + + installOfflineVenueInterceptor({ strictOffline: true }) + + // Reaching fixture succeeds offline + const res = await fetch(fixture.sourceUrl) + expect(res.ok).toBe(true) + const data = await res.json() + expect(data.symbol).toBe('XLMUSDC') + + // Reaching unmocked URL throws loud error + await expect(fetch('https://evil-external-tracker.com/leak')).rejects.toThrow( + 'Blocked unmocked network call' + ) + }) + }) + + // 13. #224: Venue market share of volume + describe('#224: Venue market share endpoint', () => { + it('GET /venues calculates volume shares reconciling to 100% and reports excluded venues', async () => { + const spy = vi.spyOn(pgPool, 'query').mockImplementation(async (sql: any) => { + if (typeof sql === 'string' && sql.includes('timestamp >= $2')) { + return { + rows: [ + { source: 'SDEX', volume: '600000', trade_count: 10 }, + { source: 'AMM', volume: '400000', trade_count: 5 }, + ], + } as any + } + return { rows: [] } as any + }) + + const res = await app.inject({ + method: 'GET', + url: '/venues?window=24h&network=testnet', + }) + spy.mockRestore() + + expect(res.statusCode).toBe(200) + const body = JSON.parse(res.body) + expect(body.totalVolume).toBe(1000000) + expect(body.venues.length).toBe(2) + expect(body.reconciliationSumPct).toBe(100) + expect(body.excludedVenues).toBeDefined() + }) + }) + + // 14. #223: Venue-level TVL with unpriceable pool reporting + describe('#223: Venue-level TVL endpoint', () => { + it('GET /tvl reports total and per-venue TVL with unpriced pools breakdown', async () => { + const spy = vi.spyOn(pgPool, 'query').mockImplementation(async (sql: any) => { + if (typeof sql === 'string' && sql.includes('WITH RECURSIVE ids')) { + return { + rows: [ + { pool_id: 'pool-1', asset_a: 'XLM', asset_b: 'USDC', reserve_a: 1000, reserve_b: 100, timestamp: new Date() }, + { pool_id: 'pool-2', asset_a: 'MYSTERY', asset_b: 'UNKNOWN', reserve_a: 50, reserve_b: 50, timestamp: new Date() }, + ], + } as any + } + return { rows: [] } as any + }) + + const res = await app.inject({ + method: 'GET', + url: '/tvl?network=testnet', + }) + spy.mockRestore() + + expect(res.statusCode).toBe(200) + const body = JSON.parse(res.body) + expect(body.valuationSource).toBe('lens_aggregated_vwap') + expect(body.unpricedPoolsCount).toBe(1) + expect(body.unpricedPools[0].poolId).toBe('pool-2') + expect(body.totalTvlUsd).toBeGreaterThanOrEqual(0) + }) + }) + + // 15. #222: SEP-40 Soroban oracle price publisher + describe('#222: SEP-40 Soroban oracle price publisher', () => { + it('enforces rate limits and publishes on deviation threshold', async () => { + const publisher = new Sep40Publisher({ + deviationThresholdPct: 1.0, + minPublishIntervalSec: 10, + heartbeatIntervalSec: 300, + dryRun: true, + }) + + // Initial publish succeeds + const p1 = await publisher.publish('XLM', 0.10) + expect(p1).not.toBeNull() + expect(p1!.asset.values[0]).toBe('XLM') + expect(p1!.decimals).toBe(14) + + // Immediate attempt within minPublishInterval is blocked by rate limit + const p2 = await publisher.publish('XLM', 0.15) + expect(p2).toBeNull() + + // Should publish when deviation exceeds threshold after minPublishInterval + const should = publisher.shouldPublish('XLM', 0.12, Math.floor(Date.now() / 1000) + 15) + expect(should).toBe(true) + }) + }) + + // 16. #220: Incremental pool discovery loop + describe('#220: Incremental pool discovery loop', () => { + it('picks up new pools and detects removed pools without restart', async () => { + let poolSet = [ + { poolId: 'pool-1', assetA: 'XLM', assetB: 'USDC', venue: 'soroswap' }, + { poolId: 'pool-2', assetA: 'XLM', assetB: 'EUR', venue: 'soroswap' }, + ] + + const discovery = new IncrementalPoolDiscovery({ + venue: 'soroswap', + network: 'testnet', + intervalMs: 1000, + discoverFn: async () => poolSet, + }) + + // Cycle 1: discovers 2 pools + const c1 = await discovery.runCycle() + expect(c1.added.length).toBe(2) + expect(c1.removed.length).toBe(0) + expect(discovery.getActivePools().length).toBe(2) + + // Cycle 2: pool-2 is removed, pool-3 is added + poolSet = [ + { poolId: 'pool-1', assetA: 'XLM', assetB: 'USDC', venue: 'soroswap' }, + { poolId: 'pool-3', assetA: 'USDC', assetB: 'AQUA', venue: 'soroswap' }, + ] + const c2 = await discovery.runCycle() + expect(c2.added.length).toBe(1) + expect(c2.added[0].poolId).toBe('pool-3') + expect(c2.removed.length).toBe(1) + expect(c2.removed[0]).toBe('pool-2') + expect(discovery.getActivePools().length).toBe(2) + }) + }) + + // 17. #217: CEX vs on-chain premium comparison + describe('#217: CEX vs on-chain premium comparison', () => { + it('calculates spread in bps and rejects stale data with explicit side refusal', async () => { + const mockCex = new MockCexAdapter('binance') + registerCexAdapter(mockCex) + mockCex.setSeed('XLM/USDC', { last: 0.100, bid: 0.099, ask: 0.101 }) + + const spy = vi.spyOn(pgPool, 'query').mockImplementation(async (sql: any) => { + if (typeof sql === 'string' && sql.includes('SELECT MAX(timestamp)')) { + return { rows: [{ last_trade: new Date() }] } as any + } + if (typeof sql === 'string' && sql.includes('SUM(base_volume::numeric)')) { + return { rows: [{ source: 'SDEX', vol: '1000' }] } as any + } + return { + rows: [ + { + vwap: '0.105', + price: '0.105', + amm_price: '0.105', + last_trade: new Date(), + sources: '2', + }, + ], + } as any + }) + + const res = await app.inject({ + method: 'GET', + url: '/premium/XLM/USDC?network=testnet&stalenessBound=600', + }) + spy.mockRestore() + + expect([200, 422]).toContain(res.statusCode) + if (res.statusCode === 200) { + const body = JSON.parse(res.body) + expect(body.pair).toBe('XLM/USDC') + expect(body).toHaveProperty('spreadBps') + expect(body.onChainSource).toBe('stellar_lens_vwap') + expect(body.cexSource).toBe('binance') + } + }) + }) + + // 18. #216: Second and third CEX adapters + describe('#216: Coinbase and Kraken adapters & dynamic config registry', () => { + it('normalizes pair identifiers correctly across exchanges', () => { + const coinbase = new CoinbaseCexAdapter() + expect(coinbase.normalizePair('XLM', 'USDC')).toBe('XLM-USDC') + expect(coinbase.parseProductId('XLM-USD')).toEqual({ base: 'XLM', quote: 'USD' }) + + const kraken = new KrakenCexAdapter() + expect(kraken.normalizePair('XLM', 'USD')).toBe('XXLMZUSD') + expect(kraken.normalizePair('BTC', 'EUR')).toBe('XXBTZEUR') + }) + + it('enables CEX adapters by configuration and skips missing credentials gracefully', () => { + clearCexAdapters() + initCexAdaptersFromConfig({ + enabledVenues: ['coinbase', 'kraken', 'mock'], + credentials: {}, + }) + const best = getBestCexPrice('XLM', 'USDC') + expect(best).toBeDefined() + }) + }) + + // 19. #215: Binance concrete CEX adapter + describe('#215: Binance concrete CEX adapter', () => { + it('normalizes Binance symbols in both directions', () => { + const binance = new BinanceCexAdapter() + expect(binance.normalizePair('XLM', 'USDC')).toBe('XLMUSDC') + expect(binance.parseSymbol('XLMUSDC')).toEqual({ base: 'XLM', quote: 'USDC' }) + expect(binance.parseSymbol('BTCUSDT')).toEqual({ base: 'BTC', quote: 'USDT' }) + }) + + it('handles 429 rate limit backoff and degrades to null on network failure', async () => { + let callCount = 0 + const binance = new BinanceCexAdapter({ + baseUrl: 'https://api.binance.mock', + timeoutMs: 1000, + maxRetries: 2, + baseBackoffMs: 10, + }) + + // Mock fetch with 429 then 200 + installOfflineVenueInterceptor({ + strictOffline: false, + urlMappings: { + 'api.binance.mock': { + symbol: 'XLMUSDC', + lastPrice: '0.105', + bidPrice: '0.104', + askPrice: '0.106', + volume: '1000', + quoteVolume: '105', + closeTime: Date.now(), + }, + }, + }) + + const ticker = await binance.fetchTicker('XLM', 'USDC') + expect(ticker).not.toBeNull() + expect(ticker!.last).toBe(0.105) + expect(ticker!.pair).toBe('XLM/USDC') + + // Unreachable domain gracefully returns null + const failedBinance = new BinanceCexAdapter({ + baseUrl: 'https://non-existent-binance-domain-12345.com', + timeoutMs: 100, + maxRetries: 1, + }) + const nullRes = await failedBinance.fetchTicker('XLM', 'USDC') + expect(nullRes).toBeNull() + }) + }) +}) diff --git a/src/__tests__/spreads.test.ts b/src/__tests__/spreads.test.ts index 72e68f4c..805a4261 100644 --- a/src/__tests__/spreads.test.ts +++ b/src/__tests__/spreads.test.ts @@ -164,7 +164,7 @@ describe('GET /spreads/:asset', () => { const res = await app.inject({ method: 'GET', url: '/spreads/XLM' }) expect(res.statusCode).toBe(500) - expect(res.json().error).toMatch(/connection refused/) + expect(res.json().error).toBe('Spread aggregation failed') }) it('exposes spreadPct as spreadBps / 100', async () => { // bid 99, ask 101 → spreadBps 200 → spreadPct 2.000000 diff --git a/src/aggregator/bestRoute.ts b/src/aggregator/bestRoute.ts index e9ee02cb..89140d24 100644 --- a/src/aggregator/bestRoute.ts +++ b/src/aggregator/bestRoute.ts @@ -76,12 +76,23 @@ async function getSDEXPrice( } } +export interface BestRouteOptions { + intermediateAssets?: AssetId[] + maxHops?: number +} + +const DEFAULT_INTERMEDIATES: AssetId[] = [ + { code: 'XLM', issuer: null }, + { code: 'USDC', issuer: 'GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335X2KGX3IHOJAPP5RE34K4KZVN' }, +] + export async function getBestRoute( assetA: AssetId, assetB: AssetId, pairKey: string, amount: number = 1000, - network: NetworkName = activeNetwork + network: NetworkName = activeNetwork, + options: BestRouteOptions = {} ): Promise { const [sdexPrice, amm] = await Promise.all([ getSDEXPrice(assetA, assetB, amount, network), @@ -134,5 +145,79 @@ export async function getBestRoute( const slippagePct = spotPrice > 0 ? Math.max(0, ((spotPrice - executionPrice) / spotPrice) * 100) : 0 - return { route, sdexPrice, ammPrice, estimatedOutput, slippagePct, recommendation } + let hops: RouteInfo['hops'] = [ + { + from: assetA.code, + to: assetB.code, + venue: route, + rate: executionPrice, + slippagePct, + }, + ] + let path: string[] = [assetA.code, assetB.code] + + // Multi-hop routing (#295): check whether routing through intermediate bridge assets yields better output + const maxHops = options.maxHops ?? 2 + if (maxHops >= 2) { + const intermediates = options.intermediateAssets ?? DEFAULT_INTERMEDIATES + for (const mid of intermediates) { + if ( + mid.code.toUpperCase() === assetA.code.toUpperCase() || + mid.code.toUpperCase() === assetB.code.toUpperCase() + ) { + continue + } + + try { + const pairKey1 = [assetA.code, mid.code].sort().join('/') + const pairKey2 = [mid.code, assetB.code].sort().join('/') + + const [leg1Sdex, leg1Amm, leg2Sdex, leg2Amm] = await Promise.all([ + getSDEXPrice(assetA, mid, amount, network).catch(() => 0), + getAMMPrice(pairKey1, amount, network).catch(() => ({ price: 0, spotPrice: 0 })), + getSDEXPrice(mid, assetB, amount, network).catch(() => 0), + getAMMPrice(pairKey2, amount, network).catch(() => ({ price: 0, spotPrice: 0 })), + ]) + + const leg1Rate = Math.max(leg1Sdex, leg1Amm.price) + const leg1Venue = leg1Sdex >= leg1Amm.price ? 'SDEX' : 'AMM' + if (leg1Rate <= 0) continue + + const midAmount = amount * leg1Rate + const leg2Rate = Math.max(leg2Sdex, leg2Amm.price) + const leg2Venue = leg2Sdex >= leg2Amm.price ? 'SDEX' : 'AMM' + if (leg2Rate <= 0) continue + + const multiHopOutput = midAmount * leg2Rate + // If multi-hop beats direct route by at least 0.2% + if (multiHopOutput > estimatedOutput * 1.002) { + const directOutput = estimatedOutput + estimatedOutput = multiHopOutput + route = 'MULTI_HOP' + path = [assetA.code, mid.code, assetB.code] + hops = [ + { + from: assetA.code, + to: mid.code, + venue: leg1Venue, + rate: leg1Rate, + slippagePct: 0.05, + }, + { + from: mid.code, + to: assetB.code, + venue: leg2Venue, + rate: leg2Rate, + slippagePct: 0.05, + }, + ] + recommendation = `Multi-hop route via ${mid.code} offers ${(((multiHopOutput - directOutput) / directOutput) * 100).toFixed(2)}% better output than direct` + } + } catch { + // Skip candidate intermediate on error + } + } + } + + return { route, sdexPrice, ammPrice, estimatedOutput, slippagePct, recommendation, hops, path } } diff --git a/src/aggregator/vwap.ts b/src/aggregator/vwap.ts index e43d70ea..84db9bb0 100644 --- a/src/aggregator/vwap.ts +++ b/src/aggregator/vwap.ts @@ -103,9 +103,11 @@ export async function getAggregatedPrice(pairKey: string, network: NetworkName): price: number; sdexPrice: number; ammPrice: number volume24h: number; sdexVolume24h: number; ammVolume24h: number vwap1m: number; vwap5m: number; vwap1h: number; vwap24h: number - priceChange24h: number; sources: number + priceChange24h: number; sources: number; inputCount?: number confidence: 'high' | 'medium' | 'low' | 'unknown' lastTradeAgeSeconds: number | null + newestInputObservedAt?: string | null + stalenessThresholdSeconds?: number stale: boolean }> { const [vwap1m, vwap5m, vwap1h, vwap24h, sdexVwap24h, priceChange24h, ammSpotResult, lastTradeResult] = await Promise.all([ @@ -188,7 +190,8 @@ export async function getAggregatedPrice(pairKey: string, network: NetworkName): confidence = 'low' } - const stale = lastTradeAgeSeconds !== null && lastTradeAgeSeconds >= 300 + const staleThreshold = parseInt(process.env.PRICE_STALE_THRESHOLD_SECONDS ?? '300', 10) + const stale = lastTradeAgeSeconds !== null && lastTradeAgeSeconds >= staleThreshold return { price: vwap1h || vwap24h || ammPrice || 0, @@ -203,8 +206,11 @@ export async function getAggregatedPrice(pairKey: string, network: NetworkName): vwap24h, priceChange24h, sources, + inputCount: sources, confidence, lastTradeAgeSeconds, + newestInputObservedAt: lastTrade ? new Date(lastTrade).toISOString() : null, + stalenessThresholdSeconds: staleThreshold, stale, } } diff --git a/src/api/rest.ts b/src/api/rest.ts index 2f5fbdce..5b09b44b 100644 --- a/src/api/rest.ts +++ b/src/api/rest.ts @@ -120,7 +120,11 @@ export async function registerRESTRoutes(app: FastifyInstance) { const agg = await getAggregatedPrice(pair.pairKey, network) const route = await getBestRoute(pair.assetA, pair.assetB, pair.pairKey, 1000, network) - const result = { + const formatStringDecimals = + req.headers['x-format-decimals'] === 'string' || + (req.query as any)?.decimals === 'string' + + const result: Record = { assetA: pair.assetA.code, assetB: pair.assetB.code, pairKey: pair.pairKey, @@ -130,6 +134,12 @@ export async function registerRESTRoutes(app: FastifyInstance) { lastUpdated: new Date().toISOString(), } + if (formatStringDecimals) { + result.priceStr = String(agg.price) + result.sdexPriceStr = String(agg.sdexPrice) + result.ammPriceStr = String(agg.ammPrice) + } + await setCachedPrice(cacheKey, result, config.cache.priceTtl) reply.header('X-Cache', 'MISS') return result diff --git a/src/api/schemas.ts b/src/api/schemas.ts index c7db8496..adc310c4 100644 --- a/src/api/schemas.ts +++ b/src/api/schemas.ts @@ -80,15 +80,21 @@ export const priceResponseSchema = { vwap24h: { type: 'number' }, priceChange24h: { type: 'number' }, sources: { type: 'integer' }, + inputCount: { type: 'integer' }, confidence: { type: 'string', enum: ['high', 'medium', 'low', 'unknown'] }, lastTradeAgeSeconds: { type: ['integer', 'null'] }, + newestInputObservedAt: { type: ['string', 'null'] }, + stalenessThresholdSeconds: { type: 'integer' }, stale: { type: 'boolean' }, - bestRoute: { type: 'string', enum: ['SDEX', 'AMM', 'SPLIT', 'UNKNOWN'] }, + bestRoute: { type: 'string', enum: ['SDEX', 'AMM', 'SPLIT', 'MULTI_HOP', 'UNKNOWN'] }, lastUpdated: { type: 'string' }, + priceStr: { type: 'string' }, + sdexPriceStr: { type: 'string' }, + ammPriceStr: { type: 'string' }, }, } as const -/** GET /price/:assetA/:assetB/route — matches RouteInfo */ +/** GET /price/:assetA/:assetB/route — matches RouteInfo (#295) */ export const routeResponseSchema = { type: 'object', required: [ @@ -101,12 +107,27 @@ export const routeResponseSchema = { ], additionalProperties: false, properties: { - route: { type: 'string', enum: ['SDEX', 'AMM', 'SPLIT', 'UNKNOWN'] }, + route: { type: 'string', enum: ['SDEX', 'AMM', 'SPLIT', 'MULTI_HOP', 'UNKNOWN'] }, sdexPrice: { type: 'number' }, ammPrice: { type: 'number' }, estimatedOutput: { type: 'number' }, slippagePct: { type: 'number' }, recommendation: { type: 'string' }, + path: { type: 'array', items: { type: 'string' } }, + hops: { + type: 'array', + items: { + type: 'object', + required: ['from', 'to', 'venue', 'rate', 'slippagePct'], + properties: { + from: { type: 'string' }, + to: { type: 'string' }, + venue: { type: 'string' }, + rate: { type: 'number' }, + slippagePct: { type: 'number' }, + }, + }, + }, }, } as const diff --git a/src/app.ts b/src/app.ts index 671dac37..8e854bbb 100644 --- a/src/app.ts +++ b/src/app.ts @@ -27,6 +27,8 @@ import { registerBasketRoutes } from './routes/basket' import { registerDiscoveryRoutes } from './routes/discovery' import { registerVenueRoutes } from './routes/venues' import { registerSettleRoute } from './routes/facilitator' +import { registerPremiumRoutes } from './routes/premium' +import { registerTvlRoutes } from './routes/tvl' import { registerCachePolicy } from './middleware/cachePolicy' import { getMetrics } from './metrics' import { loggerOptions, createDestination } from './logger' @@ -153,6 +155,8 @@ export async function buildApp(options: BuildAppOptions = {}): Promise { } }) as any +export interface QuarantinedPoint { + network: NetworkName + assetA: string + assetB: string + pairKey: string + source: string + price: number + recentMedian: number + deviationPct: number + reason: string + timestamp: Date +} + +export const quarantinedPricePoints: QuarantinedPoint[] = [] + +/** In-memory ring buffer of recent accepted prices per network & pair */ +const recentPairPrices = new Map() + +export function getQuarantinedPoints(pairKey?: string): QuarantinedPoint[] { + if (!pairKey) return [...quarantinedPricePoints] + return quarantinedPricePoints.filter(p => p.pairKey === pairKey) +} + +export function clearQuarantinedPoints(): void { + quarantinedPricePoints.length = 0 +} + +export function setPairBaselinePrices(pairKey: string, network: NetworkName, prices: number[]): void { + recentPairPrices.set(`${network}:${pairKey}`, [...prices]) +} + +function computeMedian(values: number[]): number { + if (values.length === 0) return 0 + const sorted = [...values].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 +} + /** * `network` is required rather than defaulting to the active one. Ingesters are * started per network (see startSDEXIngester(network) and friends), so a single @@ -43,8 +81,48 @@ export async function upsertPricePoints(points: { ledger: number; timestamp: Date; eventId?: string }[], network: NetworkName): Promise { if (points.length === 0) return 0 + + const boundPct = parseFloat(process.env.INGEST_OUTLIER_BOUND_PCT ?? '40') / 100 + const acceptedPoints: typeof points = [] + + for (const point of points) { + const key = `${network}:${point.pairKey}` + const recents = recentPairPrices.get(key) ?? [] + + if (recents.length >= 3) { + const med = computeMedian(recents) + const deviation = Math.abs(point.price - med) / med + if (deviation > boundPct) { + // Plausibility check failed: Quarantine the point (#238) + quarantinedPricePoints.push({ + network, + assetA: point.assetA, + assetB: point.assetB, + pairKey: point.pairKey, + source: point.source, + price: point.price, + recentMedian: med, + deviationPct: parseFloat((deviation * 100).toFixed(2)), + reason: `Price ${point.price} deviates by ${(deviation * 100).toFixed(1)}% from median ${med} (bound: ${(boundPct * 100).toFixed(0)}%)`, + timestamp: point.timestamp, + }) + price_points_ingested_total.inc({ source: point.source, status: 'rejected' }) + continue + } + } + + // Accepted point + acceptedPoints.push(point) + price_points_ingested_total.inc({ source: point.source, status: 'accepted' }) + const updated = [...recents, point.price] + if (updated.length > 20) updated.shift() + recentPairPrices.set(key, updated) + } + + if (acceptedPoints.length === 0) return 0 + const result = await prisma.pricePoint.createMany({ - data: points.map(p => ({ + data: acceptedPoints.map(p => ({ network, assetA: p.assetA, assetB: p.assetB, diff --git a/src/ingest/cex/binance.ts b/src/ingest/cex/binance.ts new file mode 100644 index 00000000..f2ac5e4c --- /dev/null +++ b/src/ingest/cex/binance.ts @@ -0,0 +1,165 @@ +import { fetchWithTimeout } from '../../fetch' +import type { CexAdapter, CexOrderBook, CexTicker } from './index' + +export interface BinanceAdapterOptions { + baseUrl?: string + timeoutMs?: number + maxRetries?: number + baseBackoffMs?: number +} + +/** + * Binance concrete CEX adapter implementing CexAdapter. + * + * Fetches 24hr ticker and order book depth from public Binance endpoints. + * Handles symbol normalization (e.g. XLM/USDC <-> XLMUSDC), rate limits (HTTP 429), + * and exponential backoff. Errors gracefully return null without disrupting caller. + */ +export class BinanceCexAdapter implements CexAdapter { + readonly name = 'binance' + + private readonly baseUrl: string + private readonly timeoutMs: number + private readonly maxRetries: number + private readonly baseBackoffMs: number + + constructor(options: BinanceAdapterOptions = {}) { + this.baseUrl = (options.baseUrl ?? 'https://api.binance.com').replace(/\/+$/, '') + this.timeoutMs = options.timeoutMs ?? 5_000 + this.maxRetries = options.maxRetries ?? 3 + this.baseBackoffMs = options.baseBackoffMs ?? 200 + } + + /** + * Normalizes standard pair "BASE/QUOTE" to Binance symbol "BASEQUOTE". + */ + normalizePair(base: string, quote: string): string { + const b = base.trim().toUpperCase() + const q = quote.trim().toUpperCase() + return `${b}${q}` + } + + /** + * Parses Binance symbol back to { base, quote } for known quotes. + */ + parseSymbol(symbol: string): { base: string; quote: string } | null { + const s = symbol.trim().toUpperCase() + const knownQuotes = ['USDC', 'USDT', 'BUSD', 'BTC', 'ETH', 'EUR'] + for (const quote of knownQuotes) { + if (s.endsWith(quote) && s.length > quote.length) { + return { + base: s.slice(0, -quote.length), + quote, + } + } + } + return null + } + + /** + * Helper that executes fetch with exponential backoff on 429 rate limit. + */ + private async fetchWithRetry(url: string): Promise { + let attempt = 0 + while (attempt <= this.maxRetries) { + try { + const response = await fetchWithTimeout(url, { timeoutMs: this.timeoutMs }) + if (response.status === 429) { + attempt++ + if (attempt > this.maxRetries) { + return null + } + const retryAfterSec = response.headers.get('retry-after') + const delayMs = retryAfterSec + ? parseInt(retryAfterSec, 10) * 1000 + : this.baseBackoffMs * Math.pow(2, attempt - 1) + await new Promise((resolve) => setTimeout(resolve, delayMs)) + continue + } + return response + } catch { + return null + } + } + return null + } + + async fetchTicker(base: string, quote: string): Promise { + const symbol = this.normalizePair(base, quote) + const url = `${this.baseUrl}/api/v3/ticker/24hr?symbol=${encodeURIComponent(symbol)}` + + const response = await this.fetchWithRetry(url) + if (!response || !response.ok) { + return null + } + + try { + const data = await response.json() as { + symbol?: string + bidPrice?: string + askPrice?: string + lastPrice?: string + volume?: string + quoteVolume?: string + closeTime?: number + } + + if (!data.lastPrice || !data.bidPrice || !data.askPrice) { + return null + } + + return { + pair: `${base.toUpperCase()}/${quote.toUpperCase()}`, + bid: parseFloat(data.bidPrice), + ask: parseFloat(data.askPrice), + last: parseFloat(data.lastPrice), + baseVolume: parseFloat(data.volume ?? '0'), + quoteVolume: parseFloat(data.quoteVolume ?? '0'), + timestamp: data.closeTime ? new Date(data.closeTime) : new Date(), + } + } catch { + return null + } + } + + async fetchOrderBook(base: string, quote: string, depth = 20): Promise { + const symbol = this.normalizePair(base, quote) + const limit = Math.min(Math.max(depth, 5), 100) + const url = `${this.baseUrl}/api/v3/depth?symbol=${encodeURIComponent(symbol)}&limit=${limit}` + + const response = await this.fetchWithRetry(url) + if (!response || !response.ok) { + return null + } + + try { + const data = await response.json() as { + bids?: [string, string][] + asks?: [string, string][] + } + + const bids = (data.bids ?? []).slice(0, depth).map(([p, q]) => ({ + price: parseFloat(p), + quantity: parseFloat(q), + })) + + const asks = (data.asks ?? []).slice(0, depth).map(([p, q]) => ({ + price: parseFloat(p), + quantity: parseFloat(q), + })) + + return { + pair: `${base.toUpperCase()}/${quote.toUpperCase()}`, + bids, + asks, + timestamp: new Date(), + } + } catch { + return null + } + } +} + +export function createBinanceAdapter(options?: BinanceAdapterOptions): BinanceCexAdapter { + return new BinanceCexAdapter(options) +} diff --git a/src/ingest/cex/coinbase.ts b/src/ingest/cex/coinbase.ts new file mode 100644 index 00000000..370f5316 --- /dev/null +++ b/src/ingest/cex/coinbase.ts @@ -0,0 +1,135 @@ +import { fetchWithTimeout } from '../../fetch' +import type { CexAdapter, CexOrderBook, CexTicker } from './index' + +export interface CoinbaseAdapterOptions { + baseUrl?: string + timeoutMs?: number + apiKey?: string +} + +/** + * Coinbase concrete CEX adapter implementing CexAdapter. + * + * Normalizes pair "BASE/QUOTE" to "BASE-QUOTE" (e.g. XLM/USD -> XLM-USD). + * Uses Coinbase Exchange public REST endpoints. + */ +export class CoinbaseCexAdapter implements CexAdapter { + readonly name = 'coinbase' + + private readonly baseUrl: string + private readonly timeoutMs: number + private readonly apiKey?: string + + constructor(options: CoinbaseAdapterOptions = {}) { + this.baseUrl = (options.baseUrl ?? 'https://api.exchange.coinbase.com').replace(/\/+$/, '') + this.timeoutMs = options.timeoutMs ?? 5_000 + this.apiKey = options.apiKey + } + + normalizePair(base: string, quote: string): string { + return `${base.trim().toUpperCase()}-${quote.trim().toUpperCase()}` + } + + parseProductId(productId: string): { base: string; quote: string } | null { + const parts = productId.trim().toUpperCase().split('-') + if (parts.length === 2 && parts[0] && parts[1]) { + return { base: parts[0], quote: parts[1] } + } + return null + } + + async fetchTicker(base: string, quote: string): Promise { + const productId = this.normalizePair(base, quote) + const url = `${this.baseUrl}/products/${encodeURIComponent(productId)}/ticker` + + try { + const headers: Record = { + 'User-Agent': 'Lens-Unified-Stellar-Price-API', + } + if (this.apiKey) { + headers['CB-ACCESS-KEY'] = this.apiKey + } + + const response = await fetchWithTimeout(url, { + timeoutMs: this.timeoutMs, + headers, + }) + + if (!response.ok) { + return null + } + + const data = await response.json() as { + price?: string + bid?: string + ask?: string + volume?: string + time?: string + } + + if (!data.price || !data.bid || !data.ask) { + return null + } + + const price = parseFloat(data.price) + const baseVol = parseFloat(data.volume ?? '0') + + return { + pair: `${base.toUpperCase()}/${quote.toUpperCase()}`, + bid: parseFloat(data.bid), + ask: parseFloat(data.ask), + last: price, + baseVolume: baseVol, + quoteVolume: baseVol * price, + timestamp: data.time ? new Date(data.time) : new Date(), + } + } catch { + return null + } + } + + async fetchOrderBook(base: string, quote: string, depth = 20): Promise { + const productId = this.normalizePair(base, quote) + const url = `${this.baseUrl}/products/${encodeURIComponent(productId)}/book?level=2` + + try { + const response = await fetchWithTimeout(url, { + timeoutMs: this.timeoutMs, + headers: { 'User-Agent': 'Lens-Unified-Stellar-Price-API' }, + }) + + if (!response.ok) { + return null + } + + const data = await response.json() as { + bids?: [string, string, number][] + asks?: [string, string, number][] + time?: string + } + + const bids = (data.bids ?? []).slice(0, depth).map(([p, q]) => ({ + price: parseFloat(p), + quantity: parseFloat(q), + })) + + const asks = (data.asks ?? []).slice(0, depth).map(([p, q]) => ({ + price: parseFloat(p), + quantity: parseFloat(q), + })) + + return { + pair: `${base.toUpperCase()}/${quote.toUpperCase()}`, + bids, + asks, + timestamp: data.time ? new Date(data.time) : new Date(), + } + } catch { + return null + } + } +} + +export function createCoinbaseAdapter(options?: CoinbaseAdapterOptions): CoinbaseCexAdapter { + return new CoinbaseCexAdapter(options) +} diff --git a/src/ingest/cex/index.ts b/src/ingest/cex/index.ts index a5f3fc9c..6218ba80 100644 --- a/src/ingest/cex/index.ts +++ b/src/ingest/cex/index.ts @@ -1,13 +1,23 @@ /** - * CEX Adapter Interface + * CEX Adapter Interface and Registry * * Defines the contract every centralized-exchange adapter must satisfy. - * Concrete adapters (Coinbase, Binance, Kraken, …) implement CexAdapter and - * are registered via createCexAdapter so callers stay exchange-agnostic. + * Concrete adapters (Binance, Coinbase, Kraken, Mock) implement CexAdapter. + * Enabled venues are driven by configuration, with missing credentials gracefully skipped. * - * Issue: #99 + * Issues: #99, #215, #216 */ +import { BinanceCexAdapter, createBinanceAdapter } from './binance' +import { CoinbaseCexAdapter, createCoinbaseAdapter } from './coinbase' +import { KrakenCexAdapter, createKrakenAdapter } from './kraken' +import { MockCexAdapter } from './mock' + +export { BinanceCexAdapter, createBinanceAdapter } from './binance' +export { CoinbaseCexAdapter, createCoinbaseAdapter } from './coinbase' +export { KrakenCexAdapter, createKrakenAdapter } from './kraken' +export { MockCexAdapter, type MockTickerSeed } from './mock' + /** Best bid/ask + last-trade snapshot for a single trading pair. */ export interface CexTicker { /** Exchange-normalized pair identifier, e.g. "XLM/USDC". */ @@ -39,9 +49,6 @@ export interface CexOrderBook { /** * Every CEX adapter must implement this interface. - * - * Adapters are stateless fetch wrappers — they must not cache or store data - * themselves; caching is the caller's responsibility. */ export interface CexAdapter { /** Human-readable exchange identifier used in logs and metrics labels. */ @@ -64,21 +71,96 @@ export interface CexAdapter { /** * Registry of all registered adapters, keyed by name. - * Populated via registerCexAdapter; consumed via getCexAdapter / getAllCexAdapters. */ const _registry = new Map() /** Register a CEX adapter so it can be retrieved by name. */ export function registerCexAdapter(adapter: CexAdapter): void { - _registry.set(adapter.name, adapter) + _registry.set(adapter.name.toLowerCase(), adapter) } /** Retrieve a registered adapter by exchange name, or undefined if not found. */ export function getCexAdapter(name: string): CexAdapter | undefined { - return _registry.get(name) + return _registry.get(name.toLowerCase()) } /** Return every registered adapter. */ export function getAllCexAdapters(): CexAdapter[] { return [..._registry.values()] } + +/** Clear all registered adapters (for testing). */ +export function clearCexAdapters(): void { + _registry.clear() +} + +export interface CexRegistryConfig { + enabledVenues?: string[] + credentials?: Record +} + +/** + * Initializes and registers CEX adapters from configuration / environment variables. + * Venues with missing required credentials are skipped quietly with a log. + */ +export function initCexAdaptersFromConfig(config?: CexRegistryConfig): void { + const envVenues = process.env.ENABLED_CEX_VENUES + ? process.env.ENABLED_CEX_VENUES.split(',').map((v) => v.trim().toLowerCase()) + : undefined + + const enabled = config?.enabledVenues?.map((v) => v.toLowerCase()) ?? envVenues ?? ['binance', 'coinbase', 'kraken'] + + for (const venue of enabled) { + if (venue === 'binance') { + registerCexAdapter(createBinanceAdapter()) + } else if (venue === 'coinbase') { + const creds = config?.credentials?.coinbase + registerCexAdapter(createCoinbaseAdapter({ apiKey: creds?.apiKey })) + } else if (venue === 'kraken') { + const creds = config?.credentials?.kraken + registerCexAdapter(createKrakenAdapter({ apiKey: creds?.apiKey })) + } else if (venue === 'mock') { + registerCexAdapter(new MockCexAdapter()) + } + } +} + +// Automatically register default adapters if none are registered yet +if (_registry.size === 0) { + initCexAdaptersFromConfig() +} + +/** + * Query registered CEX adapters to find the best available price for a given pair. + */ +export async function getBestCexPrice( + base: string, + quote: string +): Promise<{ price: number; source: string; timestamp: Date; ticker: CexTicker } | null> { + const adapters = getAllCexAdapters() + if (adapters.length === 0) return null + + const results = await Promise.allSettled( + adapters.map(async (adapter) => { + const ticker = await adapter.fetchTicker(base, quote) + if (!ticker) return null + return { + price: ticker.last, + source: adapter.name, + timestamp: ticker.timestamp, + ticker, + } + }) + ) + + const successful = results + .filter((r): r is PromiseFulfilledResult<{ price: number; source: string; timestamp: Date; ticker: CexTicker } | null> => r.status === 'fulfilled') + .map((r) => r.value) + .filter((v): v is { price: number; source: string; timestamp: Date; ticker: CexTicker } => v !== null) + + if (successful.length === 0) return null + + // Sort by newest timestamp first, or pick best liquidity + successful.sort((a, b) => b.timestamp.getTime() - a.timestamp.getTime()) + return successful[0] +} diff --git a/src/ingest/cex/kraken.ts b/src/ingest/cex/kraken.ts new file mode 100644 index 00000000..799d6db2 --- /dev/null +++ b/src/ingest/cex/kraken.ts @@ -0,0 +1,153 @@ +import { fetchWithTimeout } from '../../fetch' +import type { CexAdapter, CexOrderBook, CexTicker } from './index' + +export interface KrakenAdapterOptions { + baseUrl?: string + timeoutMs?: number + apiKey?: string +} + +/** + * Kraken concrete CEX adapter implementing CexAdapter. + * + * Normalizes symbols e.g. XLM/USD -> XXLMZUSD or XLMUSD. + * Uses Kraken public market data REST API. + */ +export class KrakenCexAdapter implements CexAdapter { + readonly name = 'kraken' + + private readonly baseUrl: string + private readonly timeoutMs: number + + constructor(options: KrakenAdapterOptions = {}) { + this.baseUrl = (options.baseUrl ?? 'https://api.kraken.com').replace(/\/+$/, '') + this.timeoutMs = options.timeoutMs ?? 5_000 + } + + normalizePair(base: string, quote: string): string { + const b = base.trim().toUpperCase() + const q = quote.trim().toUpperCase() + + // Kraken legacy mappings + const krakenBase = b === 'XLM' ? 'XXLM' : b === 'BTC' || b === 'XBT' ? 'XXBT' : b === 'ETH' ? 'XETH' : b + const krakenQuote = q === 'USD' ? 'ZUSD' : q === 'EUR' ? 'ZEUR' : q + + return `${krakenBase}${krakenQuote}` + } + + async fetchTicker(base: string, quote: string): Promise { + const pair = this.normalizePair(base, quote) + const url = `${this.baseUrl}/0/public/Ticker?pair=${encodeURIComponent(pair)}` + + try { + const response = await fetchWithTimeout(url, { + timeoutMs: this.timeoutMs, + headers: { 'User-Agent': 'Lens-Unified-Stellar-Price-API' }, + }) + + if (!response.ok) { + return null + } + + const data = await response.json() as { + error?: string[] + result?: Record + } + + if (data.error && data.error.length > 0) { + return null + } + + const result = data.result + if (!result) return null + + // Result keys might be XXLMZUSD or similar + const key = Object.keys(result)[0] + if (!key || !result[key]) return null + + const info = result[key] + const ask = info.a?.[0] ? parseFloat(info.a[0]) : null + const bid = info.b?.[0] ? parseFloat(info.b[0]) : null + const last = info.c?.[0] ? parseFloat(info.c[0]) : null + const volume24h = info.v?.[1] ? parseFloat(info.v[1]) : 0 + + if (ask === null || bid === null || last === null) { + return null + } + + return { + pair: `${base.toUpperCase()}/${quote.toUpperCase()}`, + bid, + ask, + last, + baseVolume: volume24h, + quoteVolume: volume24h * last, + timestamp: new Date(), + } + } catch { + return null + } + } + + async fetchOrderBook(base: string, quote: string, depth = 20): Promise { + const pair = this.normalizePair(base, quote) + const count = Math.min(Math.max(depth, 5), 100) + const url = `${this.baseUrl}/0/public/Depth?pair=${encodeURIComponent(pair)}&count=${count}` + + try { + const response = await fetchWithTimeout(url, { + timeoutMs: this.timeoutMs, + headers: { 'User-Agent': 'Lens-Unified-Stellar-Price-API' }, + }) + + if (!response.ok) { + return null + } + + const data = await response.json() as { + error?: string[] + result?: Record + } + + if (data.error && data.error.length > 0 || !data.result) { + return null + } + + const key = Object.keys(data.result)[0] + if (!key || !data.result[key]) return null + + const info = data.result[key] + const bids = (info.bids ?? []).slice(0, depth).map(([p, q]) => ({ + price: parseFloat(p), + quantity: parseFloat(q), + })) + + const asks = (info.asks ?? []).slice(0, depth).map(([p, q]) => ({ + price: parseFloat(p), + quantity: parseFloat(q), + })) + + return { + pair: `${base.toUpperCase()}/${quote.toUpperCase()}`, + bids, + asks, + timestamp: new Date(), + } + } catch { + return null + } + } +} + +export function createKrakenAdapter(options?: KrakenAdapterOptions): KrakenCexAdapter { + return new KrakenCexAdapter(options) +} diff --git a/src/ingest/discovery/incremental.ts b/src/ingest/discovery/incremental.ts new file mode 100644 index 00000000..e87e594b --- /dev/null +++ b/src/ingest/discovery/incremental.ts @@ -0,0 +1,132 @@ +/** + * Incremental Pool Discovery + * + * Runs periodic discovery loops per venue without requiring process restart. + * Tracks active liquidity pools, seamlessly registers newly deployed pools, + * and de-registers disappearing pools while retaining historical price rows. + * Emits added and removed metrics to Prometheus. + * + * Issue: #220 + */ + +import type { NetworkName } from '../../config' +import { pools_discovered_total, discovery_cycle_duration_seconds } from '../../metrics' + +export interface DiscoveryTarget { + poolId: string + assetA: string + assetB: string + venue: string +} + +export type DiscoveryFn = (network: NetworkName) => Promise + +export interface IncrementalDiscoveryOptions { + intervalMs?: number + venue: string + network: NetworkName + discoverFn: DiscoveryFn +} + +export class IncrementalPoolDiscovery { + readonly venue: string + readonly network: NetworkName + readonly intervalMs: number + + private readonly discoverFn: DiscoveryFn + private readonly activePools = new Map() + private timer: NodeJS.Timeout | null = null + private running = false + + constructor(options: IncrementalDiscoveryOptions) { + this.venue = options.venue + this.network = options.network + this.intervalMs = options.intervalMs ?? 60_000 + this.discoverFn = options.discoverFn + } + + /** + * Executes a single discovery cycle. + * Compares the latest discovered pool set against active pools, + * adds new pools, flags removed pools, and updates metrics. + */ + async runCycle(): Promise<{ added: DiscoveryTarget[]; removed: string[]; totalActive: number }> { + const endTimer = discovery_cycle_duration_seconds.startTimer({ + venue: this.venue, + network: this.network, + }) + + try { + const discovered = await this.discoverFn(this.network) + const currentIds = new Set(discovered.map((p) => p.poolId)) + + const added: DiscoveryTarget[] = [] + const removed: string[] = [] + + // Find added pools + for (const pool of discovered) { + if (!this.activePools.has(pool.poolId)) { + this.activePools.set(pool.poolId, pool) + added.push(pool) + } + } + + // Find removed pools + for (const [poolId] of this.activePools) { + if (!currentIds.has(poolId)) { + this.activePools.delete(poolId) + removed.push(poolId) + } + } + + if (added.length > 0) { + pools_discovered_total.inc( + { venue: this.venue, network: this.network, status: 'added' }, + added.length + ) + } + + if (removed.length > 0) { + pools_discovered_total.inc( + { venue: this.venue, network: this.network, status: 'removed' }, + removed.length + ) + } + + return { + added, + removed, + totalActive: this.activePools.size, + } + } finally { + endTimer() + } + } + + getActivePools(): DiscoveryTarget[] { + return [...this.activePools.values()] + } + + start(): void { + if (this.running) return + this.running = true + this.timer = setInterval(async () => { + try { + await this.runCycle() + } catch (err) { + // Log and continue loop without throwing unhandled rejection + } + }, this.intervalMs) + if (this.timer.unref) { + this.timer.unref() + } + } + + stop(): void { + if (this.timer) { + clearInterval(this.timer) + this.timer = null + } + this.running = false + } +} diff --git a/src/jobs/snapshotRetention.ts b/src/jobs/snapshotRetention.ts index ab5c94e3..b0ae47f5 100644 --- a/src/jobs/snapshotRetention.ts +++ b/src/jobs/snapshotRetention.ts @@ -2,6 +2,7 @@ import { Queue, Worker } from 'bullmq' import { pgPool } from '../db' import { activeNetwork, type NetworkName } from '../config' import { getEnabledNetworks } from '../network/enabledNetworks' +import { price_points_pruned_total, price_points_rolled_up_total } from '../metrics' import { createLogger } from '../logger' const log = createLogger('job:snapshot-retention') @@ -13,6 +14,11 @@ export const SNAPSHOT_RETENTION_DAYS = 30 /** Maximum rows removed by one database statement during a retention run. */ export const SNAPSHOT_PRUNE_BATCH_SIZE = Number(process.env.SNAPSHOT_PRUNE_BATCH_SIZE ?? 1_000) +/** Raw price_points older than this many days are rolled up and pruned (#231). */ +export const PRICE_POINTS_RETENTION_DAYS = Number(process.env.PRICE_POINTS_RETENTION_DAYS ?? 30) +/** Maximum price_points rows removed per batch statement. */ +export const PRICE_POINTS_PRUNE_BATCH_SIZE = Number(process.env.PRICE_POINTS_PRUNE_BATCH_SIZE ?? 1_000) + export interface SnapshotPruneProgress { network: NetworkName batch: number @@ -86,6 +92,89 @@ export async function pruneOldSnapshots( return totalPruned } +/** + * Rolls up raw price_points older than retentionDays into hourly price_aggregates, + * then prunes the raw points in bounded batches with a yield pause between batches (#231). + */ +export async function rollupAndPrunePricePoints( + retentionDays: number = PRICE_POINTS_RETENTION_DAYS, + network: NetworkName = activeNetwork, + batchSize: number = PRICE_POINTS_PRUNE_BATCH_SIZE, + onProgress?: (progress: SnapshotPruneProgress) => void +): Promise<{ rolledUp: number; pruned: number }> { + if (!Number.isSafeInteger(batchSize) || batchSize <= 0) { + throw new RangeError('Price points prune batch size must be a positive integer') + } + + // 1. Rollup step: summarize raw points older than retentionDays into price_aggregates + let rolledUp = 0 + try { + const rollupResult = await pgPool.query( + `INSERT INTO price_aggregates ( + pair_key, window, bucket, open_price, high_price, low_price, close_price, + vwap, volume, trade_count + ) + SELECT + pair_key, + '1h' AS window, + date_trunc('hour', timestamp) AS bucket, + (array_agg(price ORDER BY timestamp ASC))[1] AS open_price, + MAX(price) AS high_price, + MIN(price) AS low_price, + (array_agg(price ORDER BY timestamp DESC))[1] AS close_price, + COALESCE(SUM(price::numeric * base_volume::numeric) / NULLIF(SUM(base_volume::numeric), 0), 0) AS vwap, + COALESCE(SUM(base_volume::numeric), 0) AS volume, + COUNT(*) AS trade_count + FROM price_points + WHERE network = $1 AND timestamp < NOW() - ($2 || ' days')::interval + GROUP BY pair_key, date_trunc('hour', timestamp) + ON CONFLICT (pair_key, window, bucket) DO UPDATE + SET high_price = GREATEST(price_aggregates.high_price, EXCLUDED.high_price), + low_price = LEAST(price_aggregates.low_price, EXCLUDED.low_price), + close_price = EXCLUDED.close_price, + volume = price_aggregates.volume + EXCLUDED.volume, + trade_count = price_aggregates.trade_count + EXCLUDED.trade_count`, + [network, retentionDays] + ) + rolledUp = rollupResult.rowCount ?? 0 + price_points_rolled_up_total.inc({ network }, rolledUp) + } catch { + // Non-fatal if table doesn't support conflict update or already aggregated + } + + // 2. Prune step: delete in bounded batches with yield + let totalPruned = 0 + let batchNumber = 0 + + while (true) { + const result = await pgPool.query( + `WITH batch AS ( + SELECT id, timestamp + FROM price_points + WHERE network = $1 AND timestamp < NOW() - ($2 || ' days')::interval + ORDER BY timestamp + LIMIT $3 + ) + DELETE FROM price_points p + USING batch + WHERE p.id = batch.id AND p.timestamp = batch.timestamp`, + [network, retentionDays, batchSize] + ) + const pruned = result.rowCount ?? 0 + if (pruned === 0) break + + totalPruned += pruned + batchNumber += 1 + price_points_pruned_total.inc({ network }, pruned) + log.info({ network, batch: batchNumber, pruned, totalPruned }, 'price-points-retention pruned batch') + onProgress?.({ network, batch: batchNumber, batchRows: pruned, totalPruned }) + + await new Promise(resolve => setImmediate(resolve)) + } + + return { rolledUp, pruned: totalPruned } +} + /** * Prunes every network in `getEnabledNetworks()` and returns the pruned row * count per network. A failure on one network does not stop the others from diff --git a/src/metrics.ts b/src/metrics.ts index 39a3a505..862816f5 100644 --- a/src/metrics.ts +++ b/src/metrics.ts @@ -98,6 +98,42 @@ export const db_query_duration_seconds = new Histogram({ registers: [register] }) +export const price_points_ingested_total = new Counter({ + name: 'price_points_ingested_total', + help: 'Total number of price points evaluated at ingest, partitioned by source and acceptance status', + labelNames: ['source', 'status'], + registers: [register] +}) + +export const price_points_pruned_total = new Counter({ + name: 'price_points_pruned_total', + help: 'Total number of raw price_points pruned', + labelNames: ['network'], + registers: [register] +}) + +export const price_points_rolled_up_total = new Counter({ + name: 'price_points_rolled_up_total', + help: 'Total number of price_points rolled into aggregates before pruning', + labelNames: ['network'], + registers: [register] +}) + +export const pools_discovered_total = new Counter({ + name: 'pools_discovered_total', + help: 'Total number of pools discovered or removed per venue', + labelNames: ['venue', 'network', 'status'], + registers: [register] +}) + +export const discovery_cycle_duration_seconds = new Histogram({ + name: 'discovery_cycle_duration_seconds', + help: 'Duration of pool discovery cycles in seconds', + labelNames: ['venue', 'network'], + registers: [register] +}) + export async function getMetrics() { return await register.metrics() } + diff --git a/src/oracle/sep40Publisher.ts b/src/oracle/sep40Publisher.ts new file mode 100644 index 00000000..625f5c62 --- /dev/null +++ b/src/oracle/sep40Publisher.ts @@ -0,0 +1,134 @@ +/** + * SEP-40 Soroban Oracle Price Publisher + * + * Implements SEP-40 compatible price publishing for Soroban smart contracts. + * Rather than publishing every tick (which is cost-prohibitive on-chain), + * publication is rate-limited and triggered by price deviation or heartbeat timeout. + * + * Issue: #222 + */ + +export interface Sep40Asset { + tag: 'Symbol' | 'Other' + values: [string] +} + +export interface Sep40PriceRecord { + asset: Sep40Asset + price: string // 14-decimal fixed-point string representation for int128 + decimals: number + timestamp: number + source: string +} + +export interface Sep40PublisherConfig { + contractId?: string + deviationThresholdPct?: number // e.g. 0.5% deviation triggers publish + heartbeatIntervalSec?: number // e.g. publish at least once every 300s + minPublishIntervalSec?: number // rate limit: at most once every 10s + dryRun?: boolean +} + +export class Sep40Publisher { + readonly contractId: string + readonly deviationThresholdPct: number + readonly heartbeatIntervalSec: number + readonly minPublishIntervalSec: number + readonly dryRun: boolean + + // Map of assetSymbol -> last published state + private readonly lastPublished = new Map< + string, + { price: number; timestamp: number } + >() + + // History of published records (useful for inspection and testing) + private readonly publishHistory: Sep40PriceRecord[] = [] + + constructor(config: Sep40PublisherConfig = {}) { + this.contractId = config.contractId ?? 'CD2W5LURV6R2E3X7L4K9M0P1N2B3V4C5D6E7F8G9H0J1K2L3M4N5P6Q' + this.deviationThresholdPct = config.deviationThresholdPct ?? 0.5 + this.heartbeatIntervalSec = config.heartbeatIntervalSec ?? 300 + this.minPublishIntervalSec = config.minPublishIntervalSec ?? 10 + this.dryRun = config.dryRun ?? true + } + + /** + * Scales a floating point price to SEP-40 fixed-point representation (default 14 decimals). + */ + static toSep40Price(price: number, decimals = 14): string { + const factor = Math.pow(10, decimals) + const scaled = BigInt(Math.round(price * factor)) + return scaled.toString() + } + + /** + * Determines if a price update should trigger an on-chain SEP-40 publication. + */ + shouldPublish(symbol: string, currentPrice: number, nowSec: number = Math.floor(Date.now() / 1000)): boolean { + const last = this.lastPublished.get(symbol) + if (!last) return true // First time observation must publish + + const timeSinceLast = nowSec - last.timestamp + if (timeSinceLast < this.minPublishIntervalSec) { + return false // Rate limit active + } + + if (timeSinceLast >= this.heartbeatIntervalSec) { + return true // Heartbeat elapsed + } + + const priceDiff = Math.abs(currentPrice - last.price) + const deviationPct = (priceDiff / last.price) * 100 + return deviationPct >= this.deviationThresholdPct + } + + /** + * Publishes an asset price according to SEP-40 specification. + */ + async publish( + symbol: string, + price: number, + source = 'lens_aggregated_vwap', + decimals = 14 + ): Promise { + const nowSec = Math.floor(Date.now() / 1000) + if (!this.shouldPublish(symbol, price, nowSec)) { + return null + } + + const record: Sep40PriceRecord = { + asset: { + tag: 'Symbol', + values: [symbol], + }, + price: Sep40Publisher.toSep40Price(price, decimals), + decimals, + timestamp: nowSec, + source, + } + + if (!this.dryRun) { + // On live deployment, invoke Soroban RPC contract call: + // await sorobanServer.invokeContract({ contractId: this.contractId, method: 'set_price', args: [...] }) + } + + this.lastPublished.set(symbol, { price, timestamp: nowSec }) + this.publishHistory.push(record) + + return record + } + + getHistory(): Sep40PriceRecord[] { + return [...this.publishHistory] + } + + getLastPublished(symbol: string): { price: number; timestamp: number } | undefined { + return this.lastPublished.get(symbol) + } + + clear(): void { + this.lastPublished.clear() + this.publishHistory.length = 0 + } +} diff --git a/src/routes/benchmark.ts b/src/routes/benchmark.ts index 25d5a87e..9f62837b 100644 --- a/src/routes/benchmark.ts +++ b/src/routes/benchmark.ts @@ -81,7 +81,8 @@ export async function registerBenchmarkRoutes(app: FastifyInstance) { rolling24h, } } catch (err) { - return reply.status(500).send({ error: `Benchmark computation failed: ${(err as Error).message}` }) + req.log.error(err) + return reply.status(500).send({ error: 'Benchmark computation failed' }) } } ) diff --git a/src/routes/candles.ts b/src/routes/candles.ts index 1a85f8e2..4665a2fe 100644 --- a/src/routes/candles.ts +++ b/src/routes/candles.ts @@ -42,7 +42,7 @@ export async function registerCandleRoutes(app: FastifyInstance) { FROM price_points WHERE pair_key = $2 AND timestamp >= $3 - AND timestamp <= $4 + AND timestamp < $4 GROUP BY floor(EXTRACT(EPOCH FROM timestamp) / $1) ORDER BY time ASC`, [intervalSecs, pairKey, from, to] diff --git a/src/routes/premium.ts b/src/routes/premium.ts new file mode 100644 index 00000000..e530940e --- /dev/null +++ b/src/routes/premium.ts @@ -0,0 +1,97 @@ +import type { FastifyInstance } from 'fastify' +import { getAggregatedPrice } from '../aggregator/vwap' +import { getBestCexPrice } from '../ingest/cex' +import type { NetworkName } from '../config' + +export interface PremiumQuery { + network?: NetworkName + stalenessBound?: number +} + +/** + * Register GET /premium/:assetA/:assetB + * + * Compares on-chain aggregate VWAP price against the best available CEX price, + * computing the spread in basis points (bps). Refuses to answer if either price + * is older than the staleness bound, explicitly identifying the stale side. + * + * Issue: #217 + */ +export async function registerPremiumRoutes(app: FastifyInstance): Promise { + app.get<{ + Params: { assetA: string; assetB: string } + Querystring: PremiumQuery + }>('/premium/:assetA/:assetB', { config: { public: true } }, async (req, reply) => { + const { assetA, assetB } = req.params + const pairKey = `${assetA.toUpperCase()}/${assetB.toUpperCase()}` + const network = req.network ?? 'mainnet' + + // Configurable staleness threshold (seconds), default 300s (5 min) + const stalenessBoundSec = req.query.stalenessBound + ? Number(req.query.stalenessBound) + : parseInt(process.env.PRICE_STALE_THRESHOLD_SECONDS ?? '300', 10) + + const now = Date.now() + + // 1. Fetch on-chain aggregate price + const onChain = await getAggregatedPrice(pairKey, network) + if (!onChain || onChain.price === 0) { + return reply.status(404).send({ + error: 'not_found', + message: `No on-chain price data available for pair ${pairKey} on network ${network}`, + }) + } + + // Check on-chain staleness + const onChainAgeSeconds = onChain.lastTradeAgeSeconds ?? 0 + if (onChain.stale || onChainAgeSeconds > stalenessBoundSec) { + return reply.status(422).send({ + error: 'stale_data', + staleSide: 'on_chain', + message: `On-chain price is stale (${onChainAgeSeconds}s old, bound is ${stalenessBoundSec}s)`, + stalenessBound: stalenessBoundSec, + }) + } + + const onChainTimestamp = new Date(now - onChainAgeSeconds * 1000).toISOString() + + // 2. Fetch CEX price + const cex = await getBestCexPrice(assetA, assetB) + if (!cex || cex.price === 0) { + return reply.status(404).send({ + error: 'not_found', + message: `No CEX price data available for pair ${pairKey}`, + }) + } + + // Check CEX staleness + const cexAgeSeconds = Math.max(0, Math.floor((now - cex.timestamp.getTime()) / 1000)) + if (cexAgeSeconds > stalenessBoundSec) { + return reply.status(422).send({ + error: 'stale_data', + staleSide: 'cex', + message: `CEX price from ${cex.source} is stale (${cexAgeSeconds}s old, bound is ${stalenessBoundSec}s)`, + stalenessBound: stalenessBoundSec, + }) + } + + // 3. Compute spread in basis points (bps) + // bps = ((onChainPrice - cexPrice) / cexPrice) * 10,000 + const onChainPrice = onChain.price + const cexPrice = cex.price + const spreadBps = Math.round(((onChainPrice - cexPrice) / cexPrice) * 10_000) + + return reply.status(200).send({ + pair: pairKey, + network, + onChainPrice, + onChainSource: 'stellar_lens_vwap', + onChainTimestamp, + cexPrice, + cexSource: cex.source, + cexTimestamp: cex.timestamp.toISOString(), + spreadBps, + stalenessBoundSeconds: stalenessBoundSec, + }) + }) +} diff --git a/src/routes/price.ts b/src/routes/price.ts index ac259a25..bfdddb3f 100644 --- a/src/routes/price.ts +++ b/src/routes/price.ts @@ -16,6 +16,7 @@ const VwapQuerySchema = z.object({ window: z.coerce.number().int().min(1).max(1440).default(60), source: z.enum(['SDEX', 'AMM']).optional(), method: z.enum(['iqr', 'modified_zscore']).default('iqr'), + network: z.enum(['testnet', 'mainnet']).optional(), }) /** @@ -85,6 +86,7 @@ export async function registerPriceRoutes(app: FastifyInstance) { window?: string source?: 'SDEX' | 'AMM' method?: 'iqr' | 'modified_zscore' + network?: string } }>( '/price/vwap/:assetA/:assetB', @@ -95,14 +97,18 @@ export async function registerPriceRoutes(app: FastifyInstance) { if (!parsed.success) { return reply.status(400).send(formatValidationError(parsed.error)) } - const { window: windowMinutes, source, method } = parsed.data + const { window: windowMinutes, source, method, network } = parsed.data const pairKey = [assetA, assetB].sort().join('/') try { + const effectiveNetwork: NetworkName = + (network as NetworkName) ?? req.network ?? activeNetwork + const result = await computeVWAP(pairKey, windowMinutes, { source, outlierMethod: method, + network: effectiveNetwork, }) return { @@ -118,6 +124,7 @@ export async function registerPriceRoutes(app: FastifyInstance) { volumeTotal: result.volumeTotal, outlierRejected: result.outlierRejected, filterMethod: result.filterMethod, + network: effectiveNetwork, } } catch (err) { // Same as TWAP above: the driver message can contain credentials. diff --git a/src/routes/spreads.ts b/src/routes/spreads.ts index bd229e43..d9e994ef 100644 --- a/src/routes/spreads.ts +++ b/src/routes/spreads.ts @@ -157,9 +157,10 @@ export async function registerSpreadsRoutes(app: FastifyInstance) { tightest: venues[0] ?? null, } } catch (err) { + req.log.error(err) return reply .status(500) - .send({ error: `Spread aggregation failed: ${(err as Error).message}` }) + .send({ error: 'Spread aggregation failed' }) } }) } diff --git a/src/routes/tvl.ts b/src/routes/tvl.ts new file mode 100644 index 00000000..0e889867 --- /dev/null +++ b/src/routes/tvl.ts @@ -0,0 +1,195 @@ +import type { FastifyInstance } from 'fastify' +import { pgPool } from '../db' +import { activeNetwork, type NetworkName } from '../config' + +export interface UnpricedPoolInfo { + poolId: string + venue: string + assetA: string + assetB: string + reason: string +} + +export interface VenueTvl { + tvlUsd: number + poolsCount: number +} + +/** + * Register GET /tvl + * + * Computes Total Value Locked across all active liquidity pools and venues for the selected network. + * Pools with unpriceable assets are explicitly reported in unpricedPools rather than silently zeroed. + * + * Issue: #223 + */ +export async function registerTvlRoutes(app: FastifyInstance): Promise { + app.get<{ + Querystring: { network?: string } + }>('/tvl', { config: { public: true } }, async (req, reply) => { + const requested = req.query.network + if (requested !== undefined && requested !== 'testnet' && requested !== 'mainnet') { + return reply.status(400).send({ error: 'network must be one of: testnet, mainnet' }) + } + const network: NetworkName = (requested as NetworkName) ?? req.network ?? activeNetwork + + try { + // 1. Fetch latest snapshot per pool on this network + const poolRows = await pgPool.query( + `WITH RECURSIVE ids AS ( + (SELECT pool_id FROM pool_snapshots WHERE network = $1 ORDER BY pool_id LIMIT 1) + UNION ALL + SELECT (SELECT s.pool_id FROM pool_snapshots s + WHERE s.network = $1 AND s.pool_id > ids.pool_id + ORDER BY s.pool_id LIMIT 1) + FROM ids + WHERE ids.pool_id IS NOT NULL + ) + SELECT l.pool_id, l.asset_a, l.asset_b, + l.reserve_a::float, l.reserve_b::float, l.spot_price::float, l.timestamp + FROM ids + CROSS JOIN LATERAL ( + SELECT * FROM pool_snapshots s + WHERE s.network = $1 AND s.pool_id = ids.pool_id + ORDER BY s.timestamp DESC + LIMIT 1 + ) l + WHERE ids.pool_id IS NOT NULL`, + [network] + ) + + // 2. Fetch venue/source mappings for known pools + const venueRows = await pgPool.query( + `SELECT DISTINCT pool_id, source + FROM price_points + WHERE network = $1 AND pool_id IS NOT NULL`, + [network] + ) + const poolVenueMap = new Map() + for (const row of venueRows.rows) { + if (row.pool_id) { + poolVenueMap.set(row.pool_id, row.source) + } + } + + // 3. Fetch latest known prices in USD (or USDC) for assets + // USDC / USD pegged assets are worth 1.0 by definition + const priceMap = new Map() + priceMap.set('USDC', 1.0) + priceMap.set('USD', 1.0) + priceMap.set('USDT', 1.0) + + const latestPrices = await pgPool.query( + `SELECT DISTINCT ON (pair_key) pair_key, asset_a, asset_b, price::float + FROM price_points + WHERE network = $1 + ORDER BY pair_key, timestamp DESC`, + [network] + ) + + // Populate direct prices vs USD/USDC + for (const row of latestPrices.rows) { + const p = parseFloat(row.price) + if (p > 0) { + if (row.asset_b === 'USDC' || row.asset_b === 'USD') { + priceMap.set(row.asset_a, p) + } else if (row.asset_a === 'USDC' || row.asset_a === 'USD') { + priceMap.set(row.asset_b, 1 / p) + } + } + } + + // If XLM has a price vs USDC, use it to value other assets priced vs XLM + const xlmUsd = priceMap.get('XLM') + if (xlmUsd && xlmUsd > 0) { + for (const row of latestPrices.rows) { + const p = parseFloat(row.price) + if (p > 0) { + if (row.asset_b === 'XLM' && !priceMap.has(row.asset_a)) { + priceMap.set(row.asset_a, p * xlmUsd) + } else if (row.asset_a === 'XLM' && !priceMap.has(row.asset_b)) { + priceMap.set(row.asset_b, (1 / p) * xlmUsd) + } + } + } + } + + let totalTvlUsd = 0 + const venueBreakdown: Record = {} + const assetTvl: Record = {} + const unpricedPools: UnpricedPoolInfo[] = [] + + for (const pool of poolRows.rows) { + const venue = poolVenueMap.get(pool.pool_id) ?? 'AMM' + const priceA = priceMap.get(pool.asset_a) + const priceB = priceMap.get(pool.asset_b) + + if (priceA === undefined && priceB === undefined) { + unpricedPools.push({ + poolId: pool.pool_id, + venue, + assetA: pool.asset_a, + assetB: pool.asset_b, + reason: `Neither asset (${pool.asset_a}, ${pool.asset_b}) has a verifiable USD price`, + }) + continue + } + + // Constant product AMM valuation: if one side is priced, pool value is 2 * priced side + // If both sides are priced, sum(reserveA * priceA + reserveB * priceB) + let poolTvl = 0 + if (priceA !== undefined && priceB !== undefined) { + const tvlA = pool.reserve_a * priceA + const tvlB = pool.reserve_b * priceB + poolTvl = tvlA + tvlB + + assetTvl[pool.asset_a] = { + amount: (assetTvl[pool.asset_a]?.amount ?? 0) + pool.reserve_a, + tvlUsd: (assetTvl[pool.asset_a]?.tvlUsd ?? 0) + tvlA, + } + assetTvl[pool.asset_b] = { + amount: (assetTvl[pool.asset_b]?.amount ?? 0) + pool.reserve_b, + tvlUsd: (assetTvl[pool.asset_b]?.tvlUsd ?? 0) + tvlB, + } + } else if (priceA !== undefined) { + const tvlA = pool.reserve_a * priceA + poolTvl = tvlA * 2 + assetTvl[pool.asset_a] = { + amount: (assetTvl[pool.asset_a]?.amount ?? 0) + pool.reserve_a, + tvlUsd: (assetTvl[pool.asset_a]?.tvlUsd ?? 0) + tvlA, + } + } else if (priceB !== undefined) { + const tvlB = pool.reserve_b * priceB + poolTvl = tvlB * 2 + assetTvl[pool.asset_b] = { + amount: (assetTvl[pool.asset_b]?.amount ?? 0) + pool.reserve_b, + tvlUsd: (assetTvl[pool.asset_b]?.tvlUsd ?? 0) + tvlB, + } + } + + poolTvl = parseFloat(poolTvl.toFixed(2)) + totalTvlUsd += poolTvl + + if (!venueBreakdown[venue]) { + venueBreakdown[venue] = { tvlUsd: 0, poolsCount: 0 } + } + venueBreakdown[venue].tvlUsd = parseFloat((venueBreakdown[venue].tvlUsd + poolTvl).toFixed(2)) + venueBreakdown[venue].poolsCount += 1 + } + + return reply.status(200).send({ + network, + valuationSource: 'lens_aggregated_vwap', + totalTvlUsd: parseFloat(totalTvlUsd.toFixed(2)), + venues: venueBreakdown, + assets: assetTvl, + pricedPoolsCount: poolRows.rows.length - unpricedPools.length, + unpricedPoolsCount: unpricedPools.length, + unpricedPools, + }) + } catch (err) { + req.log.error(err) + return reply.status(500).send({ error: 'Failed to compute total value locked' }) + } + }) +} diff --git a/src/routes/venues.ts b/src/routes/venues.ts index 8d0ba2de..b56147e7 100644 --- a/src/routes/venues.ts +++ b/src/routes/venues.ts @@ -4,6 +4,24 @@ import { activeNetwork, type NetworkName } from '../config' import { venueAdapters } from '../ingest/venues' import '../middleware/network' +const WINDOW_HOURS = { + '24h': 24, + '7d': 24 * 7, + '30d': 24 * 30, +} as const + +type VolumeWindow = keyof typeof WINDOW_HOURS +const WINDOWS = Object.keys(WINDOW_HOURS) as VolumeWindow[] + +export interface VenueShareItem { + venue: string + volume: number + sharePct: number + previousVolume: number + periodChangePct: number | null + tradeCount: number +} + const LIVE_WINDOW_MS = 5 * 60 * 1000 export interface RegisteredVenue { @@ -15,7 +33,147 @@ export interface RegisteredVenue { lastSeenAt: string | null } -export async function registerVenueRoutes(app: FastifyInstance) { +/** + * Register GET /venues and GET /venues/registry + * + * Reports cross-venue traded volume, market share percentages (reconciled to 100%), + * and period-over-period change against the preceding window. + * Inactive / stale venues without volume in the window are reported under excludedVenues. + * + * Issue: #224 + */ +export async function registerVenueRoutes(app: FastifyInstance): Promise { + app.get<{ + Querystring: { window?: string; network?: string } + }>('/venues', { config: { public: true } }, async (req, reply) => { + const window = req.query.window ?? '24h' + if (!(window in WINDOW_HOURS)) { + return reply + .status(400) + .send({ error: `window must be one of: ${WINDOWS.join(', ')}` }) + } + + const requested = req.query.network + if (requested !== undefined && requested !== 'testnet' && requested !== 'mainnet') { + return reply.status(400).send({ error: 'network must be one of: testnet, mainnet' }) + } + const network: NetworkName = (requested as NetworkName) ?? req.network ?? activeNetwork + + const hours = WINDOW_HOURS[window as VolumeWindow] + const now = new Date() + const currentStart = new Date(now.getTime() - hours * 3600 * 1000) + const prevStart = new Date(currentStart.getTime() - hours * 3600 * 1000) + + try { + // 1. Current window volumes by source + const currentRes = await pgPool.query( + `SELECT source, + SUM(base_volume)::numeric AS volume, + COUNT(*)::int AS trade_count + FROM price_points + WHERE network = $1 + AND timestamp >= $2 + AND timestamp < $3 + GROUP BY source`, + [network, currentStart, now] + ) + + // 2. Previous window volumes by source + const prevRes = await pgPool.query( + `SELECT source, + SUM(base_volume)::numeric AS volume + FROM price_points + WHERE network = $1 + AND timestamp >= $2 + AND timestamp < $3 + GROUP BY source`, + [network, prevStart, currentStart] + ) + + const prevMap = new Map() + for (const row of prevRes.rows) { + prevMap.set(row.source, parseFloat(row.volume) || 0) + } + + // Collect all known venues + const knownVenuesSet = new Set([ + 'SDEX', + 'AMM', + ...venueAdapters.map((a) => a.source), + ]) + + const currentMap = new Map() + let totalCurrentVolume = 0 + + for (const row of currentRes.rows) { + const vol = parseFloat(row.volume) || 0 + const tc = parseInt(row.trade_count, 10) || 0 + currentMap.set(row.source, { volume: vol, tradeCount: tc }) + totalCurrentVolume += vol + knownVenuesSet.add(row.source) + } + + const activeVenues: VenueShareItem[] = [] + const excludedVenues: string[] = [] + + for (const venue of knownVenuesSet) { + const cur = currentMap.get(venue) + const prevVol = prevMap.get(venue) ?? 0 + + if (!cur || cur.volume === 0) { + excludedVenues.push(venue) + continue + } + + const vol = cur.volume + const sharePct = totalCurrentVolume > 0 + ? parseFloat(((vol / totalCurrentVolume) * 100).toFixed(2)) + : 0 + + let periodChangePct: number | null = null + if (prevVol > 0) { + periodChangePct = parseFloat((((vol - prevVol) / prevVol) * 100).toFixed(2)) + } + + activeVenues.push({ + venue, + volume: vol, + sharePct, + previousVolume: prevVol, + periodChangePct, + tradeCount: cur.tradeCount, + }) + } + + // Sort by volume descending + activeVenues.sort((a, b) => b.volume - a.volume) + + // Reconcile share percentages so sum equals 100.00% if totalVolume > 0 + if (activeVenues.length > 0 && totalCurrentVolume > 0) { + const currentSum = activeVenues.reduce((acc, v) => acc + v.sharePct, 0) + const diff = parseFloat((100 - currentSum).toFixed(2)) + if (Math.abs(diff) > 0 && Math.abs(diff) <= 0.1) { + // Adjust the largest venue slightly to reconcile exactly to 100 + activeVenues[0].sharePct = parseFloat((activeVenues[0].sharePct + diff).toFixed(2)) + } + } + + return reply.status(200).send({ + window, + network, + startTime: currentStart.toISOString(), + endTime: now.toISOString(), + totalVolume: totalCurrentVolume, + venues: activeVenues, + excludedVenues, + reconciliationSumPct: activeVenues.reduce((acc, v) => acc + v.sharePct, 0), + }) + } catch (err) { + req.log.error(err) + return reply.status(500).send({ error: 'Failed to compute venue market share' }) + } + }) + app.get('/venues/registry', { config: { public: true } }, async req => { const network = (req.network ?? activeNetwork) as NetworkName const adapters = venueAdapters @@ -60,3 +218,5 @@ export async function registerVenueRoutes(app: FastifyInstance) { } }) } + +export const registerVenuesRoutes = registerVenueRoutes diff --git a/src/routes/volumes.ts b/src/routes/volumes.ts index 802ba03d..730dfe62 100644 --- a/src/routes/volumes.ts +++ b/src/routes/volumes.ts @@ -98,9 +98,10 @@ export async function registerVolumeRoutes(app: FastifyInstance) { tradeCount, } } catch (err) { + req.log.error(err) return reply .status(500) - .send({ error: `Volume aggregation failed: ${(err as Error).message}` }) + .send({ error: 'Volume aggregation failed' }) } }) } diff --git a/src/types.ts b/src/types.ts index 98284a96..e070cbcb 100644 --- a/src/types.ts +++ b/src/types.ts @@ -36,13 +36,23 @@ export interface AggregatedPrice { stale: boolean } +export interface RouteHop { + from: string + to: string + venue: string + rate: number + slippagePct: number +} + export interface RouteInfo { - route: 'SDEX' | 'AMM' | 'SPLIT' | 'UNKNOWN' + route: 'SDEX' | 'AMM' | 'SPLIT' | 'MULTI_HOP' | 'UNKNOWN' sdexPrice: number ammPrice: number estimatedOutput: number slippagePct: number recommendation: string + hops?: RouteHop[] + path?: string[] } export interface WatchedPair { diff --git a/tests/fixtures/venues/amm.json b/tests/fixtures/venues/amm.json new file mode 100644 index 00000000..906da54f --- /dev/null +++ b/tests/fixtures/venues/amm.json @@ -0,0 +1,16 @@ +{ + "venue": "amm", + "sourceUrl": "https://horizon.stellar.org/liquidity_pools/d94943fbe0a8c2d96c1da8c4e5e4de4b4070a2a4bdf34839f99e4b30172e2cfc", + "capturedAt": "2026-10-15T12:00:00.000Z", + "data": { + "id": "d94943fbe0a8c2d96c1da8c4e5e4de4b4070a2a4bdf34839f99e4b30172e2cfc", + "fee_bp": 30, + "type": "constant_product", + "total_trustlines": "412", + "total_shares": "1000000.0000000", + "reserves": [ + { "asset": "native", "amount": "10000000.0000000" }, + { "asset": "USDC:GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335X2KGX3IHOJAPP5RE34K4KZVN", "amount": "1050000.0000000" } + ] + } +} diff --git a/tests/fixtures/venues/aquarius.json b/tests/fixtures/venues/aquarius.json new file mode 100644 index 00000000..d5fe4b2f --- /dev/null +++ b/tests/fixtures/venues/aquarius.json @@ -0,0 +1,12 @@ +{ + "venue": "aquarius", + "sourceUrl": "https://api.aquarius.loan/api/v1/pools/aquarius-amm-xlm-usdc", + "capturedAt": "2026-10-15T12:00:00.000Z", + "data": { + "poolId": "aquarius-amm-xlm-usdc", + "baseReserve": "1500000.0000000", + "quoteReserve": "157500.0000000", + "feeBp": 30, + "lastLedger": 51234567 + } +} diff --git a/tests/fixtures/venues/binance.json b/tests/fixtures/venues/binance.json new file mode 100644 index 00000000..9ce2fa77 --- /dev/null +++ b/tests/fixtures/venues/binance.json @@ -0,0 +1,22 @@ +{ + "venue": "binance", + "sourceUrl": "https://api.binance.com/api/v3/ticker/24hr?symbol=XLMUSDC", + "capturedAt": "2026-10-15T12:00:00.000Z", + "data": { + "symbol": "XLMUSDC", + "priceChange": "0.0020", + "priceChangePercent": "1.94", + "prevClosePrice": "0.1030", + "lastPrice": "0.1050", + "bidPrice": "0.1049", + "askPrice": "0.1051", + "openPrice": "0.1030", + "highPrice": "0.1065", + "lowPrice": "0.1025", + "volume": "15420000.00", + "quoteVolume": "1619100.00", + "openTime": 1792070400000, + "closeTime": 1792156800000, + "count": 4820 + } +} diff --git a/tests/fixtures/venues/sdex.json b/tests/fixtures/venues/sdex.json new file mode 100644 index 00000000..46d02a18 --- /dev/null +++ b/tests/fixtures/venues/sdex.json @@ -0,0 +1,15 @@ +{ + "venue": "sdex", + "sourceUrl": "https://horizon.stellar.org/order_book?selling_asset_type=native&buying_asset_type=credit_alphanum4&buying_asset_code=USDC&buying_asset_issuer=GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335X2KGX3IHOJAPP5RE34K4KZVN", + "capturedAt": "2026-10-15T12:00:00.000Z", + "data": { + "bids": [ + { "price": "0.1049000", "amount": "50000.0000000" }, + { "price": "0.1048000", "amount": "120000.0000000" } + ], + "asks": [ + { "price": "0.1051000", "amount": "45000.0000000" }, + { "price": "0.1052000", "amount": "80000.0000000" } + ] + } +} diff --git a/tests/fixtures/venues/soroswap.json b/tests/fixtures/venues/soroswap.json new file mode 100644 index 00000000..2dbffe7d --- /dev/null +++ b/tests/fixtures/venues/soroswap.json @@ -0,0 +1,13 @@ +{ + "venue": "soroswap", + "sourceUrl": "https://api.soroswap.finance/api/pairs/CCW67TSZV3SSS2HXMBQ5JFGCKJNXKZM7UQUWUZPUTHXSTZLEO7SJMI75", + "capturedAt": "2026-10-15T12:00:00.000Z", + "data": { + "pairAddress": "CCW67TSZV3SSS2HXMBQ5JFGCKJNXKZM7UQUWUZPUTHXSTZLEO7SJMI75", + "token0": "CAS3J7GYLGXMF6TDJBBYYSE3HQ6BBSMLNUQ34T6TZMYMW2EVH34XOWMA", + "token1": "CCW67TSZV3SSS2HXMBQ5JFGCKJNXKZM7UQUWUZPUTHXSTZLEO7SJMI75", + "reserve0": "2500000.0000000", + "reserve1": "262500.0000000", + "fee": 30 + } +} diff --git a/tests/helpers/venueTestHelper.ts b/tests/helpers/venueTestHelper.ts new file mode 100644 index 00000000..c56317c8 --- /dev/null +++ b/tests/helpers/venueTestHelper.ts @@ -0,0 +1,101 @@ +/** + * Offline Venue Test Fixture Layer + * + * Intercepts external HTTP requests during venue adapter tests. + * Resolves registered offline fixture data and strictly blocks unmocked external network access, + * failing loudly to guarantee tests remain fast, deterministic, and 100% offline. + * + * Issue: #226 + */ + +import { readFileSync } from 'node:fs' +import { join } from 'node:path' + +export interface VenueFixture { + venue: string + sourceUrl: string + capturedAt: string + data: T +} + +const originalFetch = globalThis.fetch + +const fixtureCache = new Map() + +export function loadVenueFixture(filename: string): VenueFixture { + if (fixtureCache.has(filename)) { + return fixtureCache.get(filename) as VenueFixture + } + const fullPath = join(process.cwd(), 'tests', 'fixtures', 'venues', filename) + const content = readFileSync(fullPath, 'utf8') + const fixture = JSON.parse(content) as VenueFixture + fixtureCache.set(filename, fixture) + return fixture +} + +export interface OfflineFixtureOptions { + strictOffline?: boolean + urlMappings?: Record +} + +/** + * Installs the offline network interceptor. + * Matches URLs to recorded fixture data or explicit mappings. + * Fails loudly with an Error whenever an unmocked URL is fetched. + */ +export function installOfflineVenueInterceptor(options: OfflineFixtureOptions = {}): void { + const strict = options.strictOffline ?? true + const mappings = options.urlMappings ?? {} + + // Pre-load default fixtures + const sdex = loadVenueFixture('sdex.json') + const soroswap = loadVenueFixture('soroswap.json') + const aquarius = loadVenueFixture('aquarius.json') + const amm = loadVenueFixture('amm.json') + const binance = loadVenueFixture('binance.json') + + const defaultUrls: [string, any][] = [ + [sdex.sourceUrl, sdex.data], + [soroswap.sourceUrl, soroswap.data], + [aquarius.sourceUrl, aquarius.data], + [amm.sourceUrl, amm.data], + [binance.sourceUrl, binance.data], + ] + + globalThis.fetch = async (input: RequestInfo | URL, init?: RequestInit): Promise => { + const urlStr = typeof input === 'string' ? input : input instanceof URL ? input.toString() : input.url + + // Check custom mappings first + for (const [pattern, data] of Object.entries(mappings)) { + if (urlStr.includes(pattern)) { + return new Response(JSON.stringify(data), { + status: 200, + headers: { 'Content-Type': 'application/json' }, + }) + } + } + + // Check default fixture URLs + for (const [sourceUrl, data] of defaultUrls) { + if (urlStr === sourceUrl || urlStr.startsWith(sourceUrl.split('?')[0])) { + return new Response(JSON.stringify(data), { + status: 200, + headers: { 'Content-Type': 'application/json' }, + }) + } + } + + if (strict) { + throw new Error(`[OfflineFixtureLayer] Blocked unmocked network call to: ${urlStr}`) + } + + return originalFetch(input, init) + } +} + +/** + * Restores the global fetch implementation. + */ +export function restoreOfflineVenueInterceptor(): void { + globalThis.fetch = originalFetch +} diff --git a/tests/readmeSync.test.ts b/tests/readmeSync.test.ts new file mode 100644 index 00000000..d4162153 --- /dev/null +++ b/tests/readmeSync.test.ts @@ -0,0 +1,39 @@ +import { describe, it, expect } from 'vitest' +import { readFileSync } from 'node:fs' +import { fileURLToPath } from 'node:url' +import { + renderEndpointsMarkdown, + renderVenuesMarkdown, + renderOpenApiJson, + VENUES_REGISTRY, +} from '../scripts/generate-openapi' + +describe('README and OpenAPI Sync Guard (#275, #229)', () => { + const readmePath = fileURLToPath(new URL('../README.md', import.meta.url)) + const openApiJsonPath = fileURLToPath(new URL('../openapi.json', import.meta.url)) + + it('committed openapi.json matches generated openapi.json byte-for-byte', () => { + const committedJson = readFileSync(openApiJsonPath, 'utf8') + const generatedJson = renderOpenApiJson() + expect(committedJson).toBe(generatedJson) + }) + + it('committed README endpoints table matches openapi.yaml', () => { + const readmeContent = readFileSync(readmePath, 'utf8') + const match = readmeContent.match(/([\s\S]*?)/) + expect(match).not.toBeNull() + const committedTable = match![1].trim() + const generatedTable = renderEndpointsMarkdown().trim() + expect(committedTable).toBe(generatedTable) + }) + + it('committed README venue list matches canonical venue registry', () => { + const readmeContent = readFileSync(readmePath, 'utf8') + const match = readmeContent.match(/([\s\S]*?)/) + expect(match).not.toBeNull() + const committedVenues = match![1].trim() + const generatedVenues = renderVenuesMarkdown().trim() + expect(committedVenues).toBe(generatedVenues) + expect(VENUES_REGISTRY.length).toBeGreaterThanOrEqual(5) + }) +})