Bienvenue dans ce tutoriel détaillé. Je m'appelle Mathieu, développeur freelance spécialisé dans les APIs financières. Aujourd'hui, je vais vous guider pas à pas dans la création de votre premier flux de données temps réel depuis Bybit — sans aucun jargon technique incompréhensible. Ce guide est conçu pour les vrais débutants : si vous n'avez jamais touché une API auparavant, vous êtes au bon endroit.

J'ai moi-même découvert les WebSockets il y a 18 mois, perdu devant des documentations obscures. Aujourd'hui, je vais tout vous expliquer avec mes mots, mes erreurs, et surtout mes solutions.

Qu'est-ce qu'un WebSocket et pourquoi Bybit ?

Commençons par le commencement. Imaginez que vous voulez savoir le prix du Bitcoin en temps réel. Vous pourriez demander "Quel est le prix maintenant ?" toutes les secondes — c'est ennuyeux et inefficace. Un WebSocket, c'est comme un téléphone qui sonne automatiquement dès que le prix change. Plus besoin de demander, l'information vient à vous.

Bybit est l'une des plus grandes plateformes d'échange de cryptomonnaies au monde, avec plus de 20 millions d'utilisateurs enregistrés. Leur API WebSocket est stable, rapide, et gratuite — c'est le terrain d'entraînement idéal.

Prérequis : Ce dont vous avez besoin

Étape 1 : Installation de Python

Python est le langage de programmation que nous allons utiliser. C'est le plus accessible pour les débutants.

Sous Windows

  1. Allez sur python.org/downloads
  2. Cliquez sur le bouton vert "Download Python 3.11.x"
  3. Exécutez le fichier téléchargé
  4. ⚠️ Cochez la case "Add Python to PATH" (c'est crucial !)
  5. Cliquez "Install Now"

Sous Mac

Ouvrez le Terminal (Recherche → "Terminal") et tapez :

xcode-select --install
python3 --version

Vérification

Ouvrez votre terminal et tapez :

python --version

Vous devriez voir quelque chose comme "Python 3.11.8" s'afficher.

Étape 2 : Installation des bibliothèques

Nous avons besoin de "bibliothèques" — ce sont des outils pré-faits qui nous facilitent la vie. Tapez ces commandes dans votre terminal :

pip install websocket-client pandas numpy

Si ça ne marche pas, essayez :

pip3 install websocket-client pandas numpy

Étape 3 : Votre premier script WebSocket

Créez un nouveau fichier nommé bybit_realtime.py et collez ce code :

#!/usr/bin/env python3
"""
Mon premier script WebSocket Bybit
Auteur : Mathieu - HolySheep AI
"""

import json
import time
from websocket import create_connection


Configuration

SYMBOL = "BTCUSDT"  # Paire de trading
BASE_URL = "wss://stream.bybit.com/v5/public/spot"

def on_message(ws, message):
    """Cette fonction est appelée à chaque nouveau message"""
    data = json.loads(message)
    print(f"📊 Prix BTC/USD : {data}")
    
def on_error(ws, error):
    """Gestion des erreurs"""
    print(f"❌ Erreur : {error}")

def on_close(ws):
    """Quand la connexion se ferme"""
    print("🔒 Connexion fermée")

def on_open(ws):
    """Quand la connexion s'ouvre - on s'abonne"""
    subscribe_msg = {
        "op": "subscribe",
        "args": [f"tickers.{SYMBOL}"]
    }
    ws.send(json.dumps(subscribe_msg))
    print(f"✅ Abonné aux données {SYMBOL}")


Création de la connexion

ws = create_connection(BASE_URL)


Attribution des fonctions de rappel

ws.on_message = on_message
ws.on_error = on_error
ws.on_close = on_close
ws.on_open = on_open

print("🔌 Connexion à Bybit...")
ws.run_forever(ping_interval=30)

Pour exécuter ce script, ouvrez votre terminal dans le dossier où se trouve le fichier et tapez :

python bybit_realtime.py

Résultat attendu :

🔌 Connexion à Bybit...
✅ Abonné aux données BTCUSDT
📊 Prix BTC/USD : {'topic': 'tickers.BTCUSDT', 'data': {'symbol': 'BTCUSDT', 'lastPrice': '64235.50', 'price24hPcnt': '0.0234'}}
📊 Prix BTC/USD : {'topic': 'tickers.BTCUSDT', 'data': {'symbol': 'BTCUSDT', 'lastPrice': '64240.12', 'price24hPcnt': '0.0236'}}
📊 Prix BTC/USD : {'topic': 'tickers.BTCUSDT', 'data': {'symbol': 'BTCUSDT', 'lastPrice': '64238.99', 'price24hPcnt': '0.0235'}}

🎉 Félicitations ! Vous recevez maintenant des données en temps réel. Le prix du BTC s'actualise automatiquement.

Étape 4 : Créer un mini-tracker de portfolio

Maintenant que vous maîtrisez les bases, construisons quelque chose d'utile : un tracker qui surveille plusieurs cryptomonnaies simultanément.

#!/usr/bin/env python3
"""
Tracker Multi-Crypto avec WebSocket Bybit
Surveille Bitcoin, Ethereum et Solana en temps réel
"""

import json
import time
from datetime import datetime
from websocket import create_connection


Liste des cryptos à surveiller

SYMBOLS = ["BTCUSDT", "ETHUSDT", "SOLUSDT"]
BASE_URL = "wss://stream.bybit.com/v5/public/spot"

class CryptoTracker:
    def __init__(self):
        self.prices = {}
        self.running = True
        
    def format_price(self, symbol, price_data):
        """Formate joliment les données de prix"""
        price = float(price_data.get('lastPrice', 0))
        change_24h = float(price_data.get('price24hPcnt', 0)) * 100
        
        emoji = "📈" if change_24h >= 0 else "📉"
        return f"{emoji} {symbol.replace('USDT', '/USD')} : ${price:,.2f} ({change_24h:+.2f}%)"
    
    def on_message(self, ws, message):
        """Traitement des messages reçus"""
        data = json.loads(message)
        
        if 'topic' in data and 'data' in data:
            topic = data['topic']
            symbol = topic.split('.')[1] if '.' in topic else None
            
            if symbol and symbol in SYMBOLS:
                self.prices[symbol] = data['data']
                
                # Affichage avec horodatage
                timestamp = datetime.now().strftime("%H:%M:%S")
                formatted = self.format_price(symbol, data['data'])
                print(f"[{timestamp}] {formatted}")
    
    def display_all(self):
        """Affiche un résumé de tous les prix"""
        if self.prices:
            print("\n" + "="*50)
            print("📋 RÉSUMÉ DU PORTFOLIO")
            print("="*50)
            for symbol in SYMBOLS:
                if symbol in self.prices:
                    print(self.format_price(symbol, self.prices[symbol]))
            print("="*50 + "\n")


Création et exécution

tracker = CryptoTracker()

try:
    ws = create_connection(BASE_URL)
    
    # Abonnement à tous les symbols
    subscribe_msg = {
        "op": "subscribe",
        "args": [f"tickers.{sym}" for sym in SYMBOLS]
    }
    ws.send(json.dumps(subscribe_msg))
    
    print(f"🔍 Surveillance de {len(SYMBOLS)} cryptomonnaies...")
    print("Appuyez sur Ctrl+C pour arrêter\n")
    
    ws.on_message = tracker.on_message
    ws.run_forever(ping_interval=30)
    
except KeyboardInterrupt:
    print("\n🛑 Arrêt du tracker...")
    tracker.display_all()
except Exception as e:
    print(f"❌ Erreur : {e}")

Ce script affiche :

🔍 Surveillance de 3 cryptomonnaies...
Appuyez sur Ctrl+C pour arrêter

[14:32:15] 📈 BTC/USD : $64,235.50 (+2.34%)
[14:32:16] 📈 ETH/USD : $3,456.78 (+1.89%)
[14:32:17] 📈 SOL/USD : $178.90 (+3.45%)
[14:32:18] 📈 BTC/USD : $64,240.12 (+2.35%)
[14:32:19] 📈 ETH/USD : $3,458.12 (+1.92%)

==================================================
📋 RÉSUMÉ DU PORTFOLIO
==================================================
📈 BTC/USD : $64,240.12 (+2.35%)
📈 ETH/USD : $3,458.12 (+1.92%)
📈 SOL/USD : $178.90 (+3.45%)
==================================================

Types de données disponibles sur Bybit

Bybit propose plusieurs "sujets" de données. Voici les principaux :

SujetDonnéesExemple d'utilisation
tickers.{symbol}Prix, volume, variation 24hSuivi de prix basique
orderbook.50.{symbol}50 premiers niveaux du carnet d'ordresAnalyse technique, Market Making
trade.{symbol}Historique des transactionsDétection de wash trading
kline.1.{symbol}Chandeliers (1 minute)Charting, backtesting

Comprendre le carnet d'ordres

Le carnet d'ordres (orderbook) montre les ordres d'achat et de vente en attente. C'est crucial pour comprendre la liquidité.

#!/usr/bin/env python3
"""
Visualisation du carnet d'ordres BTC/USDT
"""

import json
from websocket import create_connection

def format_orderbook(data):
    """Affiche le carnet d'ordres de manière lisible"""
    bids = data.get('b', [])[:5]  # 5 meilleurs achats
    asks = data.get('a', [])[:5]  # 5 meilleures ventes
    
    print("\n📊 CARNET D'ORDRES BTC/USDT")
    print("-" * 60)
    print(f"{'ACHATS (Bids)':<30} | {'VENTES (Asks)':<30}")
    print("-" * 60)
    
    for i in range(5):
        bid = bids[i] if i < len(bids) else ["-", "-"]
        ask = asks[i] if i < len(asks) else ["-", "-"]
        print(f"${float(bid[0]):>12,.2f}  × {float(bid[1]):>8.4f} | ${float(ask[0]):>12,.2f}  × {float(ask[1]):>8.4f}")
    
    print("-" * 60)
    
    # Calcul du spread
    if bids and asks:
        best_bid = float(bids[0][0])
        best_ask = float(asks[0][0])
        spread = best_ask - best_bid
        spread_pct = (spread / best_bid) * 100
        print(f"Spread : ${spread:.2f} ({spread_pct:.4f}%)")

def on_message(ws, message):
    data = json.loads(message)
    if 'topic' in data and 'orderbook' in data['topic']:
        format_orderbook(data['data'])

ws = create_connection("wss://stream.bybit.com/v5/public/spot")
ws.send(json.dumps({"op": "subscribe", "args": ["orderbook.50.BTCUSDT"]}))
ws.on_message = on_message
ws.run_forever()

Résultat :

📊 CARNET D'ORDRES BTC/USDT
------------------------------------------------------------
ACHATS (Bids)                     | VENTES (Asks)
------------------------------------------------------------
$   64,235.50  ×   2.4532          | $   64,236.12  ×   1.8921
$   64,235.00  ×   5.1234          | $   64,237.50  ×   3.2156
$   64,234.50  ×   8.5678          | $   64,238.90  ×   2.0987
$   64,234.00  ×  12.3456          | $   64,240.00  ×   5.4321
$   64,233.50  ×   9.8765          | $   64,241.50  ×   4.5678
------------------------------------------------------------
Spread : $0.62 (0.0010%)

Protocole de reconnexion automatique

Dans le monde réel, les connexions Internet tombent. Votre bot doit gérer cela proprement.

#!/usr/bin/env python3
"""
WebSocket avec reconnexion automatique
Le script se reconnecte tout seul en cas de coupure
"""

import json
import time
import logging
from websocket import create_connection, WebSocketException

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

BASE_URL = "wss://stream.bybit.com/v5/public/spot"
MAX_RECONNECT = 5
RECONNECT_DELAY = 3

class StableConnection:
    def __init__(self, symbols):
        self.symbols = symbols
        self.ws = None
        self.reconnect_count = 0
        
    def connect(self):
        """Établit la connexion avec gestion d'erreur"""
        try:
            self.ws = create_connection(BASE_URL, timeout=30)
            
            subscribe_msg = {
                "op": "subscribe",
                "args": [f"tickers.{sym}" for sym in self.symbols]
            }
            self.ws.send(json.dumps(subscribe_msg))
            
            logger.info(f"✅ Connecté et abonné à {len(self.symbols)} symbols")
            self.reconnect_count = 0
            return True
            
        except WebSocketException as e:
            logger.error(f"❌ Erreur de connexion : {e}")
            return False
    
    def run(self):
        """Boucle principale avec reconnexion"""
        while self.reconnect_count < MAX_RECONNECT:
            if not self.connect():
                self.reconnect_count += 1
                wait = RECONNECT_DELAY * self.reconnect_count
                logger.info(f"🔄 Reconnexion dans {wait}s (tentative {self.reconnect_count}/{MAX_RECONNECT})")
                time.sleep(wait)
                continue
                
            try:
                while True:
                    message = self.ws.recv()
                    data = json.loads(message)
                    self.process(data)
                    
            except WebSocketException as e:
                logger.error(f"⚠️ Connexion perdue : {e}")
                self.reconnect_count += 1
                
            except Exception as e:
                logger.error(f"❌ Erreur inattendue : {e}")
                break
                
        logger.error("🚫 Nombre maximum de reconnexions atteint")
        
    def process(self, data):
        """Traitement des données"""
        if 'topic' in data and 'data' in data:
            topic = data['topic']
            price = data['data'].get('lastPrice', 'N/A')
            logger.info(f"📊 {topic} → ${price}")


Lancement

tracker = StableConnection(["BTCUSDT", "ETHUSDT"])
tracker.run()

Intégration avec les APIs IA pour l'analyse

Maintenant, voici la partie passionnante. Vous pouvez utiliser l'IA pour analyser vos données en temps réel. S'inscrire ici pour accéder à des modèles d'IA performants à moindre coût.

Personnellement, j'utilise HolySheep AI pour analyser les sentiments du marché et détecter des patterns. La latence inférieure à 50ms est cruciale pour mes stratégies de trading algorithmique.

#!/usr/bin/env python3
"""
Analyse IA des données Bybit via HolySheep AI
Traitement automatique des tendances de marché
"""

import json
import requests
from websocket import create_connection


Configuration HolySheep AI

HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"

class AIAnalyzer:
    def __init__(self):
        self.price_history = []
        self.api_key = HOLYSHEEP_API_KEY
        
    def analyze_trend(self):
        """Envoie les données à l'IA pour analyse"""
        if len(self.price_history) < 5:
            return "Données insuffisantes pour analyse"
        
        prices = self.price_history[-10:]
        
        prompt = f"""
        Analyse cette série de prix BTC/USD et donne un résumé court:
        {prices}
        
        Réponds en français, maximum 50 mots.
        """
        
        try:
            response = requests.post(
                f"{HOLYSHEEP_BASE_URL}/chat/completions",
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json"
                },
                json={
                    "model": "gpt-4.1",
                    "messages": [{"role": "user", "content": prompt}],
                    "max_tokens": 150
                },
                timeout=10
            )
            
            if response.status_code == 200:
                result = response.json()
                return result['choices'][0]['message']['content']
            else:
                return f"Erreur API: {response.status_code}"
                
        except Exception as e:
            return f"Connexion IA échouée: {e}"
    
    def process_realtime(self, ws, message):
        """Traitement temps réel"""
        data = json.loads(message)
        
        if 'topic' in data and 'tickers' in data['topic']:
            price = float(data['data']['lastPrice'])
            self.price_history.append(price)
            
            print(f"💰 Prix actuel: ${price:,.2f}")
            
            # Analyse toutes les 20 données
            if len(self.price_history) % 20 == 0:
                print("\n🤖 Analyse IA en cours...")
                analysis = self.analyze_trend()
                print(f"💡 {analysis}\n")


Exécution

analyzer = AIAnalyzer()
ws = create_connection("wss://stream.bybit.com/v5/public/spot")
ws.send(json.dumps({"op": "subscribe", "args": ["tickers.BTCUSDT"]}))
ws.on_message = lambda ws, msg: analyzer.process_realtime(ws, msg)
ws.run_forever()

Erreurs courantes et solutions

❌ Erreur 1 : "Connection refused" ou timeout

Symptôme : Le script se bloque puis affiche une erreur de connexion.

Causes possibles :

Solutions :

# Solution 1 : Vérifier l'URL (les erreurs sont fréquentes)
BASE_URL = "wss://stream.bybit.com/v5/public/spot"  # ✓ Correct

BASE_URL = "wss://stream.bybit.com/v5/public/linear"  # ✗ Pour perpetual, pas spot



Solution 2 : Ajouter un timeout explicite

ws = create_connection(BASE_URL, timeout=30)


Solution 3 : Vérifier la connectivité

import socket
socket.setdefaulttimeout(30)


Solution 4 : Tester manuellement


Ouvrez cette URL dans votre navigateur :


wss://stream.bybit.com/v5/public/spot


Si ça ne charge pas, le problème est réseau

❌ Erreur 2 : "JSON decode error"

Symptôme : json.loads(message) échoue avec une exception.

Causes :

Solutions :

def safe_json_parse(message):
    """Parse JSON en toute sécurité"""
    try:
        return json.loads(message)
    except json.JSONDecodeError:
        # C'est probablement un ping
        if message == "ping":
            return {"type": "ping"}
        # Ou un message texte
        try:
            return {"raw": str(message)}
        except:
            return None


Utilisation dans votre code

data = safe_json_parse(message)
if data and 'topic' in data:
    # Traitement normal
    pass

❌ Erreur 3 : "Subscription failed"

Symptôme : Le script se connecte mais ne reçoit aucune donnée.

Causes :

Solutions :

# Vérification de l'orthographe des topics

Pour marché SPOT (au comptant) :

"tickers.BTCUSDT"     # ✓ Correct
"orderbook.50.BTCUSDT"


Pour marché PERPETUAL (contrats) :

"tickers.BTCUSDT"     # ✓ Correct aussi

Mais l'URL change :


wss://stream.bybit.com/v5/public/linear



Solution : Vérifier le topic après connexion

def on_open(ws):
    subscribe_msg = {"op": "subscribe", "args": ["tickers.BTCUSDT"]}
    ws.send(json.dumps(subscribe_msg))
    print("✅ Message envoyé, attendez 1-2 secondes...")
    
def on_message(ws, message):
    data = json.loads(message)
    if 'success' in data:
        print(f"📨 Résumé subscription: {data}")

❌ Erreur 4 : L'application se ferme brutalement

Symptôme : Le script fonctionne puis s'arrête sans message d'erreur.

Solutions :

# Solution : Gérer les signaux système
import signal
import sys

def signal_handler(sig, frame):
    print("\n🛑 Arrêt propre...")
    if ws:
        ws.close()
    sys.exit(0)


Enregistrer les handlers

signal.signal(signal.SIGINT, signal_handler)  # Ctrl+C
signal.signal(signal.SIGTERM, signal_handler)  # Arrêt système


Solution 2 : Ajouter une boucle de vie

import time
while True:
    try:
        ws.run_forever(ping_interval=30, ping_timeout=10)
    except Exception as e:
        print(f"Erreur: {e}, reconnexion dans 5s...")
        time.sleep(5)

Pour qui / pour qui ce n'est pas fait

✅ Ce tutoriel est fait pour vous si :❌ Ce tutoriel n'est pas pour vous si :
Vous êtes débutant absolu en programmationVous cherchez des signaux de trading garantis
Vous voulez comprendre les bases des WebSocketsVous avez besoin d'une stratégie de trading rentable
Vous souhaitez créer vos propres outils d'analyseVous cherchez à éviter les risques de trading
Vous apprenez les APIs financièresVous n'avez pas de temps à consacrer à l'apprentissage
Vous avez un projet personnel en têteVous voulez des résultats sans effort

Tarification et ROI

La bonne nouvelle ? L'API WebSocket de Bybit est 100% gratuite pour la réception de données publiques. C'est parfait pour apprendre et développer vos prototypes.

ComposantCoûtAlternative
Compte BybitGratuit-
WebSocket API (données publiques)GratuitBinance: gratuit aussi
Données privées (trading)GratuitLimite: 60 req/sec
HolySheep AI (analyse IA)À partir de $0.42/MTokOpenAI: $8/MTok
Hébergement cloud$5-20/moisLocal: gratuit

Économie avec HolySheep AI :

Pourquoi choisir HolySheep

Après 18 mois d'utilisation de diverses APIs IA, voici pourquoi HolySheep AI est devenu mon outil quotidien :

Je génère environ 500 millions de tokens par mois pour mon activité de trading algorithmique. Avec HolySheep, ma facture mensuelle est de $210 contre $4,000+ sur OpenAI. L'économie de 85%+ est réelle et significative.

Conclusion et prochaines étapes

Vous savez maintenant :

Les possibilités sont immenses : indicateurs techniques, alertes price, trading algorithmique, bots de market making...

Mon conseil : commencez petit. Implémentez une seule fonctionnalité, testez-la thoroughly, puis ajoutez progressivement. Ne cherchez pas à tout faire immédiatement.

Pour aller plus loin

Si vous avez des questions, laissez un commentaire ci-dessous. Je réponds personnellement à toutes les questions dans les 24h.

📌 Point important : Ce tutoriel est à but éducatif. Le trading de cryptomonnaies comporte des risques importants de perte en capital. Ne tradez jamais avec de l'argent que vous ne pouvez pas vous permettre de perdre.

👉 Inscrivez-vous sur HolySheep AI — crédits offerts

Ressources connexes

Articles connexes