Enriched trade stream

wss://stream.kairos.trade

SubscribeRequest {
  contract_id: "570362"
  provider: "polymarket"
  topics: ["trades.enriched", "trades.enriched.pnl"]
}

npm install @bufbuild/protobuf ws
npm install -D @bufbuild/buf @bufbuild/protoc-gen-es @types/ws

cat > buf.gen.yaml <<'EOF'
version: v2
inputs:
  - directory: proto
plugins:
  - local: protoc-gen-es
    out: gen
    opt: target=ts
EOF
npx buf generate

import WebSocket from "ws";
import { create, fromBinary, toBinary } from "@bufbuild/protobuf";
import { ErrorResponseSchema, SubscribeRequestSchema } from "./gen/kairos/v1/websocket_control_pb";
import {
  EnrichedTradeBatchSchema, IdentityStatus, PnLStatus, TradePnLBatchSchema,
  type EnrichedTrade, type MarketPnL, type TraderIdentity,
} from "./gen/kairos/enriched/v1/enriched_trade_pb";

const TAG_SUBSCRIBE = 0x10;
const TAG_ERROR = 0x17;
const TAG_ENRICHED_TRADES = 0x20;
const TAG_TRADE_PNL = 0x21;
const MAX_ROWS = 2_000;

interface Row { trade?: EnrichedTrade; pnl?: MarketPnL }

// One row per print, keyed by (provider, contract, epoch, seq).
const rows = new Map<string, Row>();
const names = new Map<string, string>(); // latest public name per wallet

function rowFor(provider: number, contract: string, epoch: bigint, seq: bigint): Row {
  const key = `${provider}:${contract}:${epoch}:${seq}`;
  let row = rows.get(key);
  if (!row) {
    row = {};
    rows.set(key, row);
    if (rows.size > MAX_ROWS) rows.delete(rows.keys().next().value!);
  }
  return row;
}

function remember(provider: number, identity?: TraderIdentity) {
  if (!identity?.wallet) return;
  const key = `${provider}:${identity.wallet}`;
  if (identity.status === IdentityStatus.READY) {
    names.set(key, identity.displayName || identity.pseudonym || identity.wallet);
  } else if (identity.status === IdentityStatus.ANONYMOUS) {
    names.delete(key);
  }
}

function render(provider: number, row: Row) {
  const trade = row.trade?.trade;
  if (!trade) return; // PnL arrived first; wait for the print
  const who = names.get(`${provider}:${trade.takerAddress.toLowerCase()}`) ?? trade.takerAddress;
  // protobuf-es keeps the PNL_STATUS_ prefix on this enum's values.
  const pnl = row.pnl?.status === PnLStatus.PNL_STATUS_READY
    ? `realized ${row.pnl.realizedPnlUsd.toFixed(2)} unrealized ${row.pnl.unrealizedPnlUsd.toFixed(2)}`
    : "no PnL";
  console.log(trade.tradeId, who, trade.price / 10_000, pnl);
}

const ws = new WebSocket("wss://stream.kairos.trade", {
  headers: {
    "X-Client-Id": process.env.KAIROS_CLIENT_ID!,
    "X-Api-Key": process.env.KAIROS_API_KEY!,
    "X-Api-Secret": process.env.KAIROS_API_SECRET!,
  },
});

ws.on("open", () => {
  const request = create(SubscribeRequestSchema, {
    contractId: "570362", // numeric market id: the trade-tape id
    provider: "polymarket",
    topics: ["trades.enriched", "trades.enriched.pnl"],
  });
  ws.send(Uint8Array.of(TAG_SUBSCRIBE, ...toBinary(SubscribeRequestSchema, request)));
});

ws.on("message", (raw: WebSocket.RawData) => {
  const data = raw as Buffer; // binary frames arrive as a Buffer by default
  const payload = data.subarray(1);
  switch (data[0]) {
    case TAG_ENRICHED_TRADES: {
      const batch = fromBinary(EnrichedTradeBatchSchema, payload);
      for (const record of batch.trades) {
        remember(batch.provider, record.taker);
        remember(batch.provider, record.maker);
        const row = rowFor(batch.provider, batch.contractId, batch.epoch, record.seq);
        row.trade = record;
        render(batch.provider, row);
      }
      break;
    }
    case TAG_TRADE_PNL: {
      const batch = fromBinary(TradePnLBatchSchema, payload);
      for (const record of batch.pnl) {
        const row = rowFor(batch.provider, batch.contractId, batch.epoch, record.seq);
        row.pnl = record.pnl; // upsert: a later frame for the same seq replaces this
        render(batch.provider, row);
      }
      break;
    }
    case TAG_ERROR: {
      const error = fromBinary(ErrorResponseSchema, payload);
      console.error(error.action, error.message);
      break;
    }
    default:
      break; // ignore other tags for forward compatibility
  }
});