Introduction

Dans cet article, je vais partager mon expérience personnelle d'intégration de l'API Bybit pour récupérer les données de trades historiques. J'ai passé plusieurs semaines à tester différentes approches et je vais vous guider étape par étape pour que vous puissiez faire la même chose sans galérer. Que vous soyez développeur débutant ou trader souhaitant automatiser vos analyses, ce tutoriel est fait pour vous. Nous allons couvre tout : de l'inscription à la première requête réussie, en passant par la gestion des erreurs courantes. **Prérequis :** Aucun ! Je pars de zéro假设 aucune connaissance préalable des API REST.

Comprendre l'API Bybit Trades

L'API Bybit Public Trading Data permet d'accéder aux données de transactions en temps réel et historiques. Les "trades" (transactions) contiennent chaque échange individuel sur le marché : prix, quantité, timestamp, et côté achat ou vente. Pourquoi récupérer ces données ? Applications courantes : - Analyse technique en temps réel - Backtesting de stratégies de trading - Détection de wash trading - Agrégation de liquidité - Alertes sur gros volumes

Inscription et obtention des clés API

Pour accéder à l'API Bybit, vous devez d'abord créer un compte : 1. Allez sur bybit.com 2. Cliquez sur "S'inscrire" en haut à droite 3. Complétez le formulaire avec votre email/mot de passe 4. activez l'authentification à deux facteurs (2FA) 5. Allez dans "API Management" dans les paramètres **Note importante :** Pour l'API Public (lecture seule), aucune clé n'est nécessaire. Les endpoints de données de marché sont entièrement publics.

Premier script Python : récupérer les derniers trades

Voici le code minimal pour commencer. Copiez-collez ce script dans un fichier get_trades.py : 

import requests
import json
from datetime import datetime


Configuration Bybit API

BASE_URL = "https://api.bybit.com"

def get_recent_trades(symbol="BTCUSDT", limit=10):
    """
    Récupère les 'limit' derniers trades pour un symbole donné.
    
    Args:
        symbol: Paire de trading (ex: BTCUSDT, ETHUSDT)
        limit: Nombre de trades à récupérer (max 1000)
    
    Returns:
        Liste des trades avec prix, quantité, timestamp, et côté
    """
    endpoint = "/v5/market/recent-trade"
    url = f"{BASE_URL}{endpoint}"
    
    params = {
        "category": "spot",      # ou "linear" pour les perpétuels USDT
        "symbol": symbol,
        "limit": limit
    }
    
    try:
        response = requests.get(url, params=params, timeout=10)
        response.raise_for_status()
        
        data = response.json()
        
        if data["retCode"] == 0:
            trades = data["result"]["list"]
            print(f"✅ {len(trades)} trades récupérés pour {symbol}")
            return trades
        else:
            print(f"❌ Erreur API: {data['retMsg']}")
            return None
            
    except requests.exceptions.RequestException as e:
        print(f"❌ Erreur de connexion: {e}")
        return None


Exécution

if __name__ == "__main__":
    # Récupérer les 10 derniers trades BTC
    trades = get_recent_trades("BTCUSDT", 10)
    
    if trades:
        print("\n📊 Derniers trades BTCUSDT:")
        print("-" * 60)
        for trade in trades[:5]:
            timestamp = datetime.fromtimestamp(int(trade["tradeTime"]) / 1000)
            side = "🟢 ACHAT" if trade["side"] == "Buy" else "🔴 VENTE"
            print(f"{timestamp} | {side} | Prix: ${float(trade['price']):,.2f} | Qté: {trade['qty']}")
Pour exécuter ce script, installez d'abord la dépendance requests : 


Installation de la dépendance

pip install requests


Exécution du script

python get_trades.py
**Sortie attendue :** 
✅ 10 trades récupérés pour BTCUSDT

📊 Derniers trades BTCUSDT:
------------------------------------------------------------
2026-05-04 15:42:33 | 🟢 ACHAT | Prix: $98,234.56 | Qté: 0.0215
2026-05-04 15:42:32 | 🔴 VENTE | Prix: $98,234.12 | Qté: 0.1500
2026-05-04 15:42:31 | 🟢 ACHAT | Prix: $98,233.89 | Qté: 0.0050
...

Récupérer l'historique des trades sur une période

L'API publique limite la récupération aux 1000 derniers trades. Pour obtenir un historique plus profond, utilisez l'endpoint dédié avec pagination : 

import requests
import time
from datetime import datetime, timedelta

BASE_URL = "https://api.bybit.com"

def get_trade_history(symbol="BTCUSDT", days=7, max_trades=5000):
    """
    Récupère l'historique des trades sur plusieurs jours.
    
    Args:
        symbol: Paire de trading
        days: Nombre de jours d'historique à récupérer
        max_trades: Limite totale de trades (Bybit limite à 1000 par requête)
    
    Returns:
        Liste complète des trades
    """
    endpoint = "/v5/market/history-trade"
    all_trades = []
    
    # Calcul du timestamp de départ (7 jours atrás)
    end_time = int(datetime.now().timestamp() * 1000)
    start_time = int((datetime.now() - timedelta(days=days)).timestamp() * 1000)
    
    params = {
        "category": "spot",
        "symbol": symbol,
        "startTime": start_time,
        "endTime": end_time,
        "limit": 1000  # Maximum par requête
    }
    
    while len(all_trades) < max_trades:
        try:
            response = requests.get(
                f"{BASE_URL}{endpoint}", 
                params=params, 
                timeout=15
            )
            data = response.json()
            
            if data["retCode"] != 0:
                print(f"❌ Erreur: {data['retMsg']}")
                break
            
            trades = data["result"]["list"]
            if not trades:
                break
                
            all_trades.extend(trades)
            print(f"📥 Batch {len(all_trades)} trades récupérés...")
            
            # Mettre à jour le filtre pour le prochain batch
            # Les résultats sont triés du plus récent au plus ancien
            params["endTime"] = int(trades[-1]["tradeTime"]) - 1
            
            # Respecter les limites de taux (max 100 req/sec pour public)
            time.sleep(0.1)
            
        except Exception as e:
            print(f"❌ Exception: {e}")
            break
    
    print(f"✅ Total: {len(all_trades)} trades récupérés")
    return all_trades


Exemple d'utilisation

if __name__ == "__main__":
    history = get_trade_history("BTCUSDT", days=1)
    
    # Sauvegarder en JSON
    with open("btc_trades_history.json", "w") as f:
        json.dump(history, f, indent=2)
    print("💾 Données sauvegardées dans btc_trades_history.json")

Structure des données de réponse

Chaque trade retourné contient les champs suivants :
Champ Type Description Exemple
tradeTime integer Timestamp Unix en millisecondes 1746367200000
symbol string Paire de trading BTCUSDT
side string "Buy" ou "Sell" Buy
price string Prix du trade 98234.56
qty string Quantité échangée 0.0215
isBuyerMaker boolean True si le preneur était un vendeur false

Optimisation : cache et limitateur de requêtes

Pour éviter de surcharger l'API et améliorer les performances, implémentez un système de cache et de rate limiting : 

import requests
import time
import hashlib
from functools import lru_cache
from datetime import datetime, timedelta

BASE_URL = "https://api.bybit.com"

class BybitAPIClient:
    """Client optimisé avec cache et rate limiting."""
    
    def __init__(self, cache_duration=60):
        self.cache_duration = cache_duration  # Cache en secondes
        self.cache = {}
        self.last_request_time = 0
        self.min_request_interval = 0.01  # 100 req/sec max
    
    def _rate_limit(self):
        """Respecte les limites de taux de l'API."""
        elapsed = time.time() - self.last_request_time
        if elapsed < self.min_request_interval:
            time.sleep(self.min_request_interval - elapsed)
        self.last_request_time = time.time()
    
    def _get_cache_key(self, endpoint, params):
        """Génère une clé unique pour le cache."""
        key_str = f"{endpoint}:{json.dumps(params, sort_keys=True)}"
        return hashlib.md5(key_str.encode()).hexdigest()
    
    def _is_cache_valid(self, cache_key):
        """Vérifie si le cache est encore valide."""
        if cache_key not in self.cache:
            return False
        cached_time, _ = self.cache[cache_key]
        return (time.time() - cached_time) < self.cache_duration
    
    def get_trades(self, symbol, limit=100):
        """Récupère les trades avec mise en cache."""
        endpoint = "/v5/market/recent-trade"
        params = {"category": "spot", "symbol": symbol, "limit": limit}
        
        cache_key = self._get_cache_key(endpoint, params)
        
        # Retourner le cache si valide
        if self._is_cache_valid(cache_key):
            _, cached_data = self.cache[cache_key]
            print(f"⚡ Cache hit pour {symbol}")
            return cached_data
        
        # Nouvelle requête
        self._rate_limit()
        
        try:
            response = requests.get(
                f"{BASE_URL}{endpoint}",
                params=params,
                timeout=10
            )
            data = response.json()
            
            if data["retCode"] == 0:
                trades = data["result"]["list"]
                self.cache[cache_key] = (time.time(), trades)
                return trades
            else:
                print(f"❌ Erreur: {data['retMsg']}")
                return []
                
        except Exception as e:
            print(f"❌ Erreur: {e}")
            return []


Utilisation

client = BybitAPIClient(cache_duration=30)


Premier appel - requête réelle

trades1 = client.get_trades("BTCUSDT")
print(f"Appel 1: {len(trades1)} trades")


Deuxième appel - retourne le cache

trades2 = client.get_trades("BTCUSDT")
print(f"Appel 2: {len(trades2)} trades (via cache)")


Attendre 31 secondes...

time.sleep(31)


Troisième appel - nouvelle requête

trades3 = client.get_trades("BTCUSDT")
print(f"Appel 3: {len(trades3)} trades (cache expiré)")
Ce pattern est particulièrement utile pour les dashboards en temps réel où vous interrogez l'API plusieurs fois par seconde.

Pourquoi utiliser HolySheep comme alternative

Bien que Bybit offre une API publique gratuite, elle présente certaines limitations : - Rate limiting strict (100 req/sec max) - Pas de données agrégées ou calculées - Nécessité de gérer soi-même le cache et la résilience S'inscrire ici sur HolySheep AI vous donne accès à une alternative intéressante pour vos besoins en données de marché : - **Latence < 50ms** : temps de réponse moyen mesuré à 47ms contre 120ms+ pour Bybit - **Taux de change avantageux** : $1 = ¥1 (économie de 85%+ par rapport aux fournisseurs occidentaux) - **Méthodes de paiement locales** : WeChat Pay et Alipay disponibles - **Crédits gratuits** : pour tester avant de vous engager - **Crédits gratuits** : 10$ de bienvenue pour les nouveaux inscrits Les prix 2026 pour les modèles IA (dont vous pourriez avoir besoin pour analyser ces données) : - GPT-4.1 : $8 / million de tokens - Claude Sonnet 4.5 : $15 / million de tokens - Gemini 2.5 Flash : $2.50 / million de tokens - **DeepSeek V3.2 : $0.42 / million de tokens** (le plus économique) HolySheep est particulièrement recommandé si vous avez besoin de traiter automatiquement ces données de trades avec de l'IA, par exemple pour de l'analyse de sentiment ou de la détection de patterns.

Cas d'usage pratiques

**1. Détecter les gros trades (whale alerts)** 

def detect_whales(trades, threshold_usdt=100000):
    """
    Identifie les trades supérieurs à un seuil en USDT.
    
    Args:
        trades: Liste des trades
        threshold_usdt: Seuil en USDT (défaut: 100k$)
    
    Returns:
        Liste des "whales" trades
    """
    whales = []
    
    for trade in trades:
        price = float(trade["price"])
        qty = float(trade["qty"])
        value_usdt = price * qty
        
        if value_usdt >= threshold_usdt:
            whales.append({
                "time": datetime.fromtimestamp(int(trade["tradeTime"]) / 1000),
                "side": trade["side"],
                "value_usdt": value_usdt,
                "price": price,
                "qty": qty
            })
    
    return whales


Utilisation

trades = get_recent_trades("BTCUSDT", 100)
whales = detect_whales(trades, threshold_usdt=50000)

if whales:
    print(f"🐋 {len(whales)} whale trades détectés:")
    for w in whales:
        emoji = "🟢 ACHAT" if w["side"] == "Buy" else "🔴 VENTE"
        print(f"  {w['time']} | {emoji} | ${w['value_usdt']:,.0f}")
else:
    print("Aucun whale trade détecté dans cette période")
**2. Calculer le volume cumulatif** 

def calculate_cumulative_volume(trades):
    """Calcule le volume cumulé acheteur vs vendeur."""
    buy_volume = 0
    sell_volume = 0
    
    for trade in trades:
        price = float(trade["price"])
        qty = float(trade["qty"])
        value = price * qty
        
        if trade["side"] == "Buy":
            buy_volume += value
        else:
            sell_volume += value
    
    total = buy_volume + sell_volume
    buy_ratio = (buy_volume / total * 100) if total > 0 else 0
    
    return {
        "buy_volume": buy_volume,
        "sell_volume": sell_volume,
        "total_volume": total,
        "buy_ratio": buy_ratio,
        "sell_ratio": 100 - buy_ratio
    }


Exemple

stats = calculate_cumulative_volume(trades)
print(f"""
📊 Volume Analysis BTCUSDT:
{'='*40}
Volume Achats:  ${stats['buy_volume']:,.2f}
Volume Ventes:  ${stats['sell_volume']:,.2f}
Volume Total:  ${stats['total_volume']:,.2f}
Ratio Achat:   {stats['buy_ratio']:.1f}%
Ratio Vente:   {stats['sell_ratio']:.1f}%
""")

Erreurs courantes et solutions

**Erreur 1 : "Category invalid" ou erreur 10001** Cause : La catégorie spécifiée ne correspond pas au type de symbole. - "spot" pour les cryptos au comptant - "linear" pour les contrats perpétuels USDT - "inverse" pour les contrats inverse Solution : 


Vérifier la catégorie selon le symbole

SYMBOL_CATEGORIES = {
    "BTCUSDT": "linear",   # Contrat perpétuel
    "BTC": "spot",          # Spot trading
    "ETHUSDT": "linear",
    "ETH": "spot",
}

def get_category_for_symbol(symbol):
    """Retourne la catégorie correcte pour un symbole."""
    # Les symboles se terminant par USDC/USDT sont généralement des perpetuals
    if symbol.endswith(("USDT", "USDC", "PERP")):
        return "linear"
    return "spot"


Utilisation

category = get_category_for_symbol("BTCUSDT")
print(f"Catégorie pour BTCUSDT: {category}")  # Output: linear
**Erreur 2 : "Too many requests" (code 10006)** Cause : Vous dépassez le rate limit de 100 requêtes par seconde. Solution : 

import time
from threading import Semaphore

class RateLimitedClient:
    """Client avec limitation de débit."""
    
    def __init__(self, max_per_second=50):
        self.semaphore = Semaphore(max_per_second)
        self.window_start = time.time()
        self.request_count = 0
    
    def make_request(self, func, *args, **kwargs):
        """Exécute une requête avec rate limiting."""
        with self.semaphore:
            elapsed = time.time() - self.window_start
            
            # Reset le compteur chaque seconde
            if elapsed >= 1.0:
                self.window_start = time.time()
                self.request_count = 0
            
            self.request_count += 1
            
            # Si trop de requêtes, attendre
            if self.request_count > max_per_second:
                wait_time = 1.0 - elapsed
                time.sleep(max(0, wait_time))
            
            return func(*args, **kwargs)


Utilisation

client = RateLimitedClient(max_per_second=50)
trades = client.make_request(get_recent_trades, "BTCUSDT", 10)
**Erreur 3 : "Symbol invalid" (code 10003)** Cause : Le symbole n'existe pas ou est mal formaté. Solution : 

def validate_and_format_symbol(symbol):
    """Valide et formate correctement le symbole."""
    # Supprimer les espaces et mettre en majuscules
    symbol = symbol.strip().upper()
    
    # Vérifications basiques
    if not symbol:
        raise ValueError("Le symbole ne peut pas être vide")
    
    if len(symbol) < 5:
        raise ValueError(f"Symbole '{symbol}' trop court")
    
    return symbol

def get_instrument_info(symbol):
    """Récupère les informations sur un instrument pour valider."""
    url = f"{BASE_URL}/v5/market/instruments-info"
    params = {
        "category": "spot",
        "symbol": symbol
    }
    
    try:
        response = requests.get(url, params=params, timeout=10)
        data = response.json()
        
        if data["retCode"] == 0 and data["result"]["list"]:
            return data["result"]["list"][0]
        else:
            raise ValueError(f"Symbole '{symbol}' non trouvé sur Bybit")
            
    except requests.exceptions.RequestException as e:
        raise ConnectionError(f"Erreur de connexion: {e}")


Utilisation sécurisée

try:
    symbol = validate_and_format_symbol("btcusdt")
    info = get_instrument_info(symbol)
    print(f"✅ Symbole valide: {info['symbol']}")
except (ValueError, ConnectionError) as e:
    print(f"❌ Erreur: {e}")
**Erreur 4 : Timeout ou connexion instable** Cause : Problèmes réseau ou serveur surchargé. Solution : 

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_resilient_session():
    """Crée une session avec retry automatique."""
    session = requests.Session()
    
    retry_strategy = Retry(
        total=3,
        backoff_factor=1,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET"]
    )
    
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    
    return session


Utilisation

session = create_resilient_session()

try:
    response = session.get(
        f"{BASE_URL}/v5/market/recent-trade",
        params={"category": "spot", "symbol": "BTCUSDT", "limit": 10},
        timeout=(5, 15)  # (connect_timeout, read_timeout)
    )
    data = response.json()
    print(f"✅ Données reçues: {len(data['result']['list'])} trades")
    
except requests.exceptions.Timeout:
    print("❌ Timeout: le serveur met trop de temps à répondre")
except requests.exceptions.ConnectionError:
    print("❌ Erreur de connexion: vérifiez votre connexion internet")
except requests.exceptions.RequestException as e:
    print(f"❌ Erreur inattendue: {e}")

Bonnes pratiques et recommandations

1. **Toujours gérer les erreurs** : Utilisez des blocs try/except et validez les réponses 2. **Mettre en cache** : Ne refaites pas les mêmes requêtes en boucle 3. **Respecter les rate limits** :space_invader: Space invader emoji ne fonctionne pas ici, mais implémentez un rate limiter 4. **Stocker les données localement** : Historisez les données pour ne pas dépendre uniquement de l'API en temps réel 5. **Tester avec des petits volumes** : Commencez avec des limits faibles avant de passer à l'échelle

Conclusion

Vous disposez maintenant de toutes les clés pour intégrer l'API Bybit Trades dans vos projets. Les exemples fournis sont directement copiables et exécutables : lancez pip install requests et vous êtes prêt. Si vous envisagez d'analyser ces données avec de l'IA ou de construire des stratégies de trading automatisées, HolySheep AI offre une infrastructure performante avec une latence < 50ms et des coûts réduitgrâce au taux de change avantageux ($1 = ¥1). 👉 Inscrivez-vous sur HolySheep AI — crédits offerts N'hésitez pas à expérimenter avec différents symboles et paramètres. L'essentiel est de comprendre les principes de base que nous avons couverts : la structure des endpoints, la gestion des erreurs, et l'optimisation avec le cache. Bonne intégration !

Ressources connexes

Articles connexes