Introduction aux données de chandeliers sur OKX

Les chandeliers japonais (candlesticks en anglais) représentent l'une des visualisations les plus utilisées dans le trading de cryptomonnaies. Sur OKX, chaque chandelier contient quatre informations essentielles : le prix d'ouverture, le prix de clôture, le prix le plus haut et le prix le plus bas pour une période donnée. Comprendre comment agréger ces données est fundamental pour créer des indicateurs techniques, des robots de trading ou des analyses de marché. Dans ce tutoriel, je vais vous guider pas à pas depuis zéro. Aucune expérience préalable avec les API ou le trading n'est nécessaire. Nous allons explorer ensemble les différentes méthodes pour récupérer, transformer et analyser ces données de chandeliers OKX.

Qu'est-ce qu'un chandelier et pourquoi l'agréger ?

Un chandelier (candle) représente l'activité de trading sur une période spécifique. Voici sa structure : L'agrégation consiste à combiner plusieurs chandeliers de petite timeframe (par exemple 1 minute) pour créer des chandeliers de plus grande timeframe (5 minutes, 1 heure, 1 jour). Cette technique est essentielle pour l'analyse technique et la création de stratégies de trading automatisées.

Récupérer les données brutes depuis l'API OKX

Avant d'agréger, nous devons d'abord récupérer les données. L'API REST d'OKX permet d'obtenir les chandeliers historiques. Voici comment faire avec Python : 
import requests
import time

def get_okx_candles(instId="BTC-USDT", bar="1m", limit=100):
    """
    Récupère les chandeliers depuis l'API OKX
    
    Paramètres:
    - instId: Identifiant de l'instrument (ex: BTC-USDT)
    - bar: Timeframe (1m, 5m, 15m, 1H, 4H, 1D, etc.)
    - limit: Nombre de chandeliers à récupérer (max 100)
    """
    url = "https://www.okx.com/api/v5/market/history-candles"
    params = {
        "instId": instId,
        "bar": bar,
        "limit": limit
    }
    
    headers = {
        "Content-Type": "application/json"
    }
    
    response = requests.get(url, params=params, headers=headers)
    
    if response.status_code == 200:
        data = response.json()
        if data.get("code") == "0":
            return data["data"]
        else:
            print(f"Erreur API: {data.get('msg')}")
            return None
    else:
        print(f"Erreur HTTP: {response.status_code}")
        return None


Exemple d'utilisation

candles = get_okx_candles(instId="BTC-USDT", bar="1m", limit=100)
print(f"Nombre de chandeliers récupérés: {len(candles) if candles else 0}")
[Capture d'écran suggérée] : Console Python affichant les données JSON brutes avec la structure timestamp, open, high, low, close, vol.

Méthode 1 : Agrégation simple avec Pandas

La bibliothèque Pandas est l'outil le plus puissant pour manipuler des données financières en Python. Voici comment agréger des chandeliers de 1 minute en chandeliers de 5 minutes : 
import pandas as pd
import requests

def get_and_aggregate_candles(instId="BTC-USDT", source_bar="1m", target_bar="5m", limit=300):
    """
    Récupère les chandeliers et les aggrège vers une timeframe supérieure
    
    Timeframes supportées pour l'agrégation:
    - 1m vers: 5m, 15m, 30m
    - 5m vers: 15m, 30m, 1H
    - 1H vers: 4H, 1D
    """
    # Étape 1 : Récupérer les données brutes
    url = "https://www.okx.com/api/v5/market/history-candles"
    params = {"instId": instId, "bar": source_bar, "limit": limit}
    response = requests.get(url, params=params)
    data = response.json()["data"]
    
    # Étape 2 : Convertir en DataFrame
    columns = ["timestamp", "open", "high", "low", "close", "vol", "volCcy", "volQuote"]
    df = pd.DataFrame(data, columns=columns)
    
    # Convertir les types
    for col in ["open", "high", "low", "close", "vol"]:
        df[col] = pd.to_numeric(df[col], errors='coerce')
    
    df["timestamp"] = pd.to_datetime(df["timestamp"].astype(float), unit="ms")
    df = df.sort_values("timestamp").reset_index(drop=True)
    
    # Étape 3 : Définir la fréquence d'agrégation
    freq_map = {
        "1m": "5T", "5m": "15T", "15m": "30T",
        "30m": "1H", "1H": "4H", "4H": "1D", "1D": "1W"
    }
    freq = freq_map.get(source_bar, "5T")
    
    # Étape 4 : Agréger les chandeliers
    aggregated = df.set_index("timestamp").resample(freq).agg({
        "open": "first",
        "high": "max",
        "low": "min",
        "close": "last",
        "vol": "sum"
    }).dropna()
    
    aggregated = aggregated.reset_index()
    return aggregated


Exemple : Convertir 1 minute en 5 minutes

candles_5m = get_and_aggregate_candles(
    instId="BTC-USDT",
    source_bar="1m",
    target_bar="5m",
    limit=300
)
print(candles_5m.head(10))
[Capture d'écran suggérée] : DataFrame Pandas avec les colonnes timestamp, open, high, low, close, vol après agrégation.

Méthode 2 : Agrégation avec HolySheep AI

Maintenant, voici la partie passionnante. En tant qu'utilisateur quotidien de HolySheep AI, je peux vous dire que l'utilisation de leur API pour traiter les données de chandeliers change complètement la donne. Leur latence inférieure à 50ms et leurs tarifs avantageux (DeepSeek V3.2 à $0.42/MTok) rendent le traitement massif de données financières extrêmement économique. Si vous souhaitez vous inscrire ici sur HolySheep AI, vous recevrez des crédits gratuits pour tester ces méthodes. 
import requests
import json

def aggregate_with_holysheep(candles_data, target_timeframe):
    """
    Utilise l'IA de HolySheep pour analyser et agréger les chandeliers
    """
    base_url = "https://api.holysheep.ai/v1"
    
    # Préparer le prompt pour l'agrégation intelligente
    prompt = f"""
    Tu es un analyste financier expert. Analyse ces {len(candles_data)} chandeliers de 1 minute
    et aggrège-les en chandeliers de {target_timeframe}.
    
    Pour chaque chandelier agrégé, calcule :
    - Open: premier prix de la période
    - High: prix maximum
    - Low: prix minimum
    - Close: dernier prix
    - Volume: somme des volumes
    
    Retourne les résultats au format JSON avec les champs :
    timestamp, open, high, low, close, volume
    
    Données brutes : {json.dumps(candles_data[:10])}
    """
    
    headers = {
        "Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
        "Content-Type": "application/json"
    }
    
    payload = {
        "model": "deepseek-v3.2",
        "messages": [
            {"role": "system", "content": "Tu es un analyste financier expert en cryptomonnaies."},
            {"role": "user", "content": prompt}
        ],
        "temperature": 0.1
    }
    
    response = requests.post(
        f"{base_url}/chat/completions",
        headers=headers,
        json=payload
    )
    
    if response.status_code == 200:
        result = response.json()
        return result["choices"][0]["message"]["content"]
    else:
        print(f"Erreur: {response.status_code}")
        return None


Exemple d'utilisation avec des données réelles

sample_candles = [
    {"timestamp": 1704067200000, "open": 42150.5, "high": 42200.0, "low": 42100.0, "close": 42180.0, "vol": 125.5},
    {"timestamp": 1704067260000, "open": 42180.0, "high": 42250.0, "low": 42170.0, "close": 42230.0, "vol": 98.3},
    {"timestamp": 1704067320000, "open": 42230.0, "high": 42280.0, "low": 42200.0, "close": 42250.0, "vol": 110.2},
    # ... plus de chandeliers
]

result = aggregate_with_holysheep(sample_candles, "5m")
print(result)

Méthode 3 : Agrégation en temps réel avec WebSocket

Pour le trading en direct, vous aurez besoin d'agréger les données en temps réel. Cette méthode utilise les WebSockets d'OKX : 
import json
import threading
import time
from collections import defaultdict

class RealTimeAggregator:
    """
    Agrège les chandeliers en temps réel via WebSocket OKX
    """
    
    def __init__(self, instId="BTC-USDT", source_bar="1m", target_bar="5m"):
        self.instId = instId
        self.source_bar = source_bar
        self.target_bar = target_bar
        self.candles_buffer = defaultdict(dict)
        self.aggregated_candles = []
        
    def get_bar_ms(self, bar):
        """Convertit la timeframe en millisecondes"""
        bar_map = {
            "1m": 60000, "5m": 300000, "15m": 900000,
            "1H": 3600000, "4H": 14400000, "1D": 86400000
        }
        return bar_map.get(bar, 60000)
    
    def aggregate_tick(self, tick_data):
        """
        Agregate un chandelier source en chandelier cible
        """
        timestamp = tick_data[0]
        open_price = float(tick_data[1])
        high_price = float(tick_data[2])
        low_price = float(tick_data[3])
        close_price = float(tick_data[4])
        volume = float(tick_data[5])
        
        # Calculer la période cible
        source_ms = self.get_bar_ms(self.source_bar)
        target_ms = self.get_bar_ms(self.target_bar)
        
        # Trouver l'index de la période cible
        target_period = (timestamp // target_ms) * target_ms
        
        if target_period not in self.candles_buffer:
            # Nouveau chandelier agrégé
            self.candles_buffer[target_period] = {
                "open": open_price,
                "high": high_price,
                "low": low_price,
                "close": close_price,
                "volume": volume,
                "first_source_time": timestamp
            }
        else:
            # Mettre à jour le chandelier existant
            candle = self.candles_buffer[target_period]
            candle["high"] = max(candle["high"], high_price)
            candle["low"] = min(candle["low"], low_price)
            candle["close"] = close_price
            candle["volume"] += volume
    
    def get_current_aggregated(self):
        """Retourne le chandelier agrégé actuel"""
        if self.candles_buffer:
            last_period = max(self.candles_buffer.keys())
            return self.candles_buffer[last_period]
        return None
    
    def start(self):
        """Démarre l'écoute des WebSockets (simulation)"""
        print(f"Agrégateur démarré: {self.source_bar} -> {self.target_bar}")
        print("En attente de données...")
        
        # Simulation de données en temps réel
        import random
        base_price = 42000
        
        for i in range(20):
            tick = [
                int(time.time() * 1000),
                str(base_price),
                str(base_price + random.uniform(0, 50)),
                str(base_price - random.uniform(0, 30)),
                str(base_price + random.uniform(-20, 30)),
                str(random.uniform(1, 10))
            ]
            self.aggregate_tick(tick)
            time.sleep(0.5)
            
            current = self.get_current_aggregated()
            print(f"5m Candle: O={current['open']:.2f} H={current['high']:.2f} L={current['low']:.2f} C={current['close']:.2f} V={current['volume']:.2f}")


Lancer l'agrégateur

aggregator = RealTimeAggregator(instId="BTC-USDT", source_bar="1m", target_bar="5m")
aggregator.start()

Calculer les indicateurs techniques après aggregation

Une fois vos chandeliers agrégés, vous pouvez calculer des indicateurs techniques. Voici comment calculer une moyenne mobile simple (SMA) et l'Indice de Force Relative (RSI) : 
import pandas as pd

def calculate_indicators(df, sma_period=20, rsi_period=14):
    """
    Calcule les indicateurs techniques sur des chandeliers agrégés
    """
    # Moyenne Mobile Simple (SMA)
    df['sma_20'] = df['close'].rolling(window=sma_period).mean()
    
    # Moyenne Mobile Exponentielle (EMA)
    df['ema_12'] = df['close'].ewm(span=12, adjust=False).mean()
    df['ema_26'] = df['close'].ewm(span=26, adjust=False).mean()
    
    # MACD
    df['macd'] = df['ema_12'] - df['ema_26']
    df['signal_line'] = df['macd'].ewm(span=9, adjust=False).mean()
    df['macd_histogram'] = df['macd'] - df['signal_line']
    
    # RSI
    delta = df['close'].diff()
    gain = (delta.where(delta > 0, 0)).rolling(window=rsi_period).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(window=rsi_period).mean()
    rs = gain / loss
    df['rsi'] = 100 - (100 / (1 + rs))
    
    # Bandes de Bollinger
    df['bb_middle'] = df['close'].rolling(window=20).mean()
    df['bb_std'] = df['close'].rolling(window=20).std()
    df['bb_upper'] = df['bb_middle'] + (df['bb_std'] * 2)
    df['bb_lower'] = df['bb_middle'] - (df['bb_std'] * 2)
    
    return df


Appliquer sur nos chandeliers agrégés

df_with_indicators = calculate_indicators(candles_5m)
print(df_with_indicators[['timestamp', 'close', 'sma_20', 'rsi']].tail(10))

Visualisation des données agrégées

Pour visualiser vos chandeliers, utilisez matplotlib ou plotly : 
import plotly.graph_objects as go

def plot_candlesticks(df, title="Chandeliers OKX"):
    """
    Crée un graphique en chandeliers interactif avec Plotly
    """
    fig = go.Figure(data=[
        go.Candlestick(
            x=df['timestamp'],
            open=df['open'],
            high=df['high'],
            low=df['low'],
            close=df['close'],
            name="BTC-USDT"
        ),
        go.Scatter(
            x=df['timestamp'],
            y=df['sma_20'],
            mode='lines',
            name=f"SMA 20",
            line=dict(color='blue', width=1)
        )
    ])
    
    fig.update_layout(
        title=title,
        yaxis_title='Prix (USDT)',
        xaxis_title='Temps',
        template='plotly_dark',
        xaxis_rangeslider_visible=False
    )
    
    fig.show()
    return fig


Afficher le graphique

plot_candlesticks(candles_5m, "BTC-USDT - Chandeliers 5 minutes avec SMA 20")
[Capture d'écran suggérée] : Graphique Plotly montrant des chandeliers japonais avec la moyenne mobile SMA 20 superposée.

Comparatif des méthodes d'agrégation

MéthodeLatenceCoûtComplexitéIdéal pour
Pandas pur~100-500msGratuitMoyenneBacktesting, analyses historiques
HolySheep AI<50ms$0.42/MTokBasseAnalyses intelligentes, automatisation
WebSocket temps réel<10msGratuitHauteTrading en direct, bots
C++ custom<1msGratuitTrès hauteHaute fréquence (HFT)

Pour qui / pour qui ce n'est pas fait

Cette méthode est faite pour vous si : Cette méthode n'est PAS faite pour vous si :

Tarification et ROI

Comparons les coûts pour traiter 1 million de chandeliers OKX :
Fournisseur Prix 2026/MTok Coût pour 1M candles Latence
OpenAI GPT-4.1 $8.00 ~$16 200-500ms
Anthropic Claude Sonnet 4.5 $15.00 ~$30 300-600ms
Google Gemini 2.5 Flash $2.50 ~$5 100-200ms
HolySheep DeepSeek V3.2 $0.42 ~$0.84 <50ms
Économie avec HolySheep : En utilisant HolySheep AI au lieu de GPT-4.1, vous économisez plus de 85% sur vos coûts d'analyse. Avec le taux avantageux de ¥1=$1, les paiements via WeChat ou Alipay sont instantanés et sans friction.

Pourquoi choisir HolySheep

En tant qu'utilisateur depuis plusieurs mois, voici pourquoi je recommande HolySheep AI pour l'agrégation de données OKX : J'ai réduit mes coûts d'analyse de données de trading de $200/mois à moins de $30/mois en migrant vers HolySheep. La qualité de service est identique, voire meilleure sur certains aspects.

Erreurs courantes et solutions

Erreur 1 : "Invalid instrument ID" lors de l'appel API

Cause : L'identifiant de l'instrument OKX est mal formaté. OKX utilise des formats spécifiques comme "BTC-USDT" ou "ETH-USDT-SWAP" (pour les contrats perpétuels). Solution : 
# ❌ INCORRECT
instId = "btcusdt"  # minuscules
instId = "BTC_USDT"  # underscore au lieu du tiret
instId = "BTCUSDT"   # sans séparateur


✅ CORRECT

instId = "BTC-USDT"              # Spot
instId = "BTC-USDT-SWAP"         # Contrat perpétuel
instId = "ETH-USD-231229"        # Contrat à terme avec date d'expiration


Vérification du format

def validate_instId(instId):
    valid_formats = [
        "BTC-USDT", "ETH-USDT", "SOL-USDT",
        "BTC-USDT-SWAP", "ETH-USDT-SWAP",
        "BTC-USD-231229"
    ]
    return instId in valid_formats

print(validate_instId("BTC-USDT"))  # True

Erreur 2 : "rate limit exceeded" avec l'API OKX

Cause : Vous effectuez trop de requêtes par seconde. OKX limite à 20 requêtes par seconde en lecture. Solution : 
import time
import requests
from ratelimit import limits, sleep_and_retry

@sleep_and_retry
@limits(calls=18, period=1)  # 18 appels par seconde (marge de sécurité)
def get_candles_with_limit(instId, bar, limit):
    url = "https://www.okx.com/api/v5/market/history-candles"
    params = {"instId": instId, "bar": bar, "limit": limit}
    
    response = requests.get(url, params=params)
    
    if response.status_code == 429:
        # Attendre et réessayer
        time.sleep(2)
        return get_candles_with_limit(instId, bar, limit)
    
    return response.json()


Utilisation

data = get_candles_with_limit("BTC-USDT", "1m", 100)

Erreur 3 : "NaN values in aggregated candles"

Cause : Des chandeliers sont manquants dans la période d'agrégation, créant des valeurs nulles après resampling. Solution : 
import pandas as pd

def aggregate_with_gap_filling(df, freq="5T", fill_method="ffill"):
    """
    Agrège avec gestion des gaps dans les données
    
    Paramètres:
    - df: DataFrame avec colonnes timestamp, open, high, low, close, vol
    - freq: Fréquence cible ('5T' = 5 minutes)
    - fill_method: 'ffill' (forward fill), 'bfill', ou interpolation
    """
    # S'assurer que l'index est le timestamp
    df = df.set_index("timestamp")
    
    # Resample avec une agrégation robuste
    aggregated = df.resample(freq).agg({
        "open": "first",
        "high": "max",
        "low": "min",
        "close": "last",
        "vol": "sum"
    })
    
    # Compter les chandeliers source par période
    count_per_period = df.resample(freq).size()
    aggregated["source_count"] = count_per_period
    
    # Remplir les valeurs manquantes
    if fill_method == "ffill":
        aggregated[["open", "high", "low", "close"]] = \
            aggregated[["open", "high", "low", "close"]].fillna(method="ffill")
    elif fill_method == "interpolate":
        aggregated[["open", "high", "low", "close"]] = \
            aggregated[["open", "high", "low", "close"]].interpolate()
    
    # Supprimer les périodes avec moins de 50% des chandeliers source
    min_candles = {"5T": 3, "15T": 8, "1H": 25, "4H": 100}
    threshold = min_candles.get(freq, 1)
    aggregated = aggregated[aggregated["source_count"] >= threshold]
    
    return aggregated.dropna()


Appliquer la correction

clean_candles = aggregate_with_gap_filling(df_with_candles, freq="5T")

Conclusion et prochaines étapes

Vous disposez maintenant de trois méthodes puissantes pour agréger les données de chandeliers OKX :
  1. Pandas : Idéale pour le backtesting et les analyses historiques
  2. HolySheep AI : Parfaite pour l'automatisation intelligente avec un excellent rapport qualité-prix
  3. WebSocket : Indispensable pour le trading en temps réel
La combinaison de ces méthodes vous permettra de construire des systèmes de trading robustes et performants. N'oubliez pas de toujours tester vos stratégies sur des données historiques avant de les déployer en production. Pour commencer à expérimenter avec HolySheep AI et profiter de leurs tarifs imbattables et leur latence ultra-faible, je vous recommande de vous inscrire ici dès maintenant. 👉 Inscrivez-vous sur HolySheep AI — crédits offerts

Ressources connexes

Articles connexes