Comment s'abonner aux flux de marché via WebSocket Binance ? Code de niveau production

La solution optimale pour obtenir les données de marché en temps réel via WebSocket sur Binance est d'utiliser le point de terminaison de flux combiné wss://stream.binance.com:9443/stream?streams=. Une seule connexion peut s'abonner jusqu'à 1024 flux, avec une latence inférieure à 50ms et une consommation de poids quasi nulle. Cet article fournit le code complet en Python et Node.js pour les flux simples, combinés, les abonnements dynamiques, la maintenance du carnet d'ordres local et la reconnexion automatique. Avant de commencer, si vous n'avez pas encore de compte Binance, vous pouvez consulter le portail officiel Binance ou utiliser le lien de création de compte gratuite.

I. Aperçu des points de terminaison WebSocket

Limites de connexion : Un maximum de 300 connexions WebSocket par IP, 1024 flux par connexion, avec une déconnexion automatique toutes les 24 heures. Le client doit impérativement implémenter une logique de reconnexion.

II. Liste des types de flux (Stream)

Binance propose 11 types principaux de flux de marché :

Règle de nommage : Minuscules + séparateur @, par exemple btcusdt@ticker au lieu de BTCUSDT@ticker.

III. Abonnement à un flux simple avec 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']}: Dernier prix {data['c']}, "
                  f"Variation 24h {data['P']}%, "
                  f"Volume {data['v']}")

asyncio.run(subscribe_ticker("BTCUSDT"))

Exemple de sortie :

BTCUSDT: Dernier prix 68420.12, Variation 24h 2.35%, Volume 45231.2
BTCUSDT: Dernier prix 68421.00, Variation 24h 2.36%, Volume 45231.8

IV. Flux combiné pour plusieurs paires de trading

Une seule connexion s'abonnant au ticker de 10 paires + aux lignes K de 5 paires :

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"]:  # Imprimer seulement à la fermeture de la bougie
                    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. Abonnement/Désabonnement dynamique (Méthode SUBSCRIBE)

Une fois la connexion établie, vous pouvez ajouter ou supprimer des abonnements via des messages JSON :

import asyncio, json
import websockets

async def dynamic_sub():
    url = "wss://stream.binance.com:9443/ws"
    async with websockets.connect(url) as ws:
        # S'abonner au bookTicker de BTC et 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"Confirmation d'abonnement: {data}")
                continue
            count += 1
            print(f"{data['s']} Bid {data['b']} Ask {data['a']}")

            # Se désabonner de ETH après 50 messages, garder seulement BTC
            if count == 50:
                await ws.send(json.dumps({
                    "method": "UNSUBSCRIBE",
                    "params": ["ethusdt@bookTicker"],
                    "id": 2
                }))
                print("Abonnement ETH annulé")

asyncio.run(dynamic_sub())

VI. Fusion incrémentielle de profondeur (Maintenance locale de l'Order Book)

Les stratégies haute fréquence nécessitent souvent un carnet d'ordres local. Le processus est :

1. S'abonner au flux différentiel @depth@100ms, mettre en tampon.
2. Appel REST GET /api/v3/depth?symbol=X&limit=1000 pour obtenir l'instantané.
3. Jeter les messages du tampon où u < lastUpdateId.
4. Appliquer les messages où U <= lastUpdateId+1 <= u.
5. Appliquer les messages suivants directement, pu doit être égal au u du message précédent.

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

class OrderBook:
    def __init__(self, symbol):
        self.symbol = symbol.lower()
        self.bids = SortedDict()  # Prix -> Quantité
        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:
            # Étape 1 : Recevoir les différentiels pendant un moment
            asyncio.create_task(self._collect(ws))
            await asyncio.sleep(1)
            # Étape 2 : Charger l'instantané
            await self.load_snapshot()
            # Étape 3 : Appliquer les différentiels valides du tampon
            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
            # Étape 4 : Appliquer les suivants directement
            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"Bid {best_bid} Ask {best_ask} Spread {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. Reconnexion et Heartbeat

Le WebSocket Binance exige que le client réponde au pong dans les 3 minutes, et la connexion est fermée de force toutes les 24 heures :

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,  # Envoyer ping toutes les 3 minutes
                ping_timeout=10,
                close_timeout=5
            ) as ws:
                retry = 0  # Réinitialiser après connexion réussie
                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)  # Exponential backoff, max 60s
            logging.warning(f"WS déconnecté {e}, reconnexion dans {wait}s (Tentative {retry})")
            await asyncio.sleep(wait)

VIII. Exemple Node.js WebSocket

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('Connexion réussie'));
  ws.on('message', (raw) => {
    const { stream, data } = JSON.parse(raw);
    console.log(`${data.s} Bid ${data.b} Ask ${data.a}`);
  });
  ws.on('close', () => {
    console.log('Déconnecté, reconnexion dans 3s');
    setTimeout(connect, 3000);
  });
  ws.on('error', (err) => console.error('Erreur', err.message));
}

connect();

IX. Foire aux questions (FAQ)

Q1 : Le WebSocket consomme-t-il également du poids d'API ?

R : L'établissement de la connexion coûte 2 de poids ; les messages poussés par la suite ne consomment absolument aucun poids. Cependant, si le nombre de connexions dépasse 300 par IP, elles seront rejetées.

Q2 : Pourquoi ne reçois-je aucun message après l'abonnement ?

R : Trois raisons courantes : 1) le nom du flux doit être en minuscules (btcusdt et non BTCUSDT) ; 2) la paire est mal écrite (par exemple BTC-USDT au lieu de BTCUSDT) ; 3) les messages d'abonnement doivent être envoyés au point de terminaison wss://.../ws et non /stream.

Q3 : Quelle est la différence entre @depth et @depth@100ms ?

R : @depth pousse un instantané complet (20 niveaux) toutes les 1000ms ; @depth@100ms pousse un différentiel (incrémentiel) toutes les 100ms. Pour maintenir un carnet d'ordres local, il faut utiliser @100ms, alors que @depth suffit pour l'affichage visuel.

Q4 : Peut-on passer des ordres via WebSocket ?

R : Oui. Binance propose une API WebSocket (wss://ws-api.binance.com:443/ws-api/v3) pour passer des ordres signés, avec une latence de 30-50ms inférieure au REST. Cependant, l'écosystème étant moins mature, la plupart des stratégies utilisent encore le REST pour les ordres et le WS pour l'écoute.

Q5 : Comment assurer une transition fluide lors de la déconnexion après 24 heures ?

R : Stratégie de double connexion roulante : après 23 heures, établissez une nouvelle connexion B. Une fois que B reçoit des messages de manière stable, fermez l'ancienne connexion A pour éviter toute perte de données instantanée. Les frameworks de niveau production comme python-binance intègrent déjà cette logique.

Après avoir exploré la solution WebSocket, retournez à la navigation par catégorie pour consulter d'autres tutoriels SDK dans la catégorie « Intégration API ».