¿Cómo suscribirse a cotizaciones por WebSocket en Binance? Código de nivel de producción para flujos individuales y combinados

La mejor solución para obtener cotizaciones en tiempo real mediante WebSocket en Binance es utilizar el punto de conexión de flujos combinados wss://stream.binance.com:9443/stream?streams=. Una sola conexión permite suscribirse hasta a 1024 flujos, con una latencia de mensaje inferior a 50 ms y un consumo de peso casi nulo. Este artículo proporciona el código completo en Python y Node.js para flujos individuales, flujos combinados, suscripciones dinámicas, emparejamiento local de profundidad y reconexión por desconexión, todo listo para ejecutarse. Antes de comenzar, si aún no tiene una cuenta de Binance, puede visitar el sitio oficial de Binance para ver el acceso oficial o completar la apertura de cuenta a través del enlace de registro gratuito.

I. Resumen de puntos de conexión (Endpoints) de WebSocket

Límites de conexión: un máximo de 300 conexiones WebSocket por IP, cada conexión puede suscribirse a un máximo de 1024 flujos. La conexión se desconecta automáticamente cada 24 horas, por lo que el cliente debe implementar la reconexión.

II. Lista de tipos de Stream

Binance ofrece 11 tipos principales de flujos de cotización:

Reglas de nomenclatura: en minúsculas y separadas por el símbolo @, por ejemplo btcusdt@ticker en lugar de BTCUSDT@ticker.

III. Suscripción a flujo individual con Python asyncio

import asyncio, json
import websockets

async def subscribe_ticker(symbol: str):
    url = f"wss://stream.binance.com:9443/ws/{symbol.lower()}@ticker"
    async with websockets.connect(url, ping_interval=180) as ws:
        async for message in ws:
            data = json.loads(message)
            print(f"{data['s']}: Último precio {data['c']}, "
                  f"Variación 24h {data['P']}%, "
                  f"Volumen {data['v']}")

asyncio.run(subscribe_ticker("BTCUSDT"))

Ejemplo de salida:

BTCUSDT: Último precio 68420.12, Variación 24h 2.35%, Volumen 45231.2
BTCUSDT: Último precio 68421.00, Variación 24h 2.36%, Volumen 45231.8

IV. Suscripción a múltiples pares de trading con flujos combinados

Una conexión para suscribirse al ticker de 10 pares de trading + líneas K de 5 pares:

import asyncio, json
import websockets

STREAMS = [
    "btcusdt@ticker", "ethusdt@ticker", "bnbusdt@ticker",
    "solusdt@ticker", "xrpusdt@ticker", "adausdt@ticker",
    "dogeusdt@ticker", "maticusdt@ticker", "dotusdt@ticker",
    "avaxusdt@ticker",
    "btcusdt@kline_1m", "ethusdt@kline_1m",
    "bnbusdt@kline_1m", "solusdt@kline_1m", "xrpusdt@kline_1m"
]

async def combined_stream():
    streams = "/".join(STREAMS)
    url = f"wss://stream.binance.com:9443/stream?streams={streams}"
    async with websockets.connect(url, ping_interval=180) as ws:
        async for message in ws:
            payload = json.loads(message)
            stream = payload["stream"]
            data = payload["data"]
            if "@ticker" in stream:
                print(f"[{stream}] {data['s']}: {data['c']}")
            elif "@kline" in stream:
                k = data["k"]
                if k["x"]:  # Solo imprimir cuando la línea K se cierra
                    print(f"[{stream}] {k['s']} {k['i']} "
                          f"O:{k['o']} H:{k['h']} L:{k['l']} C:{k['c']}")

asyncio.run(combined_stream())

V. Suscripción/Cancelación dinámica (Método SUBSCRIBE)

Una vez establecida la conexión, se pueden agregar o eliminar suscripciones dinámicamente mediante mensajes JSON:

import asyncio, json
import websockets

async def dynamic_sub():
    url = "wss://stream.binance.com:9443/ws"
    async with websockets.connect(url) as ws:
        # Suscribirse a bookTicker de BTC y ETH
        await ws.send(json.dumps({
            "method": "SUBSCRIBE",
            "params": ["btcusdt@bookTicker", "ethusdt@bookTicker"],
            "id": 1
        }))

        count = 0
        async for msg in ws:
            data = json.loads(msg)
            if "result" in data:
                print(f"Confirmación de suscripción: {data}")
                continue
            count += 1
            print(f"{data['s']} Compra 1 {data['b']} Venta 1 {data['a']}")

            # Tras recibir 50 mensajes, cancelar suscripción de ETH y mantener solo BTC
            if count == 50:
                await ws.send(json.dumps({
                    "method": "UNSUBSCRIBE",
                    "params": ["ethusdt@bookTicker"],
                    "id": 2
                }))
                print("Suscripción de ETH cancelada")

asyncio.run(dynamic_sub())

VI. Fusión incremental de profundidad (Mantenimiento local del libro de órdenes)

Las estrategias de alta frecuencia suelen requerir un libro de órdenes a nivel de emparejamiento local. El proceso es:

1. Suscribirse al flujo diferencial @depth@100ms y colocarlo en un búfer.
2. Llamada REST GET /api/v3/depth?symbol=X&limit=1000 para obtener una instantánea (snapshot).
3. Descartar los mensajes en el búfer donde u < lastUpdateId.
4. Aplicar individualmente los mensajes donde U <= lastUpdateId+1 <= u.
5. Aplicar directamente los mensajes posteriores; pu debe ser igual al u del mensaje anterior.

import asyncio, json, aiohttp
import websockets
from sortedcontainers import SortedDict

class OrderBook:
    def __init__(self, symbol):
        self.symbol = symbol.lower()
        self.bids = SortedDict()  # Precio -> Cantidad
        self.asks = SortedDict()
        self.last_update_id = 0
        self.buffer = []

    def apply_update(self, bids, asks):
        for price, qty in bids:
            price, qty = float(price), float(qty)
            if qty == 0:
                self.bids.pop(price, None)
            else:
                self.bids[price] = qty
        for price, qty in asks:
            price, qty = float(price), float(qty)
            if qty == 0:
                self.asks.pop(price, None)
            else:
                self.asks[price] = qty

    async def load_snapshot(self):
        async with aiohttp.ClientSession() as s:
            url = f"https://api.binance.com/api/v3/depth?symbol={self.symbol.upper()}&limit=1000"
            async with s.get(url) as r:
                snap = await r.json()
                self.last_update_id = snap["lastUpdateId"]
                self.apply_update(snap["bids"], snap["asks"])

    async def run(self):
        url = f"wss://stream.binance.com:9443/ws/{self.symbol}@depth@100ms"
        async with websockets.connect(url) as ws:
            # Paso 1: Recibir diferenciales por un tiempo
            asyncio.create_task(self._collect(ws))
            await asyncio.sleep(1)
            # Paso 2: Obtener instantánea
            await self.load_snapshot()
            # Paso 3: Aplicar diferenciales válidos del búfer
            for diff in self.buffer:
                if diff["u"] < self.last_update_id:
                    continue
                self.apply_update(diff["b"], diff["a"])
                self.last_update_id = diff["u"]
            self.buffer = None
            # Paso 4: Aplicar directamente los siguientes
            async for msg in ws:
                diff = json.loads(msg)
                self.apply_update(diff["b"], diff["a"])
                self.last_update_id = diff["u"]
                best_bid = self.bids.keys()[-1] if self.bids else 0
                best_ask = self.asks.keys()[0] if self.asks else 0
                print(f"Compra 1 {best_bid} Venta 1 {best_ask} Dif {best_ask-best_bid:.2f}")

    async def _collect(self, ws):
        if self.buffer is None:
            return
        async for msg in ws:
            if self.buffer is None:
                break
            self.buffer.append(json.loads(msg))

ob = OrderBook("BTCUSDT")
asyncio.run(ob.run())

VII. Reconexión por desconexión y latido (Heartbeat)

El WebSocket de Binance requiere que el cliente responda a un pong en un plazo de 3 minutos, y la conexión se desconecta obligatoriamente cada 24 horas:

import asyncio, json, websockets, logging

async def resilient_ws(url, on_message, max_retries=1000):
    retry = 0
    while retry < max_retries:
        try:
            async with websockets.connect(
                url,
                ping_interval=180,  # Enviar un ping cada 3 minutos
                ping_timeout=10,
                close_timeout=5
            ) as ws:
                retry = 0  # Restablecer reintentos al conectar con éxito
                async for message in ws:
                    await on_message(json.loads(message))
        except (websockets.ConnectionClosed, asyncio.TimeoutError) as e:
            retry += 1
            wait = min(2 ** retry, 60)  # Retroceso exponencial, máximo 60s
            logging.warning(f"WS desconectado {e}, reintentando en {wait}s (intento {retry})")
            await asyncio.sleep(wait)

VIII. Ejemplo de WebSocket en Node.js

const WebSocket = require('ws');

const streams = ['btcusdt@bookTicker', 'ethusdt@bookTicker'].join('/');
const url = `wss://stream.binance.com:9443/stream?streams=${streams}`;

function connect() {
  const ws = new WebSocket(url);
  ws.on('open', () => console.log('Conexión exitosa'));
  ws.on('message', (raw) => {
    const { stream, data } = JSON.parse(raw);
    console.log(`${data.s} Compra 1 ${data.b} Venta 1 ${data.a}`);
  });
  ws.on('close', () => {
    console.log('Desconectado, reintentando en 3s');
    setTimeout(connect, 3000);
  });
  ws.on('error', (err) => console.error('Error', err.message));
}

connect();

IX. Preguntas frecuentes (FAQ)

P1: ¿El WebSocket también consume peso?

R: Establecer la conexión consume 2 de peso; el envío posterior de mensajes no consume peso en absoluto. Sin embargo, si el número de conexiones supera las 300 por IP, estas serán rechazadas.

P2: ¿Por qué no recibo ningún mensaje después de suscribirme?

R: Hay tres razones comunes: 1) el nombre del flujo (stream) debe estar en minúsculas (btcusdt, no BTCUSDT); 2) el par de trading está mal escrito (por ejemplo, usar BTC-USDT en lugar de BTCUSDT); 3) el mensaje de suscripción debe realizarse en el punto de conexión wss://.../ws y no en el de /stream.

P3: ¿Cuál es la diferencia entre @depth y @depth@100ms?

R: @depth envía una instantánea completa (20 niveles) cada 1000 ms; @depth@100ms envía un diferencial (incremental) cada 100 ms. Para el mantenimiento local del libro de órdenes, debe usar @100ms; para visualización, @depth es suficiente.

P4: ¿Se pueden realizar órdenes a través de WebSocket?

R: Sí. Binance ofrece una API de WebSocket (wss://ws-api.binance.com:443/ws-api/v3) para realizar órdenes firmadas, con una latencia entre 30 y 50 ms menor que REST. Sin embargo, el ecosistema no es tan maduro y la mayoría de las estrategias siguen usando REST para ordenar y WS para monitorear.

P5: ¿Cómo realizar una transición fluida tras la desconexión automática de 24 horas?

R: Esquema de rotación de doble conexión: a las 23 horas, establezca una nueva conexión B; una vez que B reciba mensajes de forma estable, cierre la conexión antigua A para evitar la pérdida instantánea de datos. Los marcos de nivel de producción como python-binance ya incluyen esta lógica de forma nativa.

Después de revisar la solución de WebSocket, regrese a la Navegación de categorías para consultar otros tutoriales de SDK en la categoría «Integración API».