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.okx.com:8443/ws/v5/public";7const CRYPTO_PAIRS: Array<[string, string]> = MAJOR_BASES.filter((b) => b !== "PAXG").map((b) => [b, "USDT"] as [string, string]).concat([["TRX", "USDT"]]);8/** USDT markets against fiat → live FX proxies (STABLECOIN_PROXY). */9const FX_PROXY_PAIRS: Array<[string, string]> = [["USDT", "EUR"], ["USDT", "AUD"], ["USDT", "BRL"], ["USDT", "SGD"], ["USDT", "TRY"], ["USDT", "AED"]];10const PAIRS = [...CRYPTO_PAIRS, ...FX_PROXY_PAIRS];11const SYMBOLS = PAIRS.map(([b, q]) => `${b}-${q}`);1213/** OKX public WebSocket v5, `tickers` channel. Requires a text "ping" at least every 30 s. */14export const okxWs = defineConnector({15  metadata: {16    id: "okx-ws",17    name: "OKX — tickers channel",18    version: "1.0.0",19    sourceId: "okx",20    organization: "OKX",21    sourceType: "WEBSOCKET",22    jurisdiction: null,23    rightsStatus: "PUBLIC_ATTRIBUTED",24    realtimeStatus: "REALTIME",25    expectedLatencyMs: 500,26    supportsStreaming: true,27    supportsHistorical: false,28    assetClasses: ["CRYPTO", "FOREX"],29    exchanges: ["okx"],30    homepage: "https://www.okx.com/docs-v5/en/#public-data-websocket-tickers-channel",31    description: "Public OKX v5 WebSocket tickers channel (last, bid/ask with sizes, 24h open/high/low/volume) for USDT spot majors. Exchange timestamp `ts`.",32    rightsNotes: "Public market data displayed with attribution to OKX.",33    termsUrl: "https://www.okx.com/help/terms-of-service",34    sourceFamily: "okx",35    enabled: true,36  },37  seeds: cryptoSeeds(CRYPTO_PAIRS, "okx", (b, q) => `${b}-${q}`),38  defaultSymbols: SYMBOLS,39  async start(ctx) {40    const ws = ctx.openWebSocket(WS_URL, {41      label: "okx",42      staleAfterMs: 60_000,43      heartbeat: { intervalMs: 20_000, message: "ping" },44      onOpen: (sock) => sock.send({ op: "subscribe", args: ctx.watchedSymbols().map((instId) => ({ channel: "tickers", instId })) }),45      onMessage: (data) => {46        if (data === "pong") return;47        let msg: any;48        try {49          msg = JSON.parse(data);50        } catch {51          return;52        }53        if (msg?.arg?.channel === "tickers" && Array.isArray(msg.data)) ctx.emit(raw("okx-ws", "okx", "tickers", msg));54        else if (msg?.event === "error") ctx.reportError(new Error(String(msg.msg ?? "okx error")), { code: msg.code });55      },56    });57    ws.connect();58  },59  normalize(r): NormalizedBatch {60    const m = r.payload as { arg?: { channel?: string }; data?: Array<Record<string, unknown>> };61    if (r.kind !== "tickers" || m.arg?.channel !== "tickers" || !Array.isArray(m.data)) return { observations: [] };62    const observations = [];63    for (const t of m.data) {64      if (typeof t.instId !== "string") continue;65      const [base, quote] = t.instId.split("-");66      if (!base || !quote) continue;67      observations.push(68        ...pairObservations(t.instId, base, quote, "okx", { last: t.last, open: t.open24h, high: t.high24h, low: t.low24h, bid: t.bidPx, ask: t.askPx, bidSize: t.bidSz, askSize: t.askSz, volume: t.vol24h }, {69          sourceTimestamp: parseTimestamp(t.ts),70          timestampTrust: "EXCHANGE",71          rightsStatus: "PUBLIC_ATTRIBUTED",72          realtimeStatus: "REALTIME",73        }),74      );75    }76    return { observations };77  },78  fixturesDir: "fixtures",79});80