Market Ingestion

The ingestion layer continuously fetches market data from prediction market exchanges and crypto price feeds, normalizes it into a unified schema, and stores it in PostgreSQL. Most pipelines run on BullMQ repeatable jobs; some use long-lived WebSocket connectors.

Exchange Connectors

Polymarket

Polymarket data comes through three APIs:

• Gamma API -- Market listings, metadata, categories, and event groupings
• CLOB API -- Order book depth, historical price candles (fidelity=60 hourly, fidelity=1440 daily)
• CLOB WebSocket (wss://ws-subscriptions-clob.polymarket.com/ws/market) -- Real-time price updates for short-duration markets

Connector fetches the top events by volume (capped at 200 events).

DFlow (Kalshi via Solana)

DFlow data comes through two REST APIs:

• Metadata API (DFLOW_METADATA_API_URL) -- Market listings, prices, volumes
• Trade API (DFLOW_TRADE_API_URL) -- Order quotes, status (proxied through backend at /api/v1/dflow/* due to CORS)

DFlow tokenizes Kalshi markets as SPL Token-2022 outcome tokens on Solana. The connector also caps event fetching at 200.

Polymarket RTDS (Chainlink)

PolymarketRtdsConnector streams real-time Chainlink oracle prices for BTC/ETH/SOL/XRP from wss://ws-live-data.polymarket.com. Subscribes to the crypto_prices_chainlink topic on connect, throttles broadcasts to 250ms per symbol, and caches latest prices in Redis (crypto:prices:latest:chainlink, 2min TTL) as a fallback for REST endpoints after restarts. Writes ticks to CryptoPriceTick (cleaned up after 24h). Broadcasts to the crypto:prices Socket.IO channel. Runs on the dedicated worker-rtds container (WORKER_DOMAIN=rtds).

Polymarket WebSocket

PolymarketWsConnector streams real-time prediction market prices for short-duration markets. Lazy -- only connects when markets are registered via registerMarket(tokenId, marketId). Has a 120s watchdog for silent freeze detection, exponential backoff reconnect, and 500ms batched DB writes.

DFlow WebSocket

DflowWsConnector streams real-time prices and order book depth for DFlow (Kalshi) markets from wss://prediction-markets-api.dflow.net/api/v1/ws. Authenticates via x-api-key header — no RSA-PSS signing required. Subscribes globally to the prices channel on connect ({ all: true }) for full-state ticker updates, and per-ticker to orderbook for depth snapshots. Has a 120s watchdog for freeze detection, exponential backoff reconnect, and 250ms batched DB writes. Gated behind WS_EXPANSION_ENABLED.

Sync Schedules

The market-sync BullMQ queue runs three repeatable jobs:

Memory Optimizations

The price-sync job processes ~4,500 markets every 60s. Two critical gates prevent the Node process from growing past 5 GB:

1. Delta-only writes -- Both MarketPriceSnapshot inserts AND Market row updates skip when |price_change| <= 0.0001. Phase 3.4 (2026-04-26) extended the existing snapshot gate to cover the UPDATE itself, eliminating ~11M no-op writes per 12 days.
2. Deferred price-change query -- The expensive updatePriceChanges() runs only during full-sync (every 5 min), not every price-sync cycle.

Event Group Aggregation

MarketSyncProcessor.aggregateEventGroupStats() rolls child-market volumeTotal + liquidity into the EventGroup totals. Phase 3.3 (2026-04-26) collapsed the per-group prisma.market.aggregate loop (~1000 calls/cycle) into a single UPDATE EventGroup ... FROM (SELECT ... GROUP BY eventGroupId). One round-trip per cycle now does the whole rollup.

Short-Duration Pipeline (5m/15m markets)

Rapid-cycle crypto Up/Down markets run on an isolated short-duration-sync BullMQ queue (separate from market-sync):

ShortDurationPipelineService handles discovery and slot prediction. When durationType is SHORT_5M or SHORT_15M, market listing is force-sorted by closesAt ASC.

The frontend's TimeframeNav generates expected clock-aligned slots (past 2h + future 30min) and merges with real sibling markets using 60s tolerance dedup. Missing markets show as dimmed/disabled placeholders.

Trade Sync

TradeSyncService.insertTrades() uses INSERT INTO ExchangeTrade ... ON CONFLICT ("source","externalId") DO NOTHING RETURNING source, externalId (raw SQL via Prisma.sql/Prisma.join).

This replaced the prior two-step (findMany pre-check + createMany skipDuplicates) -- createMany returns only a count, but we need the inserted (source, externalId) pairs to drive the insider-signals enqueue hook on fresh rows. The unique index (source, externalId) on ExchangeTrade makes the ON CONFLICT path efficient.

Two trade producers feed insider signals:

1. TradeSyncService.enqueueInsiderSignalsClassify() -- runs from the hot-set trade-sync (every 5 min, ~850 tracked markets). Pre-filters source=polymarket && takerAddress && amount >= INSIDER_SIGNALS_MIN_USD, enqueues classify-trade with jobId: pm:{txHash} (BullMQ de-dupes).
2. InsiderTradePollerService -- global recent-trades poller (every 30s, off by default). Hits Polymarket Data API /trades?limit=500 (no market filter), maps conditionId → internal marketId via one Prisma query.

Normalization

NormalizerService converts exchange-specific data into the unified Market model:

Key normalizations:

• Slug generation -- Consistent URL-safe slugs from market titles
• Category mapping -- Exchange-specific categories mapped to Vezta categories
• Price format -- Both exchanges normalize to yesPrice / noPrice as Decimal(10,6)
• Volume -- Standardized to USD with Decimal(18,2) precision
• Timestamps -- startedAt uses the exchange's creation time (Polymarket startDate ?? createdAt, DFlow openTime); createdAt is when Vezta ingested the market

Price History Backfill

A separate price-history-backfill job (every 60s) populates historical price data for charts:

insertBackfillPoints() only inserts points older than the earliest existing snapshot (no duplicates), and only marks priceHistoryBackfilled = true if the API returned >0 points.

Chart Data

The getChartData endpoint returns OHLC data aggregated via SQL date_bin():

Backed by index MarketPriceSnapshot_chart_query_idx on (marketId, createdAt DESC) INCLUDE (yesPrice) (Phase 3.1 migration). The covering payload lets Postgres do an index-only scan -- no heap fetch for the price column.

On-Demand Ingestion

When a user lands on a market URL (slug) that is not yet in the database, the on-demand ingestion path fetches and ingests it with the same 200-cap rule. A 48-hour cleanup job removes markets that were on-demand-ingested but never traded.

Market Tiering (HOT / WARM / COLD)

Not every market is streamed and price-synced at the same cadence. A tiering model in market/tier/ decides which markets get live WebSocket subscriptions and how frequently their prices are refreshed, keeping the connector count and price-sync load bounded:

Key services: HotSetService (the working set of streamed markets), SubscriptionSetService + WsSubscriptionManager (which channels the websocket worker actually opens), and TierPromoterService (promotes/demotes markets between tiers). Viewer presence is tracked in Redis under ws:active:market:<id> keys, so a market a user is currently looking at is promoted to HOT and streamed in real time, then demoted when nobody is watching.