#Build a live trading dashboard

// protocol.ts: imported by the browser and the server.
export type Quote = { symbol: string; price: number; change: number };

/** A price level of an order book: [price, size]. */
export type Level = [price: number, size: number];

export type Order = { symbol: string; side: 'buy' | 'sell'; qty: number; limit: number };

export type MarketIn =
  // The latest quote, on `quote:<symbol>`. Sent at once on subscribe, then on every change.
  | { type: 'quote'; topic: string; quote: Quote }
  // An order book on `book:<symbol>`: the whole book as of `seq`, then numbered changes.
  | { type: 'book'; topic: string; seq: number; bids: Level[]; asks: Level[] }
  | { type: 'level'; topic: string; seq: number; side: 'bid' | 'ask'; price: number; size: number }
  // The answer to an order.
  | { type: 'placed'; id: number; orderId: string; filled: number }
  | { type: 'rejected'; id: number; reason: string };

export type MarketOut =
  | { type: 'subscribe'; topic: string; after?: number }
  | { type: 'unsubscribe'; topic: string }
  | ({ type: 'order'; id: number; clientOrderId: string } & Order);

// live.ts
import {
  createWebSocket,
  createWebSocketFeeds,
  createWebSocketRequests,
  createWebSocketState,
  createWebSocketTopics
} from 'react-websocket';
import type { Level, MarketIn, MarketOut, Order, Quote } from './protocol';

export const market = createWebSocket<MarketIn, MarketOut>('wss://market.example.com/live', {
  json: true,
  // A frozen price looks just like a quiet market. Ping every 10 s, and reconnect if nothing
  // at all arrives within 3 s of a ping.
  heartbeat: { interval: 10_000, pong: 'pong', timeout: 3_000 }
});

// Quotes: only the latest one matters, so plain topics, subscribed again after a reconnect.
export const quoteTopics = createWebSocketTopics(market, {
  subscribe: topic => ({ type: 'subscribe', topic }),
  unsubscribe: topic => ({ type: 'unsubscribe', topic })
});

export const quotes = createWebSocketState(market, {
  initialState: {} as Record<string, Quote>,
  reduce: (all, message) =>
    message.type === 'quote' ? { ...all, [message.quote.symbol]: message.quote } : all,
  batch: 'frame' // one render per frame at most, however fast quotes arrive
});

// Order books: every change counts, in order, so numbered feeds.
export const bookFeeds = createWebSocketFeeds(market, {
  subscribe: (topic, after) => ({ type: 'subscribe', topic, after }),
  unsubscribe: topic => ({ type: 'unsubscribe', topic }),
  sequence: message => (message.type === 'level' ? message : undefined),
  snapshot: message => (message.type === 'book' ? message : undefined),
  onGap: (topic, after, received) =>
    console.warn(`${topic}: expected ${after + 1}, got ${received}`)
});

export type Book = { bids: Map<number, number>; asks: Map<number, number> };

export const books = createWebSocketState(bookFeeds, {
  initialState: {} as Record<string, Book>,
  reduce: (all, message) => {
    if (message.type === 'book') {
      const book = { bids: new Map(message.bids), asks: new Map(message.asks) };
      return { ...all, [message.topic]: book };
    }
    if (message.type !== 'level') return all;
    const book = all[message.topic];
    if (!book) return all;
    const key = message.side === 'bid' ? 'bids' : 'asks';
    const side = new Map(book[key]);
    if (message.size === 0) side.delete(message.price);
    else side.set(message.price, message.size);
    return { ...all, [message.topic]: { ...book, [key]: side } };
  },
  batch: 'frame'
});

/** The best `depth` levels of one side, best first. */
export function top(side: Map<number, number>, depth: number, descending: boolean): Level[] {
  return [...side].sort((a, b) => (descending ? b[0] - a[0] : a[0] - b[0])).slice(0, depth);
}

export class OrderRejected extends Error {}

export const orders = createWebSocketRequests(market, {
  send: (order: Order & { clientOrderId: string }, id) => ({ type: 'order', id, ...order }),
  reply: message =>
    message.type === 'placed' || message.type === 'rejected' ? message.id : undefined,
  error: message => (message.type === 'rejected' ? new OrderRejected(message.reason) : undefined)
});

export async function placeOrder(order: Order) {
  // Never queue an order behind a reconnect: by then the price has moved.
  if (market.getSnapshot().status !== 'open') {
    throw new Error("You're offline. The order wasn't sent.");
  }
  // The idempotency key: the server places each one once, and answers again with its result.
  const message = { ...order, clientOrderId: crypto.randomUUID() };
  try {
    return await orders.request(message, { volatile: true, timeout: 5_000 });
  } catch (error) {
    if (error instanceof OrderRejected) throw error;
    // Sent, but no answer: it may have been placed. Ask again with the same key, after the
    // reconnect if there is one, to find out without placing it twice.
    return await orders.request(message, { timeout: 30_000 });
  }
}

// server.ts: a simulated market. In production, these messages come from your market data feed.
import { randomUUID } from 'node:crypto';
import { setTimeout as sleep } from 'node:timers/promises';
import { WebSocketServer, type WebSocket } from 'ws';
import type { Level, MarketIn, MarketOut, Order } from './protocol.ts';

type Book = {
  mid: number;
  seq: number;
  bids: Map<number, number>;
  asks: Map<number, number>;
  history: (MarketIn & { type: 'level' })[]; // the latest changes, to resume from
};

const prices = new Map<string, { price: number; open: number }>();
const books = new Map<string, Book>();
const subscribers = new Map<string, Set<WebSocket>>();
const placed = new Map<string, Promise<{ orderId: string; filled: number }>>(); // your database

function send(socket: WebSocket, message: MarketIn) {
  socket.send(JSON.stringify(message));
}

function publish(topic: string, message: MarketIn) {
  for (const socket of subscribers.get(topic) ?? []) send(socket, message);
}

function priceOf(symbol: string) {
  let entry = prices.get(symbol);
  if (entry) return entry;
  const open = round(20 + Math.random() * 480);
  prices.set(symbol, (entry = { price: open, open }));
  return entry;
}

function quote(symbol: string): MarketIn {
  const { price, open } = priceOf(symbol);
  const change = ((price - open) / open) * 100;
  return { type: 'quote', topic: `quote:${symbol}`, quote: { symbol, price, change } };
}

function bookFor(symbol: string) {
  let book = books.get(symbol);
  if (book) return book;
  const mid = priceOf(symbol).price;
  const side = (sign: number) =>
    new Map(Array.from({ length: 20 }, (_, i) => [round(mid + sign * (i + 1) * 0.01), 100]));
  books.set(symbol, (book = { mid, seq: 0, bids: side(-1), asks: side(1), history: [] }));
  return book;
}

const round = (price: number) => Math.round(price * 100) / 100;
const levels = (side: Map<number, number>): Level[] => [...side];

function subscribe(socket: WebSocket, topic: string, after: number | undefined) {
  let sockets = subscribers.get(topic);
  if (!sockets) subscribers.set(topic, (sockets = new Set()));
  sockets.add(socket);

  const [kind, symbol = ''] = topic.split(':');
  if (kind === 'quote') return send(socket, quote(symbol));
  const book = bookFor(symbol);
  const oldest = book.history[0]?.seq ?? book.seq + 1;
  if (after === undefined || after < oldest - 1) {
    // Too far behind, or new: the whole book as of now.
    send(socket, {
      type: 'book',
      topic,
      seq: book.seq,
      bids: levels(book.bids),
      asks: levels(book.asks)
    });
  } else {
    for (const change of book.history) if (change.seq > after) send(socket, change);
  }
}

async function place(order: Order) {
  await sleep(30); // your matching engine
  return { orderId: randomUUID().slice(0, 8), filled: order.qty };
}

const wss = new WebSocketServer({ port: Number(process.env.PORT ?? 8080), path: '/live' });

wss.on('connection', socket => {
  socket.on('message', data => {
    if (String(data) === 'ping') return socket.send('pong');
    const message = JSON.parse(String(data)) as MarketOut; // validate it in production
    switch (message.type) {
      case 'subscribe':
        subscribe(socket, message.topic, message.after);
        break;
      case 'unsubscribe':
        subscribers.get(message.topic)?.delete(socket);
        break;
      case 'order': {
        const { id, clientOrderId, ...order } = message;
        if (order.qty * order.limit > 1_000_000) {
          send(socket, { type: 'rejected', id, reason: 'Orders are limited to 1,000,000.' });
          break;
        }
        // Each key is placed once. Asking again answers with the same result.
        let result = placed.get(clientOrderId);
        if (!result) placed.set(clientOrderId, (result = place(order)));
        void result.then(placement => send(socket, { type: 'placed', id, ...placement }));
        break;
      }
    }
  });
  socket.on('close', () => {
    for (const sockets of subscribers.values()) sockets.delete(socket);
  });
});

// Quotes: each symbol someone watches moves about five times a second.
setInterval(() => {
  for (const [topic, sockets] of subscribers) {
    if (!topic.startsWith('quote:') || !sockets.size || Math.random() < 0.5) continue;
    const entry = priceOf(topic.slice(6));
    entry.price = round(Math.max(1, entry.price * (1 + (Math.random() - 0.5) / 500)));
    publish(topic, quote(topic.slice(6)));
  }
}, 100);

// Books: a level of each watched book changes twenty times a second.
setInterval(() => {
  for (const [symbol, book] of books) {
    const topic = `book:${symbol}`;
    if (!subscribers.get(topic)?.size) continue;
    const bid = Math.random() < 0.5;
    const price = round(book.mid + (bid ? -1 : 1) * Math.ceil(Math.random() * 25) * 0.01);
    const size = Math.random() < 0.2 ? 0 : Math.ceil(Math.random() * 500);
    if (size) (bid ? book.bids : book.asks).set(price, size);
    else (bid ? book.bids : book.asks).delete(price);
    const change: Book['history'][number] = {
      type: 'level',
      topic,
      seq: ++book.seq,
      side: bid ? 'bid' : 'ask',
      price,
      size
    };
    book.history.push(change);
    if (book.history.length > 1000) book.history.shift();
    publish(topic, change);
  }
}, 50);

// live.ts, with MessagePack instead of JSON
import { decode, encode } from '@msgpack/msgpack';
import { createWebSocket } from 'react-websocket';
import type { MarketIn, MarketOut } from './protocol';

export const market = createWebSocket<MarketIn, MarketOut>('wss://market.example.com/live', {
  binaryType: 'arraybuffer',
  parse: data => decode(data as ArrayBuffer) as MarketIn,
  serialize: message => encode(message),
  heartbeat: { interval: 10_000, pong: 'pong', timeout: 3_000 }
});