Introduction : Pourquoi Automatiser le Market Making sur OKX ?

En tant qu'ingénieur quantitatif ayant développé des systèmes de trading algorithmique depuis 2019, j'ai testé une multitude de configurations pour le market making cryptomonnaie. L'équation est simple : sans couverture automatisée, la volatilité dévore vos marges en quelques heures. J'ai migré mon infrastructure vers HolySheep AI pour la couche intelligence artificielle — sentiment analysis, optimisation des spreads — et voici mon playbook complet.

Ce guide suppose que vous maîtrisez les concepts de base du trading algorithmique et que vous avez déjà un compte OKX avec les autorisations API appropriées. Si ce n'est pas le cas, commencez par créer un compte OKX et générez vos clés API avec les permissions "Trade" et "Read".

Architecture du Système de Market Making

Un système de market making robuste se compose de trois couches distinctes :

Configuration Initiale et Prérequis

Avant de commencer, installez les dépendances Python nécessaires. J'utilise personnellement Python 3.11+ pour sa performance sur les opérations asynchrones.

pip install okx-sdk pandas numpy aiohttp python-dotenv websockets

HolySheep SDK optionnel mais recommandé

pip install holysheep-python

Créez votre fichier .env à la racine du projet :

# OKX API Configuration
OKX_API_KEY=your_okx_api_key_here
OKX_PASSPHRASE=your_okx_passphrase
OKX_SECRET_KEY=your_okx_secret_key
OKX_FLAG=0  # 0 pour production, 1 pour demo


HolySheep AI Configuration

HOLYSHEEP_API_KEY=your_holysheep_api_key


Paramètres du market maker

SYMBOL=BTC-USDT
SPREAD_BP=5  # Spread en basis points (0.05%)
MAX_POSITION=0.5  # Position max en BTC
INVESTMENT_USDT=10000

Connexion à OKX via WebSocket

La première étape critique : récupérer les données du carnet d'ordres en temps réel. C'est sur ces données que repose toute la logique de pricing. J'ai constaté que la latence médiane avec OKX est d'environ 45ms, ce qui reste acceptable pour du market making sur des paires liquides.

import json
import asyncio
import websockets
from typing import Dict, List, Optional
from dataclasses import dataclass

@dataclass
class OrderBookEntry:
    price: float
    size: float
    side: str  # 'bid' ou 'ask'

class OKXWebSocketClient:
    """Client WebSocket pour OKX avec reconnexion automatique."""
    
    def __init__(self, api_key: str, passphrase: str, secret_key: str, flag: int = 0):
        self.api_key = api_key
        self.passphrase = passphrase
        self.secret_key = secret_key
        self.flag = flag
        self.ws = None
        self.orderbook: Dict[str, List[OrderBookEntry]] = {'bids': [], 'asks': []}
        self.callbacks: List[callable] = []
        self._running = False
    
    async def connect(self, symbol: str = "BTC-USDT-SWAP"):
        """Connexion au flux WebSocket OKX.
        
        Note: endpoint OKX public, ne nécessite pas d'authentification
        """
        url = f"wss://ws.okx.com:8443/ws/v5/public"
        
        while True:
            try:
                async with websockets.connect(url, ping_interval=30) as ws:
                    self.ws = ws
                    self._running = True
                    
                    # Subscribe aux données du carnet d'ordres
                    subscribe_msg = {
                        "op": "subscribe",
                        "args": [{
                            "channel": "books5",  # 5 niveaux de profondeur
                            "instId": symbol
                        }]
                    }
                    await ws.send(json.dumps(subscribe_msg))
                    print(f"✅ Connecté au flux {symbol}")
                    
                    await self._listen()
                    
            except websockets.exceptions.ConnectionClosed as e:
                print(f"⚠️ Connexion fermée: {e.code}. Reconnexion dans 5s...")
                await asyncio.sleep(5)
            except Exception as e:
                print(f"❌ Erreur: {e}. Reconnexion dans 10s...")
                await asyncio.sleep(10)
    
    async def _listen(self):
        """Boucle principale d'écoute des messages."""
        async for message in self.ws:
            if not self._running:
                break
                
            data = json.loads(message)
            
            if 'data' in data:
                for entry in data['data']:
                    self._update_orderbook(entry)
                    # Notifier les callbacks
                    for callback in self.callbacks:
                        await callback(self.orderbook)
    
    def _update_orderbook(self, data: dict):
        """Met à jour le carnet d'ordres local."""
        bids = [
            OrderBookEntry(float(p), float(s), 'bid')
            for p, s in zip(data['bids'][:5], data['bsizes'][:5])
        ]
        asks = [
            OrderBookEntry(float(p), float(s), 'ask')
            for p, s in zip(data['asks'][:5], data['asizes'][:5])
        ]
        self.orderbook = {'bids': bids, 'asks': asks}
    
    def register_callback(self, callback: callable):
        """Enregistre une fonction de callback pour les mises à jour."""
        self.callbacks.append(callback)


Exemple d'utilisation

async def demo_orderbook_update(orderbook):
    best_bid = orderbook['bids'][0].price if orderbook['bids'] else None
    best_ask = orderbook['asks'][0].price if orderbook['asks'] else None
    if best_bid and best_ask:
        mid_price = (best_bid + best_ask) / 2
        spread_bp = ((best_ask - best_bid) / mid_price) * 10000
        print(f"Best Bid: {best_bid}, Best Ask: {best_ask}, Spread: {spread_bp:.2f} bp")


Lancement (décommenter pour tester)


ws_client = OKXWebSocketClient(api_key="", passphrase="", secret_key="")


ws_client.register_callback(demo_orderbook_update)


asyncio.run(ws_client.connect("BTC-USDT-SWAP"))

Ressources connexes
Articles connexes