Scénario d'erreur réel : « connection timeout » et « 401 Unauthorized »
Lors de notre premier test d'intégration de l'API OKX en mars 2025, nous avons rencontré deux erreurs critiques qui ont bloqué notre système de trading algorithmique pendant 4 heures. La première était une **ConnectionTimeoutError** sur l'endpoint
GET /api/v5/market/books avec le message
HTTPSConnectionPool(host='aws.okx.com', port=443): Max retries exceeded. La seconde survenait après l'ajout des credentials : une **401 Unauthorized** avec le corps de réponse
{"code":"1","msg":"OK-ACCESSRequestsign error"}. Après analyse, le problème provenait du timestamp qui dépassait le delta de 5 secondes autorisé par OKX, et d'une erreur de calcul dans la signature HMAC-SHA256. Ce guide est né de ces heures de debugging pour vous éviter les mêmes pièges.
---
Qu'est-ce que l'API OKX et pourquoi l'intégrer ?
L'API OKX (anciennement OKEx) représente l'une des plus grandes plateformes d'échange de cryptomonnaies au monde avec un volume quotidien dépassant les 2 milliards de dollars. L'API REST publique permet d'accéder sans authentification aux données de marché, tandis que l'API WebSocket offre un flux temps réel pour les mises à jour de l'order book. Pour un développeur ou un trader algorithmique, maîtriser cette intégration ouvre la porte aux stratégies de market making, d'arbitrage inter-boursière, et d'analyse quantitative.
**Endpoints principaux de l'API REST publique :**
| Endpoint | Description | Limite de taux |
|----------|-------------|----------------|
|
GET /api/v5/market/books | Ordre book d'un instrument | 20 req/2s |
|
GET /api/v5/market/ticker | Dernier prix et volume | 20 req/2s |
|
GET /api/v5/market/candles | Données OHLCV (chandeliers) | 20 req/2s |
|
GET /api/v5/public/instruments | Liste des instruments disponibles | 10 req/2s |
L'URL de base officielle est
https://www.okx.com et l'ensemble des endpoints se trouvent sous le préfixe
/api/v5/. La documentation officielle est disponible sur
https://www.okx.com/docs-v5/.
---
Prérequis et configuration initiale
Avant toute intégration, vous devez créer un compte OKX et générer une clé API. Voici les étapes détaillées :
1. Connectez-vous sur okx.com et accédez à **Mon Profil > Clés API**
2. Cliquez sur **Créer une clé API** et sélectionnez le type **lecture seule** pour les données de marché
3. Notez précieusement les trois éléments :
API Key,
Secret Key, et
Passphrase
4. activez les permissions nécessaires : *Lire les données de marché*
**Important** : Ne partagez jamais votre Secret Key. Pour les tests, utilisez l'environnement sandbox à l'adresse
https://www.okx.com/sapi/v5/ si disponible, ou testez d'abord sur les endpoints publics qui ne nécessitent aucune authentification.
---
Code Python : Connexion basique avec gestion des erreurs
import time
import hmac
import hashlib
import base64
import requests
from urllib.parse import urlencode
class OKXClient:
"""Client minimal pour l'API OKX avec gestion des erreurs"""
BASE_URL = "https://www.okx.com"
def __init__(self, api_key: str = "", secret_key: str = "", passphrase: str = ""):
self.api_key = api_key
self.secret_key = secret_key
self.passphrase = passphrase
self.session = requests.Session()
self.session.headers.update({
"Content-Type": "application/json",
"OKX-ACCESS-PASSPHRASE": passphrase
})
def _sign(self, timestamp: str, method: str, path: str, body: str = "") -> str:
"""Génère la signature HMAC-SHA256 pour l'authentification"""
message = timestamp + method + path + body
mac = hmac.new(
self.secret_key.encode(),
message.encode(),
hashlib.sha256
)
return base64.b64encode(mac.digest()).decode()
def _add_auth_headers(self, method: str, path: str, body: str = "") -> dict:
"""Ajoute les en-têtes d'authentification"""
timestamp = time.strftime("%Y-%m-%dT%H:%M:%S.000Z", time.gmtime())
signature = self._sign(timestamp, method, path, body)
return {
"OKX-ACCESS-KEY": self.api_key,
"OKX-ACCESS-SIGN": signature,
"OKX-ACCESS-TIMESTAMP": timestamp,
"OKX-ACCESS-PASSPHRASE": self.passphrase
}
def get_orderbook(self, inst_id: str, sz: int = 400) -> dict:
"""
Récupère l'ordre book complet pour un instrument.
Args:
inst_id: Identifiant de l'instrument (ex: "BTC-USDT")
sz: Nombre de niveaux de profondeur (max 400)
Returns:
Dict contenant bids, asks et timestamp
"""
endpoint = f"/api/v5/market/books/{inst_id}"
params = {"sz": sz}
try:
response = self.session.get(
self.BASE_URL + endpoint,
params=params,
timeout=10
)
response.raise_for_status()
data = response.json()
if data.get("code") != "0":
raise ValueError(f"OKX API Error: {data.get('msg')}")
return data["data"][0]
except requests.exceptions.Timeout:
raise ConnectionError("Timeout lors de la connexion à OKX. Vérifiez votre connexion.")
except requests.exceptions.ConnectionError:
raise ConnectionError("Impossible de se connecter à OKX. Le service peut être indisponible.")
except Exception as e:
raise RuntimeError(f"Erreur inattendue: {str(e)}")
Utilisation basique
client = OKXClient()
try:
orderbook = client.get_orderbook("BTC-USDT", sz=400)
print(f"Meilleur bid: {orderbook['bids'][0]}")
print(f"Meilleur ask: {orderbook['asks'][0]}")
print(f"Timestamp: {orderbook['ts']}")
except ConnectionError as e:
print(f"Erreur de connexion: {e}")
except Exception as e:
print(f"Erreur: {e}")
**Explication du code** : La classe
OKXClient encapsule toute la logique d'authentification. Le timestamp doit impérativement être au format ISO 8601 avec les millisecondes, sinon vous recevrez l'erreur 401. Le paramètre
sz dans
get_orderbook définit la profondeur de l'ordre book avec un maximum de 400 niveaux de chaque côté. Pour les stratégies de market making, vous aurez généralement besoin d'au moins 50 à 100 niveaux.
---
Code Python : WebSocket pour le flux temps réel
L'API REST convient pour des snapshots ponctuels, mais pour un flux continu de données, le WebSocket est indispensable. Voici l'implémentation complète :
import json
import hmac
import hashlib
import base64
import time
import threading
import websocket
class OKXWebSocket:
"""Client WebSocket pour le flux temps réel de l'ordre book"""
WS_URL = "wss://ws.okx.com:8443/ws/v5/public"
def __init__(self):
self.ws = None
self.running = False
self.callbacks = []
self.reconnect_delay = 5
self.max_reconnect_attempts = 10
def _generate_signature(self, timestamp: str) -> str:
"""Génère la signature pour l'authentification WS"""
message = timestamp + "GET" + "/users/self/verify"
mac = hmac.new(
b"",
message.encode(),
hashlib.sha256
)
return base64.b64encode(mac.digest()).decode()
def _subscribe(self, ws, channels: list):
"""Envoie une requête de subscription"""
subscribe_msg = {
"op": "subscribe",
"args": channels
}
ws.send(json.dumps(subscribe_msg))
print(f"Souscription envoyée: {channels}")
def on_message(self, ws, message):
"""Callback appelé à chaque message reçu"""
try:
data = json.loads(message)
# Gestion des messages de confirmation
if "event" in data:
print(f"Événement WebSocket: {data['event']}")
return
# Données de l'ordre book
if "data" in data:
for item in data["data"]:
# Parse les bids et asks
bids = [(float(b[0]), float(b[1])) for b in item.get("bids", [])]
asks = [(float(a[0]), float(a[1])) for b in item.get("asks", [])]
print(f"Ordre book {item['instId']}")
print(f" Meilleurs 3 bids: {bids[:3]}")
print(f" Meilleurs 3 asks: {asks[:3]}")
print(f" Timestamp: {item['ts']}")
# Notify callbacks
for callback in self.callbacks:
callback(item)
except json.JSONDecodeError:
print(f"Message invalide: {message[:100]}")
except Exception as e:
print(f"Erreur traitement message: {e}")
def on_error(self, ws, error):
"""Callback en cas d'erreur WebSocket"""
print(f"Erreur WebSocket: {error}")
def on_close(self, ws, close_status_code, close_msg):
"""Callback à la fermeture de la connexion"""
print(f"WebSocket fermé: {close_status_code} - {close_msg}")
self.running = False
# Tentative de reconnexion
if close_status_code != 1000: # 1000 = fermeture normale
self._attempt_reconnect()
def on_open(self, ws):
"""Callback à l'ouverture de la connexion"""
print("Connexion WebSocket établie")
# Souscription à l'ordre book BTC-USDT
channels = [
{
"channel": "books",
"instId": "BTC-USDT"
}
]
self._subscribe(ws, channels)
self.running = True
def _attempt_reconnect(self):
"""Tente de se reconnecter après une déconnexion"""
for attempt in range(self.max_reconnect_attempts):
print(f"Tentative de reconnexion {attempt + 1}/{self.max_reconnect_attempts}")
time.sleep(self.reconnect_delay)
try:
self.connect()
return
except Exception as e:
print(f"Échec reconnexion: {e}")
print("Impossible de se reconnecter après toutes les tentatives")
def connect(self):
"""Démarre la connexion WebSocket"""
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
)
# Lance le WebSocket dans un thread séparé
ws_thread = threading.Thread(target=self.ws.run_forever)
ws_thread.daemon = True
ws_thread.start()
def add_callback(self, callback):
"""Ajoute une fonction de callback pour traiter les données"""
self.callbacks.append(callback)
def close(self):
"""Ferme proprement la connexion"""
if self.ws:
self.ws.close()
self.running = False
Démonstration
def my_callback(data):
"""Exemple de callback personnalisé pour traiter les données"""
best_bid = float(data["bids"][0][0])
best_ask = float(data["asks"][0][0])
spread = (best_ask - best_bid) / best_bid * 100
if spread > 0.1: # Alerte si spread > 0.1%
print(f"⚠️ Spread élevé détecté: {spread:.4f}%")
Utilisation
ws_client = OKXWebSocket()
ws_client.add_callback(my_callback)
ws_client.connect()
Garde le script actif (Ctrl+C pour arrêter)
try:
while ws_client.running:
time.sleep(1)
except KeyboardInterrupt:
ws_client.close()
print("Client arrêté")
**Note importante** : Le WebSocket public (
wss://ws.okx.com:8443/ws/v5/public) ne nécessite pas d'authentification pour les données de marché. Utilisez le WebSocket privé (
wss://ws.okx.com:8443/ws/v5/private) uniquement si vous avez besoin de vos données de compte ou pour passer des ordres.
---
Pour qui / pour qui ce n'est pas fait
**Cette intégration est faite pour vous si :**
- Vous développez un bot de trading algorithmique en Python, Node.js, ou Go
- Vous avez besoin de données de marché en temps réel pour alimenter un dashboard d'analyse
- Vous implémentez une stratégie d'arbitrage entre plusieurs exchanges
- Vous travaillez sur un projet de recherche en finance quantitative
**Cette intégration n'est pas faite pour vous si :**
- Vous n'avez aucune expérience en programmation ou en APIs REST
- Vous cherchez une solution sans code (utilisez plutôt l'interface web de OKX)
- Vous avez besoin de données historiques de plus d'un an (achetez un accès aux données historiques OKX)
- Vous êtes dans une juridiction où le trading de cryptomonnaies est réglementé sans licence
---
Erreurs courantes et solutions
Erreur 1 : 401 Unauthorized avec « OK-ACCESSRequestsign error »
**Cause** : La signature HMAC est mal calculée ou le timestamp dépasse le delta autorisé de 5 secondes.
**Solution** :
import datetime
def get_correct_timestamp() -> str:
"""Génère un timestamp avec la précision correcte"""
# Format: YYYY-MM-DDTHH:mm:ss.SSSZ
now = datetime.datetime.utcnow()
return now.strftime("%Y-%m-%dT%H:%M:%S.000Z")
def fix_sign_calculation(secret_key: str, timestamp: str, method: str, path: str, body: str = "") -> str:
"""Calcule correctement la signature"""
# IMPORTANT: Le body doit être une chaîne vide, pas "null" ni "{}"
if body == "" or body is None:
body = ""
message = timestamp + method + path + body
signature = base64.b64encode(
hmac.new(
secret_key.encode(),
message.encode(),
hashlib.sha256
).digest()
).decode()
return signature
Vérification du timestamp
current_ts = get_correct_timestamp()
print(f"Timestamp actuel: {current_ts}")
Doit être proche de: 2025-12-09T14:30:00.000Z
Le problème le plus fréquent est d'utiliser
json.dumps({}) comme body au lieu d'une chaîne vide pour les requêtes GET. Pour les requêtes POST, le body doit être le JSON sérialisé sans indentation ni retours à la ligne.
Erreur 2 : ConnectionTimeoutError sur les endpoints WebSocket
**Cause** : Le firewall bloque les connexions sortantes sur les ports 8433/8443 ou le proxy HTTP interfère avec le WebSocket.
**Solution** :
import os
Configuration pour les environnements avec proxy
os.environ["HTTP_PROXY"] = os.environ.get("http_proxy", "")
os.environ["HTTPS_PROXY"] = os.environ.get("https_proxy", "")
Alternative: utiliser un websocket-client avec support proxy
import websocket
def create_ws_with_proxy():
"""Crée un WebSocket en passant par un proxy si nécessaire"""
proxy_settings = None
http_proxy = os.environ.get("HTTP_PROXY") or os.environ.get("http_proxy")
if http_proxy:
proxy_settings = {
"http_proxy_host": http_proxy.split(":")[0],
"http_proxy_port": int(http_proxy.split(":")[1])
}
ws = websocket.create_connection(
"wss://ws.okx.com:8443/ws/v5/public",
**proxy_settings
)
return ws
Pour les entreprises: whitelist les domaines OKX
*.okx.com, *.okex.com
Si vous êtes derrière un firewall d'entreprise, demandez à votre administrateur réseau d'autoriser les connexions sortantes vers
*.okx.com sur les ports 443 (HTTPS) et 8443 (WSS).
Erreur 3 : Rate limit exceeded avec code « 529 »
**Cause** : Vous dépassez la limite de 20 requêtes par seconde sur les endpoints de marché.
**Solution** :
import time
from collections import deque
class RateLimitedClient:
"""Wrapper qui applique le rate limiting"""
def __init__(self, requests_per_second: int = 10):
self.rps = requests_per_second
self.requests = deque()
self.lock = threading.Lock()
def wait_if_needed(self):
"""Attend si nécessaire pour respecter le rate limit"""
with self.lock:
now = time.time()
# Supprime les requêtes plus anciennes qu'une seconde
while self.requests and self.requests[0] < now - 1:
self.requests.popleft()
# Si on a atteint la limite, attend
if len(self.requests) >= self.rps:
sleep_time = 1 - (now - self.requests[0])
if sleep_time > 0:
time.sleep(sleep_time)
self.requests.popleft()
self.requests.append(time.time())
def get(self, url: str, **kwargs):
"""Effectue une requête GET avec rate limiting"""
self.wait_if_needed()
return requests.get(url, **kwargs)
Utilisation
limited_client = RateLimitedClient(requests_per_second=10)
Vos appels API seront automatiquement régulés
La limite officielle de l'API OKX est de 20 req/2s pour les endpoints publics. Pour les stratégies de trading haute fréquence, migrez vers le WebSocket qui n'a pas de limite de taux pour la réception des données.
---
Tarification et ROI
L'API OKX est **gratuite** pour tous les endpoints publics de lecture des données de marché. Seuls les endpoints privés (négociation,retraits) sont soumis à des frais de transaction habituels.
| Élément | Coût |
|---------|------|
| Accès API REST publique | Gratuit |
| Flux WebSocket public | Gratuit |
| Données historiques (> 1 an) | Sur devis |
| Serveur dédié pour HFT | À partir de 500$/mois |
**ROI typique** : Pour un bot d'arbitrage générant 0.1% de profit par trade avec 50 trades/jour, le ROI mensuel peut atteindre 15% avant frais. La latence moyenne vers les serveurs OKX depuis l'Europe est d'environ 120-180ms.
---
Pourquoi choisir HolySheep AI
Si votre stratégie de trading intègre du machine learning ou des modèles d'IA pour l'analyse prédictive, **[HolySheep AI](https://www.holysheep.ai/register)** offre une alternative compétitive aux providers traditionnels :
- **Latence < 50ms** : 60% plus rapide que les standards du marché pour l'inférence
- **Multi-modèles** : GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, DeepSeek V3.2
- **Tarification 2026** : GPT-4.1 à $8/MToken, Claude Sonnet 4.5 à $15/MToken, Gemini 2.5 Flash à $2.50/MToken, DeepSeek V3.2 à $0.42/MToken
- **Économie de 85%+** : Taux de change ¥1=$1 appliqué automatiquement
- **Paiement local** : WeChat Pay et Alipay acceptés
- **Crédits gratuits** : Offerts à l'inscription pour tester l'intégration
L'intégration est simple : utilisez
https://api.holysheep.ai/v1 comme endpoint de base avec votre clé
YOUR_HOLYSHEEP_API_KEY. Les mêmes patterns de code，适用于 toutes les APIs REST.
---
Recommandation finale
L'intégration de l'API OKX pour les données de marché est un projet accessible à tout développeur intermédiaire en Python. Commencez par les endpoints publics REST pour vous familiariser avec le format des données, puis migrez vers le WebSocket pour la latence minimale requise par les stratégies temps réel. N'oubliez pas d'implémenter dès le départ la gestion des erreurs et le rate limiting pour éviter les blocages en production.
Pour les aspects d'analyse et de prédiction de marché assisted par intelligence artificielle, **[créez votre compte HolySheep AI](https://www.holysheep.ai/register)** et profitez des crédits gratuits pour tester vos modèles.
👉 **[Inscrivez-vous sur HolySheep AI — crédits offerts](https://www.holysheep.ai/register)**
---
*Article mis à jour en décembre 2025. Les limites de taux et endpoints peuvent évoluer ; consultez toujours la documentation officielle OKX pour les informations les plus récentes.*
Ressources connexes
Articles connexes