Vous souhaitez accéder aux données de marché des cryptomonnaies en temps réel ? Dans ce tutoriel, je vais vous montrer pas à pas comment configurer une connexion entre l'API Tardis pour les données historiques et les WebSockets OKX pour les flux de données en direct. Aucun prérequis technique n'est nécessaire : nous partons de zéro.

Préambule : Comprendre les APIs de données crypto

Avant de coder, laissez-moi vous expliquer ce que nous allons faire concrètement. Une API (Interface de Programmation d'Applications) est simplement un moyen pour deux programmes informatiques de communiquer. Dans notre cas :

Imaginez que Tardis est votre bibliothèque de films passés et que OKX WebSocket est votre flux TV en direct. Vous avez besoin des deux pour avoir une vue complète du marché.

Configuration initiale de l'environnement

Avant de commencer, installez Python si ce n'est pas déjà fait. Ensuite, nous allons installer les bibliothèques nécessaires :

# Installation des bibliothèques requise
pip install requests websocket-client aiohttp pandas


Vérification de l'installation

python -c "import requests; print('Requests installé:', requests.__version__)"

[Screenshot 1 : Terminal montrant les packages installés avec succès - cherchez une ligne verte avec "Successfully installed"]

Connexion à l'API Tardis

L'API Tardis permet d'accéder à des données de marché cryptocurrency structurées. Vous aurez besoin de :

import requests
import pandas as pd
from datetime import datetime, timedelta


Configuration de l'API Tardis

TARDIS_API_KEY = "votre_cle_api_tardis"
BASE_URL_TARDIS = "https://api.tardis.dev/v1"

def recuperer_donnees_historiques(symbol, jours=7):
    """
    Récupère les données OHLCV historiques depuis Tardis
    symbol : paire de trading (ex: 'binance:btcusdt')
    jours : nombre de jours de données à récupérer
    """
    date_fin = datetime.utcnow()
    date_debut = date_fin - timedelta(days=jours)
    
    url = f"{BASE_URL_TARDIS}/historical/candles"
    params = {
        'exchange': 'binance',
        'symbol': symbol.upper().replace('/', ''),
        'from': int(date_debut.timestamp()),
        'to': int(date_fin.timestamp()),
        'interval': '1m'  # Chandeliers de 1 minute
    }
    
    headers = {
        'Authorization': f'Bearer {TARDIS_API_KEY}'
    }
    
    try:
        response = requests.get(url, params=params, headers=headers)
        response.raise_for_status()
        
        donnees = response.json()
        df = pd.DataFrame(donnees)
        
        print(f"✓ {len(df)} chandeliers récupérés pour {symbol}")
        return df
        
    except requests.exceptions.RequestException as e:
        print(f"✗ Erreur de connexion : {e}")
        return None


Exemple d'utilisation

donnees = recuperer_donnees_historiques("BTC/USDT", jours=1)

Ce code récupère les données de chandeliers (OHLCV : Open, High, Low, Close, Volume) pour analyser l'historique des prix. La fonction retourne un DataFrame pandas que vous pouvez ensuite analyser ou sauvegarder.

Connexion WebSocket OKX en temps réel

Maintenant, la partie passionnante : recevoir les prix en direct ! Les WebSockets sont différents des requêtes HTTP classiques. Avec HTTP, vous demandez des données et le serveur répond. Avec WebSocket, le serveur vous envoie les données automatiquement quand elles arrivent.

import json
import websocket
import threading
import time

class ClientWebSocketOKX:
    """Client WebSocket pour les données de marché OKX en temps réel"""
    
    def __init__(self, symbol="BTC-USDT"):
        self.symbol = symbol
        self.ws = None
        self.connexion_activee = False
        self.donnees_temps_reel = []
        
        # URL du WebSocket OKX (API publique - pas de clé requise)
        self.ws_url = "wss://ws.okx.com:8443/ws/v5/public"
    
    def on_message(self, ws, message):
        """Callback déclenché à chaque nouveau message"""
        try:
            data = json.loads(message)
            
            # Filtrer uniquement les messages de ticker
            if 'data' in data:
                for ticker in data['data']:
                    prix_achat = float(ticker['bidPx'])      # Meilleure offre d'achat
                    prix_vente = float(ticker['askPx'])      # Meilleure offre de vente
                    volume_24h = float(ticker['vol24h'])     # Volume sur 24h
                    timestamp = ticker['ts']                 # Horodatage
                    
                    print(f"🟢 {self.symbol} | Achat: ${prix_achat:,.2f} | "
                          f"Vente: ${prix_vente:,.2f} | Vol: {volume_24h:,.0f}")
                    
                    # Stocker pour analyse ultérieure
                    self.donnees_temps_reel.append({
                        'timestamp': timestamp,
                        'bid': prix_achat,
                        'ask': prix_vente,
                        'volume': volume_24h
                    })
                    
        except (json.JSONDecodeError, KeyError, TypeError) as e:
            print(f"⚠ Erreur parsing message : {e}")
    
    def on_error(self, ws, error):
        """Gestion des erreurs de connexion"""
        print(f"❌ Erreur WebSocket : {error}")
    
    def on_close(self, ws, close_status_code, close_msg):
        """Callback quand la connexion se ferme"""
        print(f"🔴 Connexion fermée ({close_status_code}): {close_msg}")
        self.connexion_activee = False
    
    def on_open(self, ws):
        """Callback à l'ouverture de la connexion - s'abonne au ticker"""
        print("🟢 Connexion établie ! Abonnement en cours...")
        
        # Format de subscription pour OKX
        subscribe_msg = {
            "op": "subscribe",
            "args": [{
                "channel": "tickers",
                "instId": self.symbol
            }]
        }
        
        ws.send(json.dumps(subscribe_msg))
        print(f"✓ Abonné au ticker {self.symbol}")
    
    def demarrer(self):
        """Démarre la connexion WebSocket dans un thread séparé"""
        websocket.enableTrace(False)  # Mettre True pour debug
        
        self.ws = websocket.WebSocketApp(
            self.ws_url,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close,
            on_open=self.on_open
        )
        
        # Lancer dans un thread pour ne pas bloquer
        self.thread = threading.Thread(target=self.ws.run_forever)
        self.thread.daemon = True
        self.thread.start()
        
        self.connexion_activee = True
        print(f"🚀 Thread WebSocket démarré pour {self.symbol}")
    
    def arreter(self):
        """Ferme proprement la connexion"""
        if self.ws:
            self.ws.close()
            print("⏹ Connexion WebSocket fermée")


--- Utilisation pratique ---

if __name__ == "__main__":
    client = ClientWebSocketOKX(symbol="BTC-USDT")
    
    print("=" * 60)
    print("RÉCEPTION EN TEMPS RÉEL - Appuyez sur Ctrl+C pour arrêter")
    print("=" * 60)
    
    client.demarrer()
    
    try:
        # Garder le programme actif pendant 60 secondes
        time.sleep(60)
    except KeyboardInterrupt:
        print("\n⏹ Arrêt demandé par l'utilisateur")
    finally:
        client.arreter()
        
        # Afficher un résumé des données collectées
        if client.donnees_temps_reel:
            print(f"\n📊 {len(client.donnees_temps_reel)} mises à jour collectées")

[Screenshot 2 : Console affichant les flux de prix en temps réel - vous verrez les prix BTC-USDT s'actualiser toutes les secondes]

Fusionner les deux flux : Historique + Temps Réel

Voici maintenant le code complet qui combine les données historiques de Tardis avec le flux temps réel d'OKX pour une analyse complète du marché :

import requests
import websocket
import json
import threading
import time
import pandas as pd
from datetime import datetime


============================================


CONFIGURATION - REMPLACEZ CES VALEURS


============================================

TARDIS_API_KEY = "votre_cle_tardis"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"  # Pour analyses avancées IA


============================================


CLASSE TARDIS - DONNÉES HISTORIQUES


============================================

class CollecteurTardis:
    """Collecte les données OHLCV historiques depuis Tardis"""
    
    BASE_URL = "https://api.tardis.dev/v1"
    
    def __init__(self, api_key):
        self.api_key = api_key
    
    def obtenir_candles(self, exchange, symbol, debut, fin, intervalle='1m'):
        """Récupère les chandeliers historiques"""
        url = f"{self.BASE_URL}/historical/candles"
        
        params = {
            'exchange': exchange,
            'symbol': symbol,
            'from': int(debut.timestamp()),
            'to': int(fin.timestamp()),
            'interval': intervalle
        }
        
        headers = {'Authorization': f'Bearer {self.api_key}'}
        
        response = requests.get(url, params=params, headers=headers)
        response.raise_for_status()
        
        return response.json()


============================================


CLASSE OKX - FLUX TEMPS RÉEL


============================================

class FluxOKX:
    """Flux WebSocket pour données en temps réel depuis OKX"""
    
    WS_URL = "wss://ws.okx.com:8443/ws/v5/public"
    
    def __init__(self):
        self.donnees = []
        self.ws = None
        self.actif = False
    
    def traitement_message(self, ws, message):
        """Traite chaque message reçu"""
        data = json.loads(message)
        
        if 'data' in data:
            for item in data['data']:
                enregistrement = {
                    'timestamp': pd.to_datetime(item['ts'], unit='ms'),
                    'dernier_prix': float(item['last']),
                    'prix_achat': float(item['bidPx']),
                    'prix_vente': float(item['askPx']),
                    'spread': float(item['askPx']) - float(item['bidPx']),
                    'volume': float(item['vol24h'])
                }
                
                self.donnees.append(enregistrement)
                print(f"💹 {enregistrement['timestamp']} | Prix: ${enregistrement['dernier_prix']:,.2f}")
    
    def abonnement(self, ws, symbol):
        """S'abonne au flux du symbole"""
        msg = {
            "op": "subscribe",
            "args": [{"channel": "tickers", "instId": symbol}]
        }
        ws.send(json.dumps(msg))
    
    def demarrer(self, symbol="BTC-USDT"):
        """Démarre le flux en arrière-plan"""
        self.ws = websocket.WebSocketApp(
            self.WS_URL,
            on_message=self.traitement_message
        )
        
        def a louverture(ws):
            self.abonnement(ws, symbol)
        
        self.ws.on_open = a_louverture
        self.actif = True
        
        thread = threading.Thread(target=self.ws.run_forever)
        thread.daemon = True
        thread.start()
    
    def arreter(self):
        """Arrête le flux"""
        if self.ws:
            self.ws.close()
        self.actif = False


============================================


FONCTION PRINCIPALE


============================================

def analyser_marche():
    """Analyse complète avec historique + temps réel"""
    
    print("=" * 70)
    print("   ANALYSE DE MARCHÉ CRYPTO - TARDIS + OKX TEMPS RÉEL")
    print("=" * 70)
    
    # Étape 1 : Charger l'historique depuis Tardis
    print("\n📚 Étape 1 : Chargement des données historiques...")
    collecteur = CollecteurTardis(TARDIS_API_KEY)
    
    try:
        depuis = datetime.utcnow() - timedelta(days=2)
        jusqua = datetime.utcnow()
        
        historique = collecteur.obtenir_candles(
            exchange='binance',
            symbol='BTCUSDT',
            debut=depuis,
            fin=jusqua,
            intervalle='5m'
        )
        
        df_historique = pd.DataFrame(historique)
        print(f"✓ {len(df_historique)} chandeliers chargés")
        
        # Calculer des indicateurs simples
        df_historique['moyenne_mobile'] = df_historique['close'].rolling(20).mean()
        prix_actuel = df_historique['close'].iloc[-1]
        moyenne = df_historique['moyenne_mobile'].iloc[-1]
        
        print(f"📊 Prix actuel (historique) : ${prix_actuel:,.2f}")
        print(f"📊 Moyenne mobile 20 périodes : ${moyenne:,.2f}")
        
    except Exception as e:
        print(f"⚠ Impossible de charger l'historique : {e}")
        print("   Continuation avec le flux temps réel uniquement...")
    
    # Étape 2 : Démarrer le flux temps réel OKX
    print("\n🚀 Étape 2 : Démarrage du flux temps réel OKX...")
    flux = FluxOKX()
    flux.demarrer(symbol="BTC-USDT")
    
    print("⏳ Réception des données en temps réel pendant 30 secondes...")
    time.sleep(30)
    
    # Étape 3 : Analyse des données temps réel
    print("\n📈 Étape 3 : Analyse des données temps réel...")
    if flux.donnees:
        df_temps_reel = pd.DataFrame(flux.donnees)
        
        prix_moyen = df_temps_reel['dernier_prix'].mean()
        prix_max = df_temps_reel['dernier_prix'].max()
        prix_min = df_temps_reel['dernier_prix'].min()
        spread_moyen = df_temps_reel['spread'].mean()
        
        print(f"✓ Nombre de mises à jour : {len(df_temps_reel)}")
        print(f"📊 Prix moyen : ${prix_moyen:,.2f}")
        print(f"📊 Prix le plus haut : ${prix_max:,.2f}")
        print(f"📊 Prix le plus bas : ${prix_min:,.2f}")
        print(f"📊 Spread moyen achat/vente : ${spread_moyen:.4f}")
    
    # Étape 4 : Optionnel - Analyse IA avec HolySheep
    print("\n🤖 Étape 4 : Analyse IA optionnelle avec HolySheep AI...")
    print("   HolySheep offre <50ms de latence et des tarifs 85%+ inférieurs")
    print("   à OpenAI. Profitez de crédits gratuits pour tester :")
    print("   👉 https://www.holysheep.ai/register")
    
    flux.arreter()
    print("\n✅ Analyse terminée avec succès !")


Exécuter si appelé directement

if __name__ == "__main__":
    analyser_marche()

[Screenshot 3 : Résultat de l'exécution montrant les chandeliers historiques et les prix temps réel] [Screenshot 4 : Graphique simple des données collected]

Calcul du spread et de la liquidité

Une métrique cruciale en trading est le spread (différence entre prix d'achat et de vente). Plus le spread est faible, plus le marché est liquide. Voici comment le calculer et l'analyser :

import numpy as np

def analyser_liquidite(donnees_df):
    """
    Analyse la liquidité du marché à partir des données collectées
    """
    if 'spread' not in donnees_df.columns:
        print("⚠ La colonne 'spread' est requise dans les données")
        return None
    
    statistiques = {
        'spread_moyen': donnees_df['spread'].mean(),
        'spread_median': donnees_df['spread'].median(),
        'spread_min': donnees_df['spread'].min(),
        'spread_max': donnees_df['spread'].max(),
        'ecart_type_spread': donnees_df['spread'].std(),
        'volatilite_prix': donnees_df['dernier_prix'].std()
    }
    
    print("\n" + "=" * 50)
    print("   ANALYSE DE LIQUIDITÉ")
    print("=" * 50)
    print(f"Spread moyen     : ${statistiques['spread_moyen']:.6f}")
    print(f"Spread médian    : ${statistiques['spread_median']:.6f}")
    print(f"Spread minimum   : ${statistiques['spread_min']:.6f}")
    print(f"Spread maximum   : ${statistiques['spread_max']:.6f}")
    print(f"Écart-type spread: ${statistiques['ecart_type_spread']:.6f}")
    print(f"Volatilité prix  : ${statistiques['volatilite_prix']:.2f}")
    
    # Évaluation de la liquidité
    spread_pourcentage = (statistiques['spread_moyen'] / donnees_df['dernier_prix'].mean()) * 100
    
    if spread_pourcentage < 0.01:
        liquidite = "EXCELLENTE 🟢"
    elif spread_pourcentage < 0.05:
        liquidite = "BONNE 🟡"
    elif spread_pourcentage < 0.1:
        liquidite = "MOYENNE 🟠"
    else:
        liquidite = "FAIBLE 🔴"
    
    print(f"\n📊 Liquidité globale : {liquidite}")
    print(f"   (Spread en % du prix : {spread_pourcentage:.4f}%)")
    
    return statistiques


Utilisation avec nos données collectées

if 'df_temps_reel' in dir():
    stats = analyser_liquidite(df_temps_reel)

Erreurs courantes et solutions

Erreur 1 : "403 Forbidden" ou "Unauthorized" - Clé API invalide

Symptôme : Vous recevez une erreur HTTP 401 ou 403 quand vous faites des requêtes à Tardis.

# ❌ MAUVAIS - Clé mal formatée
headers = {
    'Authorization': 'Bearer ma_cle_api'  # Espace manquant après Bearer
}


✅ CORRECT - Format standard

headers = {
    'Authorization': f'Bearer {TARDIS_API_KEY.strip()}'
}


✅ CORRECT - Vérification de la clé

if not TARDIS_API_KEY or len(TARDIS_API_KEY) < 20:
    raise ValueError("⚠ Clé API Tardis invalide ou manquante")
print("✓ Clé API validée")

Solutions :

Erreur 2 : WebSocket OKX se déconnecte après quelques secondes

Symptôme : La connexion WebSocket fonctionne au début mais se coupe après 10-30 secondes.

# ❌ PROBLÈME - Pas de heartbeat, le serveur coupe la connexion
ws = websocket.WebSocketApp(url)


✅ SOLUTION - Implémenter un ping automatique pour maintenir la connexion

import time

class FluxOKXRobuste:
    def __init__(self):
        self.ws = None
        self.derniere_activite = time.time()
        self.delai_inactivite = 30  # Secondes avant reconnexion
    
    def demarrer(self, symbol):
        self.ws = websocket.WebSocketApp(
            "wss://ws.okx.com:8443/ws/v5/public",
            on_message=self.on_message,
            on_ping=self.on_ping  # Gère les ping automatiquement
        )
        thread = threading.Thread(target=self._boucle_surveillance)
        thread.start()
    
    def _boucle_surveillance(self):
        """Surveille l'activité et reconnecte si nécessaire"""
        while True:
            if time.time() - self.derniere_activite > self.delai_inactivite:
                print("🔄 Reconnexion due à l'inactivité...")
                self.reconnecter()
            time.sleep(5)
    
    def on_message(self, ws, message):
        self.derniere_activite = time.time()  # Reset du timer
        # ... traitement du message
    
    def reconnecter(self):
        """Reconnexion propre"""
        if self.ws:
            self.ws.close()
        time.sleep(1)
        self.demarrer(self.symbol)

Solutions :

Erreur 3 : "JSONDecodeError: Expecting value" - Données vides

Symptôme : L'API retourne une réponse vide ou le parsing JSON échoue.

# ❌ PROBLÈMATIQUE - Pas de validation
response = requests.get(url)
data = response.json()  # Échoue si réponse vide


✅ ROBUSTE - Validation complète de la réponse

def requete_securisee(url, headers, params):
    """Requête HTTP avec gestion complète des erreurs"""
    
    try:
        response = requests.get(url, headers=headers, params=params, timeout=30)
        response.raise_for_status()
        
        # Vérifier que le corps n'est pas vide
        if not response.text:
            raise ValueError("Réponse API vide")
        
        # Vérifier que c'est du JSON valide
        try:
            data = response.json()
        except json.JSONDecodeError as e:
            raise ValueError(f"Réponse non-JSON : {e}")
        
        # Vérifier que les données ne sont pas None
        if data is None:
            raise ValueError("Données JSON null")
        
        # Cas spécial : Tardis peut retourner une liste vide
        if isinstance(data, list) and len(data) == 0:
            print("⚠ Avertissement : Aucune donnée pour cette période")
            return []
        
        return data
        
    except requests.exceptions.Timeout:
        print("⏱ Timeout - Le serveur met trop de temps à répondre")
        return None
    except requests.exceptions.ConnectionError:
        print("🌐 Erreur de connexion - Vérifiez votre internet")
        return None

Solutions :

Erreur 4 : Limite de taux (Rate Limit) dépassée

Symptôme : Erreur 429 ou message "Too many requests".

import time
from functools import wraps

def limiteur_taux(appels_par_seconde=10):
    """Décorateur pour limiter le taux d'appels API"""
    min_intervalle = 1.0 / appels_par_seconde
    
    def decorateur(fonction):
        dernier_appel = [0.0]
        
        @wraps(fonction)
        def wrapper(*args, **kwargs):
            maintenant = time.time()
            delai = min_intervalle - (maintenant - dernier_appel[0])
            
            if delai > 0:
                time.sleep(delai)
            
            resultat = fonction(*args, **kwargs)
            dernier_appel[0] = time.time()
            return resultat
        
        return wrapper
    return decorateur


Utilisation

@limiteur_taux(appels_par_seconde=5)  # Maximum 5 requêtes/seconde
def requete_tardis(url, params):
    response = requests.get(url, params=params)
    return response.json()


Pour les WebSockets, implémentez un buffer

class BufferDonnees:
    """Accumule les données pour traitement par lots"""
    
    def __init__(self, taille_max=100, intervalle=5):
        self.buffer = []
        self.taille_max = taille_max
        self.intervalle = intervalle
        self.dernier_traitement = time.time()
    
    def ajouter(self, donnee):
        self.buffer.append(donnee)
        
        # Traiter si le buffer est plein OU si le délai est écoulé
        if (len(self.buffer) >= self.taille_max or 
            time.time() - self.derniere_traitement >= self.intervalle):
            return self.traiter()
        return None
    
    def traiter(self):
        if not self.buffer:
            return None
        
        donnees = self.buffer.copy()
        self.buffer.clear()
        self.derniere_traitement = time.time()
        
        print(f"📦 Traitement de {len(donnees)} éléments")
        return donnees

Architecture recommandée pour la production

Pour un usage en production avec des données fiables, je recommande cette architecture simplifiée :

# Structure de projet recommandée
projet_crypto/
├── config.py              # Configuration centralisée
├── historique/
│   └── tardis_client.py   # Client pour données historiques
├── temps_reel/
│   └── okx_websocket.py   # Client WebSocket OKX
├── stockage/
│   └── database.py        # Sauvegarde des données
├── analyse/
│   └── indicators.py      # Calcul des indicateurs
└── main.py                # Point d'entrée


Exemple de config.py

CONFIG = {
    'tardis': {
        'api_key': 'votre_cle_tardis',
        'base_url': 'https://api.tardis.dev/v1',
        'timeout': 30
    },
    'okx': {
        'ws_url': 'wss://ws.okx.com:8443/ws/v5/public',
        'symbols': ['BTC-USDT', 'ETH-USDT', 'SOL-USDT'],
        'reconnect_delay': 5
    },
    'stockage': {
        'type': 'sqlite',  # Ou 'postgresql' pour la production
        'chemin': './data/crypto.db'
    }
}

Récapitulatif des performances

SourceLatenceType de donnéesCoût approximatif
Tardis API100-300msHistorique OHLCVSelon abonnement
OKX WebSocket<50msTemps réel tickerGratuit (tier public)
HolySheep AI<50msAnalyse IA85%+ moins cher que GPT-4

Aller plus loin avec HolySheep AI

Pour analyser automatiquement les données collectées, vous pouvez utiliser l'API HolySheep AI qui offre des performances exceptionnelles pour le traitement de données financières. HolySheep propose :

# Exemple d'analyse IA avec HolySheep
import requests

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

def analyser_avec_ia(donnees_marche, modele="deepseek-v3.2"):
    """
    Envoie les données de marché à HolySheep pour analyse IA
    """
    prompt = f"""
    Analyse les données de marché suivantes et donne une recommandation :
    
    {donnees_marche}
    
    Format de réponse souhaité :
    - Tendance : Haussière / Baissière / Neutre
    - Niveau de confiance : Élevé / Moyen / Faible
    - Recommandation : Acheter / Vendre / Attendre
    """
    
    response = requests.post(
        f"{BASE_URL_HOLYSHEEP}/chat/completions",
        headers={
            'Authorization': f'Bearer {HOLYSHEEP_API_KEY}',
            'Content-Type': 'application/json'
        },
        json={
            'model': modele,
            'messages': [{'role': 'user', 'content': prompt}]
        }
    )
    
    return response.json()


Utilisation après collecte des données

if donnees_temps_reel:
    analyse = analyser_avec_ia(df_temps_reel.describe())
    print("🤖 Analyse IA :", analyse)

Conclusion

Vous savez maintenant comment configurer une connexion complète entre l'API Tardis pour les données historiques et les WebSockets OKX pour les flux en temps réel. Les points clés à retenir :

Pour approfondir vos compétences en trading algorithmique et analyse de données crypto, je vous recommande de expérimenter avec différents symboles, intervalles de temps, et d'ajouter vos propres indicateurs techniques.

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

Ressources connexes

Articles connexes