Real‑Time Coinbase OHLCV Streaming with Node.js Native WebSocket & TypeScript

Coinbase (and Coinbase Pro) provide a WebSocket feed to stream live market data. In this tutorial we focus on the ticker channel that returns real‑time pricing data (such as price, bid, ask, volume, etc.) for selected trading pairs. Using Node’s built‑in WebSocket API and TypeScript, we’ll create a robust solution that handles errors, reconnections, and graceful shutdowns.

Note:
The Coinbase WebSocket endpoint is:
wss://ws-feed.pro.coinbase.com
Unlike Binance’s combined stream that wraps data in a {stream, data} object, Coinbase requires you to send a subscription JSON message and emits messages with properties like type, product_id, and ticker data fields.
In this tutorial, we subscribe to the ticker channel for three products: BTC-USD, ETH-USD, and BNB-USD.

1. Overview

When subscribing to Coinbase’s ticker channel, you send a JSON-formatted subscription message such as:

{
  "type": "subscribe",
  "channels": [
    {
      "name": "ticker",
      "product_ids": ["BTC-USD", "ETH-USD", "BNB-USD"]
    }
  ]
}

Once subscribed, the server will send messages with a structure similar to:

{
  "type": "ticker",
  "sequence": 123456789,
  "product_id": "BTC-USD",
  "price": "50000.12",
  "open_24h": "49000.00",
  "volume_24h": "1200.3456",
  "low_24h": "48000.00",
  "high_24h": "51000.00",
  "side": "buy",         // "buy" or "sell"
  "time": "2021-12-01T12:34:56.789Z",
  "trade_id": 12345,
  "last_size": "0.01"
}

This data provides real‑time updates about the market status for the product. We will use this data to simulate processing “ticker” updates (analogous to OHLCV details).

2. Environment Setup

Prerequisites:
- Node.js v20+ (to utilize the stable built‑in WebSocket API)
- TypeScript

Project Initialization:

npm init -y
npm install --save-dev typescript @types/node
npx tsc --init

No external WebSocket library is needed because we are using Node’s built‑in WebSocket API.

3. Implementation in TypeScript

Below is the complete code sample. This example will:

Establish a WebSocket connection to Coinbase.
Send a subscription message to the ticker channel for BTC-USD, ETH-USD, and BNB-USD.
Listen for incoming messages and parse the JSON data.
Handle error and close events with automatic reconnection.
Gracefully shut down when the process terminates.

// List of Coinbase products to subscribe to.
const products = ["BTC-USD", "ETH-USD", "BNB-USD"];
const WS_URL = "wss://ws-feed.pro.coinbase.com";

let ws: WebSocket | null = null;
let reconnectTimeout: NodeJS.Timeout | null = null;

/**
 * Connect (or reconnect) to Coinbase’s WebSocket feed.
 */
function connect() {
  ws = new WebSocket(WS_URL);

  ws.onopen = () => {
    console.log("Connected to Coinbase WebSocket API");
    // Send the subscription message upon connection.
    const subscribeMessage = {
      type: "subscribe",
      channels: [
        {
          name: "ticker",
          product_ids: products,
        },
      ],
    };
    ws?.send(JSON.stringify(subscribeMessage));
  };

  ws.onmessage = (event: MessageEvent) => {
    try {
      const message = JSON.parse(event.data.toString());
      // Coinbase emits many types of messages (e.g. subscriptions, heartbeats).
      // We are interested in "ticker" messages which provide real‑time pricing updates.
      if (message.type === "ticker") {
        const {
          product_id,
          price,
          open_24h,
          volume_24h,
          low_24h,
          high_24h,
          side,
          time,
          trade_id,
          last_size,
        } = message;
        console.log(
          `Ticker update for ${product_id} at ${time}: Price=${price}, Open=${open_24h}, ` +
            `High=${high_24h}, Low=${low_24h}, Volume=${volume_24h}, Last Size=${last_size}, Side=${side}`
        );
        // Process the ticker data as needed.
      } else {
        // Optionally log non-ticker messages or handle them accordingly.
        // console.log("Received message:", message);
      }
    } catch (err) {
      console.error("Error parsing incoming message:", err);
    }
  };

  ws.onerror = (event: Event) => {
    console.error("WebSocket error occurred:", event);
    cleanupAndReconnect();
  };

  ws.onclose = (event: CloseEvent) => {
    console.warn(
      `WebSocket closed. Code: ${event.code}, Reason: ${event.reason}`
    );
    // For Coinbase, a normal closure should be code 1000.
    if (event.code !== 1000) {
      console.log("Attempting to reconnect in 5 seconds...");
      reconnectTimeout = setTimeout(connect, 5000);
    }
  };
}

/**
 * Clean up the current connection and schedule a reconnection.
 */
function cleanupAndReconnect() {
  if (ws) {
    // Remove event handlers.
    ws.onopen = null;
    ws.onmessage = null;
    ws.onerror = null;
    ws.onclose = null;
    // Close the connection if it is OPEN or CONNECTING.
    if (ws.readyState === WebSocket.OPEN || ws.readyState === WebSocket.CONNECTING) {
      ws.close(1001, "Reconnecting"); // Code 1001: Going Away.
    }
  }
  if (reconnectTimeout) {
    clearTimeout(reconnectTimeout);
  }
  reconnectTimeout = setTimeout(connect, 5000);
}

connect();

/**
 * Gracefully shut down the WebSocket connection on process termination.
 */
function shutdown() {
  console.log("Shutting down gracefully...");
  if (reconnectTimeout) {
    clearTimeout(reconnectTimeout);
  }
  if (ws && ws.readyState === WebSocket.OPEN) {
    // Send an unsubscribe message if needed before closing.
    const unsubscribeMessage = {
      type: "unsubscribe",
      channels: [
        {
          name: "ticker",
          product_ids: products,
        },
      ],
    };
    ws.send(JSON.stringify(unsubscribeMessage));
    ws.close(1000, "Process terminated"); // Code 1000: Normal Closure.
  }
  process.exit(0);
}

process.on("SIGINT", shutdown);
process.on("SIGTERM", shutdown);

// Optional: Dynamic Subscription Management
// If you wish to subscribe or unsubscribe from additional products dynamically, you can send:
//
// // To subscribe to a new product, for example, LTC-USD:
// ws.send(
//   JSON.stringify({
//     type: "subscribe",
//     channels: [{ name: "ticker", product_ids: ["LTC-USD"] }]
//   })
// );
//
// // To unsubscribe from a product:
// ws.send(
//   JSON.stringify({
//     type: "unsubscribe",
//     channels: [{ name: "ticker", product_ids: ["LTC-USD"] }]
//   })
// );

4. Explanation

Connection Setup:

We create a WebSocket connection to wss://ws-feed.pro.coinbase.com using Node’s native API. Once the connection is open, we send a JSON-formatted subscription message to the ticker channel for our products. Coinbase then emits real‑time ticker messages for the subscribed products.
Event Handlers:
- onopen: Logs that the connection is established and sends the subscription message.
- onmessage: Parses incoming messages. The handler specifically checks for messages of type "ticker" which contain information like price, 24‑hour high, low, volume, etc.
- onerror & onclose: Manage errors/disconnects. In case of non‑normal closure (code not equal to 1000), we schedule a reconnection after 5 seconds.
Graceful Shutdown: We hook into SIGINT and SIGTERM to gracefully unsubscribe (if desired) and close the connection before terminating the process.
Dynamic Subscription Management (Optional): Coinbase supports dynamic subscriptions using JSON messages. This facility allows you to add or remove subscriptions at runtime without reopening the connection.

5. Summary

In this tutorial, we demonstrated how to:

Connect to the Coinbase WebSocket feed using Node’s native WebSocket API in a TypeScript environment.
Subscribe to the ticker channel for selected products (BTC-USD, ETH-USD, and BNB-USD).
Process incoming messages that include real‑time market data.
Handle errors, auto‑reconnect on unexpected closures, and clean up on process termination.
Optionally perform dynamic subscription management.

This robust setup forms a solid basis for any application that requires real‑time market data from Coinbase. Happy coding!