Best Practices

Guidelines for building robust, efficient WebSocket clients including backpressure handling, subscription management, and error handling.

Guidelines for building robust, efficient WebSocket clients.

Fetch Markets First

Before making any API requests or WebSocket subscriptions, always fetch the /api/v1/markets endpoint to get the list of available exchanges and symbols.

Why This Matters

Exchange IDs: Use the id field from the exchanges array (e.g., binancef, bybitf). These are the identifiers required in all API requests.
Symbol format: Use the symbol field (e.g., btc/usd), not the exchange_ticker. The exchange_ticker (e.g., BTCUSDT) is for display purposes only.
Availability: The markets endpoint tells you which exchanges and symbols are actually available, preventing invalid requests.

Example

import WebSocket from 'ws';

type MarketsResponse = {
  exchanges: Array<{
    id: string;
    symbols: Array<{ symbol: string }>;
  }>;
};

const response = await fetch('https://eu-central-1.mmt.gg/api/v1/markets', {
  headers: { 'X-API-Key': process.env.MMT_API_KEY! }
});
const markets = (await response.json()) as MarketsResponse;

const validPairs = new Set<string>();
for (const exchange of markets.exchanges) {
  for (const sym of exchange.symbols) {
    // Use exchange.id + symbol from /markets for subscriptions.
    validPairs.add(`${exchange.id}:${sym.symbol}`);
  }
}

const subscription = {
  channel: 'candles',
  exchange: 'binancef',
  symbol: 'btc/usd',
  tf: '1m'
} as const;

if (!validPairs.has(`${subscription.exchange}:${subscription.symbol}`)) {
  throw new Error('Invalid exchange/symbol pair');
}

const ws = new WebSocket(
  `wss://eu-central-1.mmt.gg/api/v1/ws?api_key=${process.env.MMT_API_KEY}`
);

ws.on('open', () => {
  ws.send(JSON.stringify({ type: 'subscribe', ...subscription }));
});

Symbol vs Exchange Ticker

Always use symbol (e.g., btc/usd) in API requests. The exchange_ticker (e.g., BTCUSDT) is only for display purposes and will not work in API calls.

Handling Message Formats

Message frame type depends on format:

format=json (default): server sends WebSocket text frames (JSON strings)
format=cbor: server sends WebSocket binary frames (CBOR bytes)

Node.js

In Node.js with the ws library:

import WebSocket, { RawData } from 'ws';
import { decode as decodeCbor } from 'cbor-x';

const ws = new WebSocket(
  `wss://eu-central-1.mmt.gg/api/v1/ws?api_key=${process.env.MMT_API_KEY}`
);

function toBuffer(data: RawData): Buffer {
  if (Buffer.isBuffer(data)) return data;
  if (Array.isArray(data)) return Buffer.concat(data);
  return Buffer.from(data);
}

ws.on('message', (data: RawData, isBinary: boolean) => {
  const payload = toBuffer(data);

  if (isBinary) {
    // format=cbor: decode binary frames
    const msg = decodeCbor(payload);
    console.log(msg);
    return;
  }

  // format=json: parse text frames
  const msg = JSON.parse(payload.toString('utf8'));
  console.log(msg);
});

Performance

If you only need JSON, use the default format=json and parse text frames directly. Use format=cbor only when you explicitly want binary CBOR payloads.

Enable WebSocket Compression (Recommended)

The WebSocket server supports permessage-deflate to reduce bandwidth for high-volume streams.

In Node.js (ws), set perMessageDeflate: true when creating the connection.
Verify compression is negotiated by checking the WebSocket extensions after connect.
Keep handlers fast even with compression enabled; decompression still adds CPU overhead.

import { RawData } from 'ws';

const messageQueue: unknown[] = [];

// Good: fast handler, offload heavy work
ws.on('message', (data: RawData, isBinary: boolean) => {
  if (isBinary) return; // Decode CBOR separately when using format=cbor
  messageQueue.push(JSON.parse(data.toString()));
});

// Background processor handles expensive work
setInterval(() => processQueue(messageQueue), 50);

// Bad: Blocking in handler
ws.on('message', async (data: RawData, isBinary: boolean) => {
  if (isBinary) return;
  const msg = JSON.parse(data.toString());
  await updateDatabase(msg);      // Slow I/O
  recalculateIndicators(msg);     // CPU-intensive work
});

Choose appropriate timeframes. Don't subscribe to 1-second candles if you only need 1-minute data. Higher timeframes mean fewer messages:

Timeframe	Messages/hour
1s	3,600
15s	240
1m	60
5m	12

Use worker threads (or separate worker processes) for CPU-intensive processing.

Subscription Management

Track and manage your subscriptions to avoid waste and duplicates.

Track Subscriptions Client-Side

Maintain a set of active subscriptions:

import WebSocket from 'ws';

type StreamSubscription = {
  channel: string;
  exchange: string;
  symbol: string;
  tf?: string;
  bucket?: number;
};

class SubscriptionManager {
  private ws: WebSocket;
  private subscriptions = new Map<string, StreamSubscription>();

  constructor(ws: WebSocket) {
    this.ws = ws;
  }

  private key(sub: StreamSubscription): string {
    return [
      sub.channel,
      sub.exchange,
      sub.symbol,
      sub.tf ?? '',
      sub.bucket ?? ''
    ].join('|');
  }

  subscribe(sub: StreamSubscription): void {
    const key = this.key(sub);
    if (this.subscriptions.has(key)) return;

    this.ws.send(JSON.stringify({ type: 'subscribe', ...sub }));
    this.subscriptions.set(key, sub);
  }

  unsubscribe(sub: StreamSubscription): void {
    const key = this.key(sub);
    if (!this.subscriptions.has(key)) return;

    this.ws.send(JSON.stringify({ type: 'unsubscribe', ...sub }));
    this.subscriptions.delete(key);
  }

  unsubscribeAll(): void {
    for (const sub of this.subscriptions.values()) {
      this.ws.send(JSON.stringify({ type: 'unsubscribe', ...sub }));
    }
    this.subscriptions.clear();
  }

  resubscribeAll(): void {
    for (const sub of this.subscriptions.values()) {
      this.ws.send(JSON.stringify({ type: 'subscribe', ...sub }));
    }
  }
}

Clean Up on Disconnect

Always unsubscribe before closing the connection:

function disconnect(ws: WebSocket, subscriptionManager: SubscriptionManager): void {
  subscriptionManager.unsubscribeAll();
  ws.close(1000, 'Client disconnect');
}

Resubscribe on Reconnect

After reconnecting, restore your subscriptions:

function onReconnect(subscriptionManager: SubscriptionManager): void {
  subscriptionManager.resubscribeAll();
}

Error Handling

Build resilient clients that handle errors gracefully.

Parse All Messages Safely

import { RawData } from 'ws';

type WsMessage = { type: string; [key: string]: unknown };

ws.on('message', (data: RawData, isBinary: boolean) => {
  let msg: WsMessage;
  try {
    // Parse JSON frames (decode CBOR separately when using format=cbor).
    if (isBinary) return;
    msg = JSON.parse(data.toString()) as WsMessage;
  } catch (e) {
    console.error('Failed to parse message:', e);
    return; // Do not crash process on malformed payload
  }

  if (msg.type === 'error') {
    handleError(msg as ErrorMessage);
    return;
  }

  processMessage(msg);
});

Handle Error Messages

type ErrorMessage = {
  type: 'error';
  code: string;
  message: string;
};

function handleError(msg: ErrorMessage): void {
  console.error(`Server error [${msg.code}]: ${msg.message}`);

  switch (msg.code) {
    case 'ALREADY_SUBSCRIBED':
    case 'NOT_SUBSCRIBED':
      return;
    case 'MAX_SUBS_EXCEEDED':
      console.error('Reduce active subscriptions or use another allowed connection.');
      return;
    case 'INVALID_STREAM':
    case 'INVALID_PARAMS':
    case 'INVALID_EXCHANGE':
    case 'INVALID_SYMBOL':
    case 'INVALID_TIMEFRAME':
    case 'INVALID_BUCKET':
    case 'AGG_NOT_ALLOWED':
      console.error('Validate channel/exchange/symbol/tf/bucket against /markets and channel docs.');
      return;
    default:
      console.warn('Unhandled error code:', msg.code);
  }
}

Log for Debugging

Include context in your logs:

type LogLevel = 'info' | 'warn' | 'error';

function log(level: LogLevel, message: string, context: Record<string, unknown> = {}): void {
  const entry = {
    timestamp: new Date().toISOString(),
    level,
    message,
    ...context
  };
  console[level](JSON.stringify(entry));
}

// Usage
log('error', 'Subscription failed', {
  channel: 'trades',
  exchange: 'binancef',
  symbol: 'btc/usd',
  error: msg.message
});

Connection Management

Optimize connection usage for efficiency and reliability.

Use a Single Connection

One connection can handle multiple subscriptions. Avoid opening multiple connections:

import WebSocket from 'ws';

const ws = new WebSocket(
  `wss://eu-central-1.mmt.gg/api/v1/ws?api_key=${process.env.MMT_API_KEY}`
);

ws.on('open', () => {
  const subscriptions = [
    { channel: 'trades', exchange: 'binancef', symbol: 'btc/usd' },
    { channel: 'candles', exchange: 'binancef', symbol: 'btc/usd', tf: '1m' },
    { channel: 'depth', exchange: 'binancef', symbol: 'btc/usd' }
  ];

  for (const sub of subscriptions) {
    ws.send(JSON.stringify({ type: 'subscribe', ...sub }));
  }
});

// Bad: Multiple connections
const tradeWs = new WebSocket('wss://eu-central-1.mmt.gg/api/v1/ws?api_key=...');
const candleWs = new WebSocket('wss://eu-central-1.mmt.gg/api/v1/ws?api_key=...');
const depthWs = new WebSocket('wss://eu-central-1.mmt.gg/api/v1/ws?api_key=...');

Use a singleton or dependency injection:

// connection.ts
import WebSocket from 'ws';

let instance: WebSocket | null = null;

export function getConnection(): WebSocket {
  if (!instance) {
    instance = new WebSocket(
      `wss://eu-central-1.mmt.gg/api/v1/ws?api_key=${process.env.MMT_API_KEY}`
    );
  }
  return instance;
}

// market-streams.ts
import { getConnection } from './connection';

const ws = getConnection();
ws.on('open', () => {
  ws.send(
    JSON.stringify({
      type: 'subscribe',
      channel: 'trades',
      exchange: 'binancef',
      symbol: 'btc/usd'
    })
  );

  ws.send(
    JSON.stringify({
      type: 'subscribe',
      channel: 'candles',
      exchange: 'binancef',
      symbol: 'btc/usd',
      tf: '1m'
    })
  );
});

Connection Pooling

For high-throughput applications that need to distribute load:

import WebSocket from 'ws';

class ConnectionPool {
  private connections: WebSocket[] = [];
  private index = 0;

  constructor(url: string, size = 3) {
    for (let i = 0; i < size; i++) {
      this.connections.push(new WebSocket(url));
    }
  }

  getConnection(): WebSocket {
    const conn = this.connections[this.index];
    this.index = (this.index + 1) % this.connections.length;
    return conn;
  }
}

Message Processing

Handle high message volumes efficiently.

Use Message Queues

Buffer messages and process in batches:

class MessageProcessor {
  private queue: Array<{ channel: string; [key: string]: unknown }> = [];
  private processing = false;
  private handlers: Record<string, (messages: Array<{ channel: string; [key: string]: unknown }>) => Promise<void>> = {};

  enqueue(message: { channel: string; [key: string]: unknown }): void {
    this.queue.push(message);
    if (!this.processing) {
      void this.processQueue();
    }
  }

  async processQueue(): Promise<void> {
    this.processing = true;

    while (this.queue.length > 0) {
      // Process in batches of 100
      const batch = this.queue.splice(0, 100);
      await this.processBatch(batch);
    }

    this.processing = false;
  }

  async processBatch(messages: Array<{ channel: string; [key: string]: unknown }>): Promise<void> {
    // Group by channel for efficient processing
    const byChannel: Record<string, Array<{ channel: string; [key: string]: unknown }>> = {};
    for (const msg of messages) {
      if (!byChannel[msg.channel]) byChannel[msg.channel] = [];
      byChannel[msg.channel].push(msg);
    }

    // Process each channel
    for (const [channel, msgs] of Object.entries(byChannel)) {
      await this.handlers[channel]?.(msgs);
    }
  }
}

Batch Downstream Writes

Don't write to databases or message buses on every single message:

type Candle = { t: number; o: number; h: number; l: number; c: number };

class CandleWriter {
  private pending: Candle[] = [];
  private flushTimer?: NodeJS.Timeout;

  queue(candle: Candle): void {
    this.pending.push(candle);

    if (!this.flushTimer) {
      this.flushTimer = setTimeout(() => {
        void this.flush();
      }, 100);
    }
  }

  async flush(): Promise<void> {
    const batch = this.pending.splice(0, this.pending.length);
    if (batch.length === 0) return;

    this.flushTimer = undefined;
    await writeCandlesBatch(batch);
  }
}

Debounce High-Frequency Updates

For trades and orderbook updates:

function debounce<T extends unknown[]>(
  fn: (...args: T) => void,
  ms: number
): (...args: T) => void {
  let timeout: NodeJS.Timeout | undefined;
  return (...args: T) => {
    clearTimeout(timeout);
    timeout = setTimeout(() => fn(...args), ms);
  };
}

const updateOrderbook = debounce((depth) => {
  publishDepthToInternalBus(depth);
}, 100); // Publish at most every 100ms

Common Pitfalls

Avoid these frequent mistakes.

Not Handling Reconnection

Problem: Connection drops and client doesn't recover.

Solution: Implement automatic reconnection with backoff:

import WebSocket from 'ws';

class ReconnectingWebSocket {
  private ws?: WebSocket;
  private reconnectDelay = 1000;
  private readonly maxDelay = 30000;

  constructor(private readonly url: string) {
    this.connect();
  }

  connect(): void {
    this.ws = new WebSocket(this.url);

    this.ws.on('open', () => {
      this.reconnectDelay = 1000; // Reset on success
      this.onReconnect?.();
    });

    this.ws.on('close', (code) => {
      if (code !== 1000) {
        setTimeout(() => this.connect(), this.reconnectDelay);
        this.reconnectDelay = Math.min(
          this.reconnectDelay * 2,
          this.maxDelay
        );
      }
    });

    this.ws.on('error', () => {
      // Close event follows in most cases.
    });
  }

  onReconnect?(): void;
}

Subscribing Multiple Times

Problem: Same channel subscribed multiple times wastes resources.

Solution: Track subscriptions client-side (see Subscription Management above).

Blocking on Message Processing

Problem: Slow processing backs up the connection.

Solution: Use async handlers and message queues:

import { RawData } from 'ws';

// Non-blocking handler
ws.on('message', (data: RawData, isBinary: boolean) => {
  if (isBinary) return;
  messageQueue.enqueue(JSON.parse(data.toString()));
});

Not Unsubscribing Before Disconnect

Problem: Server continues processing for dead connection.

Solution: Always unsubscribe first during shutdown:

process.on('SIGTERM', () => {
  subscriptionManager.unsubscribeAll();
  ws.close(1000, 'Process shutdown');
});

Ignoring Error Messages

Problem: Client doesn't know why things aren't working.

Solution: Always handle error message types:

if (msg.type === 'error') {
  console.error('Server error:', msg);
  // Take appropriate action
}

Testing

Verify your client handles edge cases.

Test Connection Drops

Simulate network failures:

import WebSocket from 'ws';

// Test helper
function simulateDisconnect(ws: WebSocket): void {
  ws.terminate(); // Abrupt close to simulate network drop
}

// Test
it('reconnects after disconnect', async () => {
  const client = new WebSocketClient(url);
  await client.connect();

  simulateDisconnect(client.ws);

  await waitFor(() => client.isConnected());
  expect(client.isConnected()).toBe(true);
});

Test Slow Client Scenarios

Verify graceful handling when processing lags:

it('handles slow processing', async () => {
  const client = new WebSocketClient(url);

  // Slow handler
  client.onMessage = async (_msg) => {
    await sleep(100); // Simulate slow processing
  };

  await client.connect();
  await client.subscribe({
    channel: 'trades',
    exchange: 'binancef',
    symbol: 'btc/usd'
  });

  // Should not crash, may disconnect
  await sleep(5000);
});

Test Reconnection Logic

Verify subscriptions restore after reconnect:

it('restores subscriptions on reconnect', async () => {
  const client = new WebSocketClient(url);
  await client.connect();
  await client.subscribe({
    channel: 'candles',
    exchange: 'binancef',
    symbol: 'btc/usd',
    tf: '1m'
  });

  simulateDisconnect(client.ws);
  await waitFor(() => client.isConnected());

  // Verify subscription restored
  expect(
    client.hasSubscription({
      channel: 'candles',
      exchange: 'binancef',
      symbol: 'btc/usd',
      tf: '1m'
    })
  ).toBe(true);
});

Summary

Area	Key Points
Markets First	Fetch `/api/v1/markets` before any requests; use `symbol` not `exchange_ticker`
Message Format	JSON uses text frames; CBOR uses binary frames
Backpressure	Process fast, use appropriate timeframes, offload work
Subscriptions	Track client-side, avoid duplicates, clean up
Errors	Parse safely, handle error types, log context
Connections	Single connection, share across components
Processing	Queue messages, batch writes, debounce downstream work
Testing	Test drops, slow clients, reconnection

Best Practices

Fetch Markets First

Why This Matters

Example

Handling Message Formats

Node.js

Enable WebSocket Compression (Recommended)

Backpressure Handling

Why This Matters

Recommendations

Subscription Management

Track Subscriptions Client-Side

Clean Up on Disconnect

Resubscribe on Reconnect

Error Handling

Parse All Messages Safely

Handle Error Messages

Log for Debugging

Connection Management

Use a Single Connection

Connection Pooling

Message Processing

Use Message Queues

Batch Downstream Writes

Debounce High-Frequency Updates

Common Pitfalls

Not Handling Reconnection

Subscribing Multiple Times

Blocking on Message Processing

Not Unsubscribing Before Disconnect

Ignoring Error Messages

Testing

Test Connection Drops

Test Slow Client Scenarios

Test Reconnection Logic

Summary

On this page