spb/market-atlas

1import { defineConnector, raw, type NormalizedBatch } from "@market-atlas/connector-sdk";2import { parseTimestamp } from "@market-atlas/market-model";3import { MAJOR_BASES } from "../_shared/crypto.js";4import { pairObservations, planPair } from "../_shared/pairs.js";56const WS_URL = "wss://stream.crypto.com/exchange/v1/market";7const SYMBOLS = [...MAJOR_BASES.filter((b) => b !== "PAXG").map((b) => `${b}_USD`), "BTC_USDT", "ETH_USDT"];89/** Crypto.com Exchange public market stream, `ticker.<instrument>`. The server sends `public/heartbeat` that must be answered. */10export const cryptocomWs = defineConnector({11  metadata: {12    id: "cryptocom-ws",13    name: "Crypto.com Exchange — ticker channel",14    version: "1.0.0",15    sourceId: "cryptocom",16    organization: "Crypto.com",17    sourceType: "WEBSOCKET",18    jurisdiction: "SG",19    rightsStatus: "PUBLIC_ATTRIBUTED",20    realtimeStatus: "REALTIME",21    expectedLatencyMs: 500,22    supportsStreaming: true,23    supportsHistorical: false,24    assetClasses: ["CRYPTO"],25    exchanges: ["cryptocom"],26    homepage: "https://exchange-docs.crypto.com/exchange/v1/rest-ws/index.html#ticker-instrument_name",27    description: "Public Crypto.com Exchange ticker stream (last, best bid/ask with sizes, 24h high/low/volume) for USD-quoted majors and BTC/ETH in USDT.",28    rightsNotes: "Public market data displayed with attribution to Crypto.com Exchange.",29    termsUrl: "https://crypto.com/exchange/document/terms-of-service",30    sourceFamily: "cryptocom",31    enabled: true,32  },33  seeds: SYMBOLS.map((s) => {34    const [b, q] = s.split("_") as [string, string];35    return { symbol: s, hint: planPair(b, q, "cryptocom").hint };36  }),37  defaultSymbols: SYMBOLS,38  async start(ctx) {39    const ws = ctx.openWebSocket(WS_URL, {40      label: "cryptocom",41      staleAfterMs: 60_000,42      heartbeat: null, // server-initiated heartbeats answered below43      onOpen: (sock) => setTimeout(() => sock.send({ id: 1, method: "subscribe", params: { channels: ctx.watchedSymbols().map((s) => `ticker.${s}`) } }), 1000), // docs: wait 1 s after connect44      onMessage: (data, sock) => {45        let msg: any;46        try {47          msg = JSON.parse(data);48        } catch {49          return;50        }51        if (msg?.method === "public/heartbeat") return void sock.send({ id: msg.id, method: "public/respond-heartbeat" });52        if (msg?.method === "subscribe" && msg.result?.channel === "ticker" && Array.isArray(msg.result.data)) ctx.emit(raw("cryptocom-ws", "cryptocom", "ticker", msg.result));53        else if (msg?.code && msg.code !== 0) ctx.reportError(new Error(String(msg.message ?? `cryptocom code ${msg.code}`)));54      },55    });56    ws.connect();57  },58  normalize(r): NormalizedBatch {59    const m = r.payload as { instrument_name?: string; data?: Array<Record<string, unknown>> };60    if (r.kind !== "ticker" || typeof m.instrument_name !== "string" || !Array.isArray(m.data)) return { observations: [] };61    const [base, quote] = m.instrument_name.split("_");62    if (!base || !quote) return { observations: [] };63    const observations = [];64    for (const t of m.data) {65      observations.push(66        ...pairObservations(m.instrument_name, base, quote, "cryptocom", { last: t.a, bid: t.b, bidSize: t.bs, ask: t.k, askSize: t.ks, high: t.h, low: t.l, volume: t.v }, {67          sourceTimestamp: parseTimestamp(t.t),68          timestampTrust: "EXCHANGE",69          rightsStatus: "PUBLIC_ATTRIBUTED",70          realtimeStatus: "REALTIME",71        }),72      );73    }74    return { observations };75  },76  fixturesDir: "fixtures",77});78