spb/market-atlas

1import { defineConnector, raw, type NormalizedBatch } from "@market-atlas/connector-sdk";2import { parseTimestamp } from "@market-atlas/market-model";3import { MAJOR_BASES, cryptoSeeds } from "../_shared/crypto.js";4import { pairObservations } from "../_shared/pairs.js";56const WS_URL = "wss://ws-feed.exchange.coinbase.com";7const CRYPTO_PAIRS: Array<[string, string]> = [...MAJOR_BASES.map((b) => [b, "USD"] as [string, string]), ["BTC", "USDT"], ["ETH", "USDT"], ["BTC", "EUR"], ["ETH", "EUR"]];8/** Stablecoin/fiat markets → live FX proxies (USDC ≈ USD). */9const FX_PROXY_PAIRS: Array<[string, string]> = [["USDC", "EUR"], ["USDC", "GBP"], ["USDT", "EUR"], ["USDT", "GBP"]];10const PAIRS = [...CRYPTO_PAIRS, ...FX_PROXY_PAIRS];11const SYMBOLS = PAIRS.map(([b, q]) => `${b}-${q}`);1213/**14 * Coinbase Exchange public market-data feed, `ticker` channel: one multiplexed connection for all15 * products, best bid/ask, last trade price, 24h stats. Exchange-generated timestamps.16 */17export const coinbaseWs = defineConnector({18  metadata: {19    id: "coinbase-ws",20    name: "Coinbase Exchange — ticker feed",21    version: "1.0.0",22    sourceId: "coinbase",23    organization: "Coinbase Global, Inc.",24    sourceType: "WEBSOCKET",25    jurisdiction: "US",26    rightsStatus: "PUBLIC_ATTRIBUTED",27    realtimeStatus: "REALTIME",28    expectedLatencyMs: 300,29    supportsStreaming: true,30    supportsHistorical: false,31    assetClasses: ["CRYPTO", "FOREX"],32    exchanges: ["coinbase"],33    homepage: "https://docs.cdp.coinbase.com/exchange/docs/websocket-overview",34    description: "Public, unauthenticated WebSocket market-data feed of Coinbase Exchange (ticker channel). One connection carries every subscribed product.",35    rightsNotes: "Public market data; displayed with attribution to Coinbase Exchange. Not for commercial redistribution as a raw feed.",36    termsUrl: "https://www.coinbase.com/legal/market_data",37    sourceFamily: "coinbase",38    enabled: true,39  },40  seeds: cryptoSeeds(CRYPTO_PAIRS, "coinbase", (b, q) => `${b}-${q}`),41  defaultSymbols: SYMBOLS,42  async start(ctx) {43    const ws = ctx.openWebSocket(WS_URL, {44      label: "coinbase",45      staleAfterMs: 90_000,46      heartbeat: { intervalMs: 20_000 },47      onOpen: (sock) => {48        sock.send({ type: "subscribe", product_ids: ctx.watchedSymbols(), channels: ["ticker", "heartbeat"] });49      },50      onMessage: (data) => {51        let msg: any;52        try {53          msg = JSON.parse(data);54        } catch {55          return;56        }57        if (msg?.type === "ticker") ctx.emit(raw("coinbase-ws", "coinbase", "ticker", msg));58        else if (msg?.type === "error") ctx.reportError(new Error(String(msg.message ?? "coinbase error")), { reason: msg.reason });59      },60    });61    ws.connect();62  },63  normalize(r): NormalizedBatch {64    const m = r.payload as Record<string, unknown>;65    if (r.kind !== "ticker" || typeof m.product_id !== "string") return { observations: [] };66    const [base, quote] = m.product_id.split("-");67    if (!base || !quote) return { observations: [] };68    return {69      observations: pairObservations(m.product_id, base, quote, "coinbase", { last: m.price, open: m.open_24h, high: m.high_24h, low: m.low_24h, bid: m.best_bid, ask: m.best_ask, bidSize: m.best_bid_size, askSize: m.best_ask_size, volume: m.volume_24h }, {70        sourceTimestamp: parseTimestamp(m.time),71        timestampTrust: "EXCHANGE",72        sequence: typeof m.sequence === "number" ? m.sequence : null,73        rightsStatus: "PUBLIC_ATTRIBUTED",74        realtimeStatus: "REALTIME",75      }),76    };77  },78  fixturesDir: "fixtures",79});80