spb/market-atlas

1import { defineConnector, raw, type NormalizedBatch } from "@market-atlas/connector-sdk";2import { parseTimestamp } from "@market-atlas/market-model";3import { pairObservations, planPair, splitVenueSymbol } from "../_shared/pairs.js";45const WS_URL = "wss://api.gemini.com/v2/marketdata";6const SYMBOLS = ["BTCUSD", "ETHUSD", "SOLUSD", "XRPUSD", "LTCUSD", "LINKUSD", "DOGEUSD", "AVAXUSD", "DOTUSD", "BTCGUSD"];78/** Gemini market data v2, `l2` subscription: initial snapshot with recent trades, then `trade` events. */9export const geminiWs = defineConnector({10  metadata: {11    id: "gemini-ws",12    name: "Gemini — market data v2 (trades)",13    version: "1.0.0",14    sourceId: "gemini",15    organization: "Gemini Trust Company, LLC",16    sourceType: "WEBSOCKET",17    jurisdiction: "US",18    rightsStatus: "PUBLIC_ATTRIBUTED",19    realtimeStatus: "REALTIME",20    expectedLatencyMs: 400,21    supportsStreaming: true,22    supportsHistorical: false,23    assetClasses: ["CRYPTO"],24    exchanges: ["gemini"],25    homepage: "https://docs.gemini.com/websocket-api/#market-data-version-2",26    description: "Public Gemini market-data v2 feed (l2 subscription): last trades for the major USD pairs with exchange timestamps.",27    rightsNotes: "Public market data displayed with attribution to Gemini.",28    termsUrl: "https://www.gemini.com/legal/api-agreement",29    sourceFamily: "gemini",30    enabled: true,31  },32  seeds: SYMBOLS.map((s) => {33    const [b, q] = splitVenueSymbol(s, null)!;34    return { symbol: s, hint: planPair(b, q, "gemini").hint, aliases: [`${b}-${q}`, `${b}/${q}`] };35  }),36  defaultSymbols: SYMBOLS,37  async start(ctx) {38    const ws = ctx.openWebSocket(WS_URL, {39      label: "gemini",40      staleAfterMs: 180_000,41      heartbeat: { intervalMs: 25_000 },42      onOpen: (sock) => sock.send({ type: "subscribe", subscriptions: [{ name: "l2", symbols: ctx.watchedSymbols() }] }),43      onMessage: (data) => {44        let msg: any;45        try {46          msg = JSON.parse(data);47        } catch {48          return;49        }50        if (msg?.type === "trade") ctx.emit(raw("gemini-ws", "gemini", "trade", msg));51        else if (msg?.type === "l2_updates" && Array.isArray(msg.trades) && msg.trades.length) ctx.emit(raw("gemini-ws", "gemini", "trade", { ...msg.trades[msg.trades.length - 1], symbol: msg.symbol, type: "trade" }));52      },53    });54    ws.connect();55  },56  normalize(r): NormalizedBatch {57    const m = r.payload as Record<string, unknown>;58    if (r.kind !== "trade" || typeof m.symbol !== "string") return { observations: [] };59    const split = splitVenueSymbol(m.symbol, null);60    if (!split) return { observations: [] };61    const [base, quote] = split;62    return {63      observations: pairObservations(m.symbol, base, quote, "gemini", { last: m.price }, {64        sourceTimestamp: parseTimestamp(m.timestamp),65        timestampTrust: "EXCHANGE",66        sequence: typeof m.event_id === "number" ? m.event_id : null,67        rightsStatus: "PUBLIC_ATTRIBUTED",68        realtimeStatus: "REALTIME",69      }),70    };71  },72  fixturesDir: "fixtures",73});74