Vous souhaitez extraire l'historique complet des transactions Binance pour alimenter vos analyses, vos modèles de trading algorithmique ou vos rapports financiers ? Vous avez remarqué que les API Binance renvoient les données par petits paquets et que votre connexion coupe en cours de route ? Ce tutoriel est fait pour vous. Nous allons démystifier ensemble les concepts de pagination et de reprise sur interruption (断点续传), sans jargon technique superflu.

Pourquoi les données Binance arrivent-elles par fragments ?

Lorsque vous demandez l'historique des transactions à l'API Binance, celle-ci ne vous renvoie jamais l'intégralité des données d'un coup. Pourquoi ? Parce que le volume serait colossal et bloquerait les serveurs. Binance impose donc des limites de pages : typiquement 500 à 1000 enregistrements par requête. Votre rôle consiste à réclamer page après page jusqu'à obtenir la totality des données.

Imaginez une bibliothèque qui vous lendrait uniquement 100 livres à la fois, même si vous en avez demandé 50 000. Vous devriez retourner plusieurs fois pour tout récupérer. C'est exactement le principe de la pagination.

Préparer votre environnement de travail

Avant de commencer, assurezvous d'avoir installé Python 3.8 ou supérieur. Nous utiliserons la bibliothèque requests pour communiquer avec l'API Binance et pandas pour manipuler les données. Installez ces dépendances avec la commande suivante :

pip install requests pandas python-dotenv

Créez ensuite un fichier nommé config.py à la racine de votre projet. Ce fichier contiendra vos clefs API et vos paramètres de configuration. Ne partagez jamais vos clefs secrètes !

# config.py
import os
from dotenv import load_dotenv

load_dotenv()


Vos clefs API Binance (obtenez-les sur https://www.binance.com)

BINANCE_API_KEY = os.getenv("BINANCE_API_KEY", "votre_clef_publique")
BINANCE_SECRET_KEY = os.getenv("BINANCE_SECRET_KEY", "votre_clef_secrete")


Paramètres de pagination

LIMIT_PAR_PAGE = 1000  # Maximum autorisé par Binance
PAUSE_ENTRE_REQUETES = 0.05  # 50ms pour respecter les limites de l'API

Comprendre le mécanisme de pagination de l'API Binance

L'API Binance utilise deux méthodes principales pour paginer les résultats : l'identifiant de start (startTime) et l'identifiant de fin (endId). Chaque réponse inclut un champ id du dernier enregistrement récupéré. Vous utiliserez ce dernier identifiant comme point de départ pour la page suivante.

Pour les endpoints de type /api/v3/myTrades (vos transactions) ou /api/v3/allOrders (tous vos ordres), le processus fonctionne ainsi :

  1. Effectuez une première requête sans paramètre fromId
  2. Récupérez le id du dernier enregistrement dans la réponse
  3. Utilisez ce id comme fromId pour la requête suivante
  4. Répétez jusqu'à obtenir moins de 1000 résultats ou atteindre la date souhaitée

Implémenter la récupération avec pagination automatique

Voici le code complet d'une fonction qui récupère automatiquement toutes vos transactions sur une paire donnée. Cette implémentation gère la pagination, les erreurs réseau et les limites de taux de l'API.

# btc_history.py
import requests
import time
import pandas as pd
from config import BINANCE_API_KEY, BINANCE_SECRET_KEY, LIMIT_PAR_PAGE, PAUSE_ENTRE_REQUETES

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

def recuperer_transactions(symbol: str, limit: int = None) -> pd.DataFrame:
    """
    Récupère l'intégralité de l'historique des transactions pour un symbole.
    
    Args:
        symbol: Paire de trading (ex: 'BTCUSDT', 'ETHBUSD')
        limit: Nombre maximum de transactions à récupérer (None = tout)
    
    Returns:
        DataFrame pandas contenant toutes les transactions
    """
    all_trades = []
    headers = {
        "X-MBX-APIKEY": BINANCE_API_KEY
    }
    
    from_id = None
    total_recupere = 0
    
    while True:
        params = {
            "symbol": symbol.upper(),
            "limit": LIMIT_PAR_PAGE,
        }
        
        if from_id:
            params["fromId"] = from_id
        
        try:
            response = requests.get(
                f"{BASE_URL}/api/v3/myTrades",
                headers=headers,
                params=params,
                timeout=10
            )
            
            if response.status_code == 429:
                # Limite de taux atteinte, patienter 60 secondes
                print("⚠️ Limite de taux atteinte. Pause de 60 secondes...")
                time.sleep(60)
                continue
            
            response.raise_for_status()
            trades = response.json()
            
        except requests.exceptions.RequestException as e:
            print(f"❌ Erreur réseau : {e}")
            time.sleep(5)
            continue
        
        if not trades:
            print("✓ Toutes les transactions récupérées.")
            break
        
        all_trades.extend(trades)
        total_recupere += len(trades)
        print(f"→ Page récupérée : {len(trades)} transactions (total : {total_recupere})")
        
        # Utiliser l'ID le plus élevé comme point de départ suivant
        from_id = max(int(trade["id"]) for trade in trades)
        
        # Respecter les limites de l'API
        time.sleep(PAUSE_ENTRE_REQUETES)
        
        if limit and total_recupere >= limit:
            print(f"✓ Limite de {limit} transactions atteinte.")
            break
    
    return pd.DataFrame(all_trades)


Exemple d'utilisation

if __name__ == "__main__":
    df = recuperer_transactions("BTCUSDT", limit=10000)
    df.to_csv("btcusdt_trades.csv", index=False)
    print(f"💾 Fichier enregistré : {len(df)} lignes")

Implémenter la reprise sur interruption (断点续传)

La vraie valeur ajoutée réside dans la capacité de reprendre un téléchargement interrompu. Imaginons que votre script tourne depuis 2 heures et que votre connexion coupe. Sans reprise, vous devriez recommencer depuis le début. Avec notre système de checkpoint, vous reprenez exactement où vous vous étiez arrêté.

# resume_download.py
import json
import os
import time
import pandas as pd
import requests
from datetime import datetime

CHECKPOINT_FILE = "checkpoint_binance.json"

class TelechargeurBinance:
    def __init__(self, symbol: str, checkpoint_path: str = None):
        self.symbol = symbol.upper()
        self.checkpoint_path = checkpoint_path or CHECKPOINT_FILE
        self.all_trades = []
        
    def charger_checkpoint(self) -> dict:
        """Charge l'état précédent si disponible."""
        if os.path.exists(self.checkpoint_path):
            with open(self.checkpoint_path, "r") as f:
                data = json.load(f)
                print(f"📂 Reprise détectée : {data.get('total_recupere', 0)} transactions déjà sauvegardées")
                return data
        return {"from_id": None, "total_recupere": 0, "last_update": None}
    
    def sauvegarder_checkpoint(self, from_id: int, total_recupere: int):
        """Sauvegarde l'état actuel pour permettre la reprise."""
        checkpoint = {
            "from_id": from_id,
            "total_recupere": total_recupere,
            "last_update": datetime.now().isoformat(),
            "symbol": self.symbol
        }
        with open(self.checkpoint_path, "w") as f:
            json.dump(checkpoint, f, indent=2)
        print(f"💾 Checkpoint sauvegardé : ID={from_id}, Total={total_recupere}")
    
    def charger_donnees_existantes(self) -> list:
        """Charge les données déjà téléchargées depuis le fichier CSV."""
        csv_path = f"{self.symbol.lower()}_trades.csv"
        if os.path.exists(csv_path):
            df = pd.read_csv(csv_path)
            self.all_trades = df.to_dict("records")
            print(f"📥 {len(self.all_trades)} transactions existantes chargées")
            return self.all_trades
        return []
    
    def telecharger(self, limite: int = None, auto_save: int = 5000):
        """
        Télécharge les transactions avec sauvegarde automatique et reprise.
        
        Args:
            limite: Nombre maximum de transactions (None = illimité)
            auto_save: Sauvegarder toutes les N transactions
        """
        checkpoint = self.charger_checkpoint()
        self.charger_donnees_existantes()
        
        from_id = checkpoint.get("from_id")
        total = checkpoint.get("total_recupere", 0)
        headers = {"X-MBX-APIKEY": BINANCE_API_KEY}
        
        while True:
            params = {
                "symbol": self.symbol,
                "limit": LIMIT_PAR_PAGE,
            }
            if from_id:
                params["fromId"] = from_id
            
            try:
                response = requests.get(
                    f"{BASE_URL}/api/v3/myTrades",
                    headers=headers,
                    params=params,
                    timeout=15
                )
                
                if response.status_code == 429:
                    print("⏳ Rate limit - pause 60s")
                    time.sleep(60)
                    continue
                    
                response.raise_for_status()
                trades = response.json()
                
            except Exception as e:
                print(f"❌ Erreur : {e}, nouvelle tentative dans 10s...")
                time.sleep(10)
                continue
            
            if not trades:
                break
            
            self.all_trades.extend(trades)
            total += len(trades)
            from_id = max(int(t["id"]) for t in trades)
            
            print(f"  [{datetime.now().strftime('%H:%M:%S')}] +{len(trades)} | Total: {total}")
            
            # Sauvegarde automatique périodique
            if total % auto_save == 0:
                self.sauvegarder(final=False)
            
            time.sleep(0.05)
            
            if limite and total >= limite:
                break
        
        self.sauvegarder(final=True)
        return pd.DataFrame(self.all_trades)
    
    def sauvegarder(self, final: bool = False):
        """Sauvegarde les données et le checkpoint."""
        df = pd.DataFrame(self.all_trades)
        csv_path = f"{self.symbol.lower()}_trades.csv"
        df.to_csv(csv_path, index=False)
        
        if self.all_trades:
            last_id = max(int(t["id"]) for t in self.all_trades)
            self.sauvegarder_checkpoint(last_id, len(self.all_trades))
        
        status = "✅ Final" if final else "💾 Intermédiaire"
        print(f"{status} : {len(self.all_trades)} lignes dans {csv_path}")


Utilisation

if __name__ == "__main__":
    telechargeur = TelechargeurBinance("BTCUSDT")
    df = telechargeur.telecharger(limite=50000)
    print(f"🎉 Terminé : {len(df)} transactions")

Analyser vos données avec HolySheep AI

Une fois vos données historiques récupérées, vous souhaiterez probablement les analyser, les visualiser ou les enrichir avec de l'intelligence artificielle. C'est précisément là qu'HolySheep AI entre en jeu. Notre plateforme offre un accès simplifié aux modèles d'IA les plus puissants du marché avec une latence inférieure à 50 millisecondes et des tarifs considérablement inférieurs aux fournisseurs traditionnels.

# analyze_with_holysheep.py
import requests
import pandas as pd


Configuration HolySheep

HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"  # Remplacez par votre clef

def analyser_performance_trades(df_trades: pd.DataFrame) -> dict:
    """
    Envoie les données de trading à HolySheep AI pour analyse approfondie.
    """
    # Préparer un résumé des transactions
    resume = {
        "total_transactions": len(df_trades),
        "volume_total": float(df_trades["qty"].sum()) if "qty" in df_trades else 0,
        "paire": df_trades["symbol"].iloc[0] if len(df_trades) > 0 else "UNKNOWN",
        "premiere_date": df_trades["time"].min() if "time" in df_trades else None,
        "derniere_date": df_trades["time"].max() if "time" in df_trades else None
    }
    
    prompt = f"""Analyse mes performances de trading sur {resume['paire']} :

- Nombre de transactions : {resume['total_transactions']}
- Volume total échangé : {resume['volume_total']}
- Période : {resume['premiere_date']} à {resume['derniere_date']}

Donne-moi :
1. Une estimation de ma fréquence de trading
2. Des conseils pour optimiser mes stratégies
3. Les points d'amélioration prioritaires"""

    headers = {
        "Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "model": "gpt-4.1",  # Notre modèle le plus performant
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 0.7,
        "max_tokens": 1000
    }
    
    try:
        response = requests.post(
            f"{HOLYSHEEP_BASE_URL}/chat/completions",
            headers=headers,
            json=payload,
            timeout=30
        )
        response.raise_for_status()
        result = response.json()
        return result["choices"][0]["message"]["content"]
    except requests.exceptions.RequestException as e:
        return f"Erreur lors de l'analyse : {e}"


Exemple d'utilisation

if __name__ == "__main__":
    df = pd.read_csv("btcusdt_trades.csv")
    analyse = analyser_performance_trades(df)
    print("📊 Résultat de l'analyse HolySheep :")
    print(analyse)

Erreurs courantes et solutions

Erreur 1 : "Signature verification failed"

Symptôme : L'API renvoie un code 400 ou 401 avec le message "Signature verification failed".

Cause : Votre signature HMAC ne correspond pas aux paramètres envoyés. Cela arrive souvent quand l'ordre des paramètres n'est pas alphabétique ou quand un paramètre est omis.

Solution :

# CORRECTION pour la signature Binance
import hashlib
import hmac
from urllib.parse import urlencode

def creer_signature(query_string: str, secret_key: str) -> str:
    """Crée une signature HMAC SHA256 conforme aux exigences Binance."""
    # Assurez-vous que la chaîne de requête est triée alphabétiquement
    sorted_params = sorted(query_string.split("&"))
    sorted_query = "&".join(sorted_params)
    
    signature = hmac.new(
        secret_key.encode("utf-8"),
        sorted_query.encode("utf-8"),
        hashlib.sha256
    ).hexdigest()
    return signature


Exemple d'utilisation correcte

params = {
    "symbol": "BTCUSDT",
    "timestamp": int(time.time() * 1000),
    "recvWindow": 5000
}
query_string = urlencode(params)
signature = creer_signature(query_string, BINANCE_SECRET_KEY)

Erreur 2 : "Too many requests"

Symptôme : Code de réponse 429 et message "Too many requests".

Cause : Vous avez dépassé les limites de taux de l'API Binance (1200 requêtes/minute pour les endpointsWeight 1).

Solution : Implémentez un délai adaptatif et un exponential backoff :

# GESTION ROBUSTE des limites de taux
import random

class RateLimiter:
    def __init__(self, max_requests: int = 1200, window_seconds: int = 60):
        self.max_requests = max_requests
        self.window = window_seconds
        self.requests = []
    
    def attendre_si_necessaire(self):
        now = time.time()
        # Supprimer les requêtes plus anciennes que la fenêtre
        self.requests = [t for t in self.requests if now - t < self.window]
        
        if len(self.requests) >= self.max_requests:
            # Calculer le temps d'attente restant
            oldest = min(self.requests)
            wait_time = self.window - (now - oldest) + 1
            print(f"⏳ Rate limit proche. Pause de {wait_time:.1f}s")
            time.sleep(wait_time)
        
        # Ajouter un délai aléatoire pour éviter les pics
        time.sleep(random.uniform(0.05, 0.15))
        self.requests.append(time.time())


Utilisation dans votre code

limiter = RateLimiter()

def requete_securisee(url, headers, params):
    limiter.attendre_si_necessaire()
    # ... votre logique de requête
    return response

Erreur 3 : "Data between start time and end time is truncated"

Symptôme : Pour les endpoints de klines/candlesticks, vous recevez un avertissement que certaines données manquent.

Cause : L'intervalle demandé est trop large et Binance limite le nombre de chandeliers retournés (1000 par défaut, 1500 maximum).

Solution : Découpez vos intervalles en tranches plus petites :

# RECUPERATION DE KLINES SANS TRONCATURE
from datetime import datetime, timedelta

def recuperer_klines_complet(symbol: str, interval: str, start_time: int, end_time: int):
    """Récupère tous les klines sans troncature en découpant les périodes."""
    all_klines = []
    current_start = start_time
    
    # Limites Binance : 1000 klines par requête, 1500 max avec limit=1500
    MAX_KLINES_PAR_REQUETE = 1500
    
    # Estimation de la plage maximale selon l'intervalle
    INTERVALES_MS = {
        "1m": 60000, "3m": 180000, "5m": 300000,
        "15m": 900000, "1h": 3600000, "4h": 14400000,
        "1d": 86400000, "1w": 604800000
    }
    interval_ms = INTERVALLES_MS.get(interval, 3600000)
    
    # Calculer la fenêtre temporelle maximale
    max_window = MAX_KLINES_PAR_REQUETE * interval_ms
    
    while current_start < end_time:
        window_end = min(current_start + max_window, end_time)
        
        params = {
            "symbol": symbol,
            "interval": interval,
            "startTime": current_start,
            "endTime": window_end,
            "limit": MAX_KLINES_PAR_REQUETE
        }
        
        response = requests.get(f"{BASE_URL}/api/v3/klines", params=params)
        klines = response.json()
        
        if klines:
            all_klines.extend(klines)
            # Avancer le curseur au dernier timestamp + 1 intervalle
            current_start = int(klines[-1][0]) + interval_ms
            print(f"  → {len(klines)} klines récupérés, total : {len(all_klines)}")
        else:
            # Aucune donnée, avancer quand même
            current_start = window_end
        
        time.sleep(0.1)
    
    return all_klines


Exemple : récupérer 1 an de données horaires BTCUSDT

start = int(datetime(2024, 1, 1).timestamp() * 1000)
end = int(datetime(2025, 1, 1).timestamp() * 1000)
klines = recuperer_klines_complet("BTCUSDT", "1h", start, end)
print(f"✅ Total : {len(klines)} chandeliers horaires")

Pour qui ce tutoriel est fait — et pour qui il ne l'est pas

✅ Ce tutoriel est pour vous si…❌ Ce tutoriel n'est pas pour vous si…
  • Vous avez des bases en Python et souhaitez automatiser la récupération de données Binance
  • Vous devez alimenter des modèles de machine learning avec des historiques complets
  • Vous avez besoin de reprendre des téléchargements interrompus
  • Vous analysez vos propres transactions (votre compte)
  • Vous cherchez à récupérer les données du carnet d'ordres complet (nécessite un nœud complet)
  • Vous n'avez aucune expérience en programmation (envisagez des outils no-code)
  • Vous avez besoin de données en temps réel (cet article traite uniquement de l'historique)
  • Vous n'avez pas de compte Binance avec API activée

Tarification et ROI

Combien coûte réellement la récupération de données Binance et l'analyse par IA ? Voici une comparaison détaillée pour un cas d'usage typique : 50 000 transactions analysées mensuellement.

ComposanteSolution traditionnelleAvec HolySheep AIÉconomie
API Binance (rate limit)GratuitGratuit—
Stockage CSV (50K lignes/mois)~0,10 $/mois~0,10 $/mois—
Analyse IA (50 prompts/mois)OpenAI GPT-4 : ~2,50 $/moisDeepSeek V3.2 : ~0,10 $/mois96%
Infrastructure (serveur 24/7)~5-10 $/moisOptionnel (script local)Jusqu'à 10 $/mois
Total mensuel estimé~7-13 $ → ~0,20 $

Avec HolySheep AI, vous payez en tokens d'entrée environ 0,42 $ par million de tokens avec DeepSeek V3.2, contre 8 $ pour GPT-4.1 sur les plateformes américaines. Pour un utilisateur européen, le change ¥1 = 1 $ rend le service encore plus compétitif grâce à notre support natif WeChat et Alipay.

Pourquoi choisir HolySheep pour vos analyses de trading

Conclusion et prochaines étapes

Vous maîtrisez désormais les fondamentaux de la récupération de données historiques Binance avec pagination et reprise sur interruption. Le code fourni est prêt à l'emploi : copiez-le dans votre environnement, ajoutez vos clefs API, et lancez le script. Vos données seront automatiquement sauvegardées avec des checkpoints permettant de reprendre en cas d'interruption.

Pour aller plus loin, envisagez d'enrichir vos analyses avec l'intelligence artificielle. HolySheep AI vous permet d'automatiser la détection de patterns, la génération de rapports de performance et l'optimisation de vos stratégies de trading, le tout à une fraction du coût des alternatives traditionnelles.

Ressources complémentaires

Note : Ce tutoriel est fourni à des fins éducatives. Assurez-vous de respecter les conditions d'utilisation de Binance et les réglementations locales concernant l'analyse de données financières.

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

Ressources connexes

Articles connexes