En tant qu'ingénieur quantitatif avec plus de 8 ans d'expérience dans le développement de stratégies de trading algorithmique, j'ai testé des dizaines de fournisseurs de données. Après avoir brûlé des centaines de dollars en abonnements inutiles et passé des nuits blanches à déboguer des connexions API capricieuses, je peux vous dire avec certitude : l'accès aux données historiques de Bybit est un cauchemar si vous ne savez pas par où commencer.
Aujourd'hui, je vais vous guider pas à pas depuis zéro. Pas de jargon technique barbant. Juste un chemin clair pour accéder aux K-lines (chandeliers japonais) historiques et aux données tick-by-tick de Bybit perpetual futures, puis les intégrer dans un pipeline de backtesting professionnel via Tardis.
Ce que vous allez apprendre
- Comprendre la structure des données Bybit perpetual futures
- Configurer votre environnement et obtenir vos premières données
- Intégrer l'API Tardis pour un accès fiable aux données historiques
- Construire un pipeline de backtesting complet
- Éviter les erreurs coûteuses que j'ai commises
Prérequis : ce dont vous avez besoin
Pour suivre ce tutoriel, vous aurez besoin de :
- Un ordinateur avec Python 3.9+ installé
- Un compte Bybit (nous verrons cela ensemble)
- Un abonnement Tardis (ou une alternative que je vous présenterai)
- 30 minutes de votre temps et beaucoup de café
💡 Astuce personnelle : J'ai commencé avec seulement 50$ sur mon compte démo. Ne vous sentez pas obligé d'investir des milliers d'euros pour apprendre. Le plus précieux n'est pas l'argent, c'est la connaissance du flux de données.
Comprendre Bybit Perpetual Futures
Qu'est-ce qu'un contrat perpétuel ?
Un contrat perpétuel Bybit est un produit dérivé qui vous permet de trader des crypto-actifs sans date d'expiration. Contrairement aux contrats à terme classiques qui expirent chaque trimestre, les perpetual futures permettent de maintenir une position indéfiniment.
Le symbole que nous allons utiliser dans ce tutoriel est BTCUSD (Bitcoin contre USD) sur Bybit. C'est le contrat le plus liquide et le plus utilisé pour les stratégies de backtesting.
Structure d'une K-line (chandelier japonais)
Chaque K-line représente un intervalle de temps et contient ces informations essentielles :
- Timestamp : moment exact du début de la période
- Open : prix d'ouverture
- High : prix le plus haut
- Low : prix le plus bas
- Close : prix de clôture
- Volume : volume échangé
Configuration de l'environnement
Installation des dépendances Python
# Créer un environnement virtuel (recommandé)
python -m venv trading_env
source trading_env/bin/activate # Linux/Mac
trading_env\Scripts\activate # Windows
Installer les bibliothèques nécessaires
pip install requests pandas numpy tardis-client
pip install mplfinance backtesting # Pour la visualisation
pip install jupyter notebook # Interface de développementVérification de votre installation
#!/usr/bin/env python3
"""Script de vérification de l'environnement de trading"""
import sys
print(f"Python version: {sys.version}")
try:
import requests
print(f"✓ requests version: {requests.__version__}")
except ImportError:
print("✗ requests non installé")
try:
import pandas as pd
print(f"✓ pandas version: {pd.__version__}")
except ImportError:
print("✗ pandas non installé")
try:
import tardis
print("✓ tardis-client installé")
except ImportError:
print("✗ tardis-client non installé")
print("\n🎯 Votre environnement est prêt!")Connexion à l'API Tardis
Pourquoi Tardis ?
Tardis est un fournisseur de données spécialisé dans les marchés de crypto-actifs. Contrairement à certaines alternatives qui limitent les requêtes ou facturent des prix prohibitifs, Tardis offre :
- Accès à plus de 50 exchanges
- Données tick-by-tick en temps réel
- Historiques sur plusieurs années
- API stable et bien documentée
Configuration de la clé API Tardis
#!/usr/bin/env python3
"""
Module de configuration pour l'API Tardis
Remplacez les variables ci-dessous par vos propres clés
"""
import os
from dotenv import load_dotenv
Charger les variables d'environnement depuis .env
load_dotenv()
Configuration Tardis
TARDIS_API_KEY = os.getenv("TARDIS_API_KEY", "YOUR_TARDIS_API_KEY")
TARDIS_API_SECRET = os.getenv("TARDIS_API_SECRET", "YOUR_TARDIS_API_SECRET")
Configuration HolySheep (notre alternative recommandée)
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
Configuration Bybit
BYBIT_WS_URL = "wss://stream.bybit.com/v5/public/linear"
BYBIT_REST_URL = "https://api.bybit.com/v5"
Validation des clés API
def validate_api_keys():
"""Valide que les clés API sont configurées"""
errors = []
if TARDIS_API_KEY == "YOUR_TARDIS_API_KEY":
errors.append("Clé API Tardis non configurée")
if HOLYSHEEP_API_KEY == "YOUR_HOLYSHEEP_API_KEY":
print("⚠️ Clé API HolySheep non configurée (optionnel)")
if errors:
raise ValueError("\n".join(errors))
return True
if __name__ == "__main__":
validate_api_keys()
print("✓ Toutes les clés API sont configurées")Récupération des données K-line historiques
Méthode 1 : Via l'API Bybit directe
#!/usr/bin/env python3
"""
Récupération des K-lines historiques Bybit via l'API REST
"""
import requests
import pandas as pd
from datetime import datetime, timedelta
class BybitKlineFetcher:
"""Classe pour récupérer les données K-line depuis Bybit"""
BASE_URL = "https://api.bybit.com/v5"
def __init__(self):
self.session = requests.Session()
self.session.headers.update({
"Content-Type": "application/json"
})
def get_klines(
self,
symbol: str = "BTCUSD",
category: str = "linear",
interval: str = "1", # 1, 3, 5, 15, 30, 60, 120, 240, 360, 720, D, W, M
start_time: int = None,
end_time: int = None,
limit: int = 200 # Maximum 1000
) -> pd.DataFrame:
"""
Récupère les K-lines historiques
Args:
symbol: Symbole du contrat (ex: BTCUSD)
category: Type de contrat (linear, spot, option)
interval: Intervalle de temps en minutes
start_time: Timestamp Unix en millisecondes
end_time: Timestamp Unix en millisecondes
limit: Nombre maximum de résultats (max 1000)
Returns:
DataFrame pandas avec les données K-line
"""
endpoint = f"{self.BASE_URL}/market/kline"
params = {
"category": category,
"symbol": symbol,
"interval": interval,
"limit": limit
}
if start_time:
params["start"] = start_time
if end_time:
params["end"] = end_time
try:
response = self.session.get(endpoint, params=params, timeout=30)
response.raise_for_status()
data = response.json()
if data.get("retCode") == 0:
klines = data["result"]["list"]
# Conversion en DataFrame
df = pd.DataFrame(klines, columns=[
"start_time", "open", "high", "low", "close", "volume", "turnover"
])
# Conversion des types
df["start_time"] = pd.to_datetime(df["start_time"].astype(int), unit="ms")
for col in ["open", "high", "low", "close", "volume", "turnover"]:
df[col] = pd.to_numeric(df[col])
# Tri chronologique
df = df.sort_values("start_time").reset_index(drop=True)
return df
else:
raise Exception(f"Erreur API Bybit: {data.get('retMsg')}")
except requests.exceptions.RequestException as e:
print(f"Erreur de connexion: {e}")
return pd.DataFrame()
Utilisation
if __name__ == "__main__":
fetcher = BybitKlineFetcher()
# Récupérer les 1000 dernières K-lines 1H
df = fetcher.get_klines(
symbol="BTCUSD",
interval="60",
limit=1000
)
print(f"📊 {len(df)} K-lines récupérées")
print(df.tail())Méthode 2 : Via Tardis (Recommandé pour le backtesting)
#!/usr/bin/env python3
"""
Récupération des données via Tardis API
Tardis offre un accès plus fiable et des données plus complètes
"""
import requests
import pandas as pd
from datetime import datetime, timedelta
import time
class TardisDataFetcher:
"""Classe pour récupérer les données via l'API Tardis"""
BASE_URL = "https://api.tardis.dev/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}"
})
def get_symbols(self, exchange: str = "bybit") -> list:
"""Récupère la liste des symboles disponibles"""
endpoint = f"{self.BASE_URL}/exchanges/{exchange}/symbols"
response = self.session.get(endpoint, timeout=30)
data = response.json()
return [s["symbol"] for s in data]
def get_historical_klines(
self,
exchange: str,
symbol: str,
start_date: str,
end_date: str,
interval: str = "1m",
limit: int = 10000
) -> pd.DataFrame:
"""
Récupère les K-lines historiques depuis Tardis
Args:
exchange: Nom de l'exchange (ex: bybit)
symbol: Symbole de trading (ex: BTC-PERPETUAL)
start_date: Date de début (YYYY-MM-DD)
end_date: Date de fin (YYYY-MM-DD)
interval: Intervalle (1m, 5m, 1h, 1d)
limit: Limite de результатов par запрос
Returns:
DataFrame avec les données OHLCV
"""
endpoint = f"{self.BASE_URL}/historical/{exchange}/{symbol}/klines"
params = {
"start_date": start_date,
"end_date": end_date,
"interval": interval,
"limit": limit
}
all_klines = []
current_start = start_date
print(f"📥 Téléchargement des données {symbol} ({start_date} → {end_date})...")
while current_start < end_date:
params["start_date"] = current_start
try:
response = self.session.get(endpoint, params=params, timeout=60)
if response.status_code == 200:
data = response.json()
if data:
all_klines.extend(data)
# Passer à la prochaine période
last_timestamp = pd.to_datetime(data[-1]["timestamp"])
current_start = last_timestamp.strftime("%Y-%m-%d")
print(f" ✓ {len(data)} records récupérés jusqu'à {current_start}")
# Respecter les limites de taux
time.sleep(0.5)
else:
break
else:
print(f"⚠️ Erreur {response.status_code}: {response.text}")
break
except Exception as e:
print(f"❌ Erreur: {e}")
time.sleep(5)
# Conversion en DataFrame
if all_klines:
df = pd.DataFrame(all_klines)
df["timestamp"] = pd.to_datetime(df["timestamp"])
# Standardisation des colonnes
df = df.rename(columns={
"open": "Open",
"high": "High",
"low": "Low",
"close": "Close",
"volume": "Volume"
})
print(f"✅ Total: {len(df)} K-lines récupérées")
return df
return pd.DataFrame()
def get_historical_trades(
self,
exchange: str,
symbol: str,
start_date: str,
end_date: str,
limit: int = 50000
) -> pd.DataFrame:
"""
Récupère les trades tick-by-tick
Returns:
DataFrame avec les trades individuels
"""
endpoint = f"{self.BASE_URL}/historical/{exchange}/{symbol}/trades"
params = {
"start_date": start_date,
"end_date": end_date,
"limit": limit
}
all_trades = []
print(f"📥 Téléchargement des trades {symbol}...")
response = self.session.get(endpoint, params=params, timeout=60)
if response.status_code == 200:
data = response.json()
for trade_batch in data:
all_trades.extend(trade_batch)
print(f"✅ {len(all_trades)} trades récupérés")
return pd.DataFrame(all_trades)
Exemple d'utilisation
if __name__ == "__main__":
API_KEY = "YOUR_TARDIS_API_KEY"
fetcher = TardisDataFetcher(API_KEY)
# Récupérer 1 mois de données BTC 1H
df = fetcher.get_historical_klines(
exchange="bybit",
symbol="BTC-PERPETUAL",
start_date="2026-03-01",
end_date="2026-04-01",
interval="1h"
)
print(df.head())Construction du pipeline de backtesting
Structure du projet
trading_backtest/
├── config/
│ ├── __init__.py
│ └── settings.py
├── data/
│ ├── __init__.py
│ └── fetcher.py
├── strategies/
│ ├── __init__.py
│ ├── base_strategy.py
│ └── example_strategy.py
├── backtest/
│ ├── __init__.py
│ ├── engine.py
│ └── metrics.py
├── notebooks/
│ └── analysis.ipynb
├── main.py
└── requirements.txtImplémentation du moteur de backtest
#!/usr/bin/env python3
"""
Moteur de backtesting pour les stratégies de trading
Version simplifiée pour les débutants
"""
import pandas as pd
import numpy as np
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime
from enum import Enum
class OrderType(Enum):
"""Types d'ordres supportés"""
MARKET = "market"
LIMIT = "limit"
class PositionSide(Enum):
"""Direction de la position"""
LONG = "long"
SHORT = "short"
FLAT = "flat"
@dataclass
class Order:
"""Représentation d'un ordre"""
timestamp: pd.Timestamp
symbol: str
side: PositionSide
order_type: OrderType
quantity: float
price: Optional[float] = None
@dataclass
class Position:
"""Représentation d'une position"""
symbol: str
side: PositionSide
quantity: float
entry_price: float
entry_time: pd.Timestamp
@property
def market_value(self) -> float:
"""Valeur marchande actuelle"""
return self.quantity * self.entry_price
def unrealized_pnl(self, current_price: float) -> float:
"""Calcul du PnL non réalisé"""
if self.side == PositionSide.LONG:
return (current_price - self.entry_price) * self.quantity
else:
return (self.entry_price - current_price) * self.quantity
def unrealized_pnl_pct(self, current_price: float) -> float:
"""PnL en pourcentage"""
if self.entry_price == 0:
return 0.0
return (self.unrealized_pnl(current_price) / (self.entry_price * self.quantity)) * 100
@dataclass
class TradeResult:
"""Résultat d'un trade"""
entry_time: pd.Timestamp
exit_time: pd.Timestamp
side: PositionSide
entry_price: float
exit_price: float
quantity: float
pnl: float
pnl_pct: float
class BacktestEngine:
"""
Moteur de backtesting simplifié
Ce moteur permet de tester des stratégies sur des données historiques
et de calculer les métriques de performance.
"""
def __init__(
self,
initial_capital: float = 10000.0,
commission: float = 0.0004, # 0.04% par trade
slippage: float = 0.0001 # 0.01% de slippage
):
self.initial_capital = initial_capital
self.commission = commission
self.slippage = slippage
# État interne
self.cash = initial_capital
self.positions: List[Position] = []
self.trades: List[TradeResult] = []
self.equity_curve: List[float] = []
self.timestamp: Optional[pd.Timestamp] = None
# Données
self.data: Optional[pd.DataFrame] = None
self.current_idx: int = 0
def load_data(self, df: pd.DataFrame) -> None:
"""
Charge les données de marché
Args:
df: DataFrame avec colonnes Open, High, Low, Close, Volume
"""
required_cols = ["Open", "High", "Low", "Close"]
if not all(col in df.columns for col in required_cols):
raise ValueError(f"Colonnes requises: {required_cols}")
self.data = df.reset_index(drop=True)
print(f"📊 Données chargées: {len(self.data)} lignes")
def _get_current_price(self) -> float:
"""Récupère le prix actuel (clôture)"""
return self.data.iloc[self.current_idx]["Close"]
def _get_next_price(self) -> float:
"""Récupère le prix du prochain bougie (pour exécution)"""
if self.current_idx + 1 < len(self.data):
return self.data.iloc[self.current_idx + 1]["Close"]
return self._get_current_price()
def open_position(
self,
side: PositionSide,
quantity: float,
order_type: OrderType = OrderType.MARKET
) -> bool:
"""
Ouvre une nouvelle position
Args:
side: Direction (LONG ou SHORT)
quantity: Quantité à trader
Returns:
True si la position est ouverte avec succès
"""
if self.positions:
print("⚠️ Une position existe déjà. Fermez-la d'abord.")
return False
# Prix d'exécution (avec slippage)
entry_price = self._get_next_price()
execution_price = entry_price * (1 + self.slippage if side == PositionSide.LONG else 1 - self.slippage)
# Coût total
cost = execution_price * quantity
commission_cost = cost * self.commission
if cost + commission_cost > self.cash:
print(f"⚠️ Capital insuffisant: requis {cost + commission_cost}, disponible {self.cash}")
return False
# Créer la position
position = Position(
symbol="BTCUSD",
side=side,
quantity=quantity,
entry_price=execution_price,
entry_time=self.timestamp
)
self.positions.append(position)
self.cash -= (cost + commission_cost)
print(f"✅ Position {side.value} ouverte: {quantity} @ {execution_price:.2f}")
return True
def close_position(
self,
order_type: OrderType = OrderType.MARKET
) -> Optional[TradeResult]:
"""
Ferme la position actuelle
Returns:
TradeResult avec les détails du trade
"""
if not self.positions:
print("⚠️ Aucune position à fermer")
return None
position = self.positions.pop(0)
# Prix de sortie (avec slippage)
exit_price = self._get_next_price()
execution_price = exit_price * (1 - self.slippage if position.side == PositionSide.LONG else 1 + self.slippage)
# PnL
pnl = position.unrealized_pnl(execution_price)
pnl_pct = position.unrealized_pnl_pct(execution_price)
# Commission
exit_cost = execution_price * position.quantity
commission = exit_cost * self.commission
# Mise à jour du capital
self.cash += (position.market_value + pnl - commission)
result = TradeResult(
entry_time=position.entry_time,
exit_time=self.timestamp,
side=position.side,
entry_price=position.entry_price,
exit_price=execution_price,
quantity=position.quantity,
pnl=pnl - commission,
pnl_pct=pnl_pct - (self.commission * 200) # Approximation
)
self.trades.append(result)
print(f"✅ Position fermée: PnL = {pnl:.2f} ({pnl_pct:.2f}%)")
return result
def run(
self,
strategy,
verbose: bool = True
) -> Dict:
"""
Exécute le backtest
Args:
strategy: Instance de la stratégie à tester
verbose: Afficher les opérations
Returns:
Dictionnaire avec les métriques de performance
"""
if self.data is None:
raise ValueError("Aucune donnée chargée. Utilisez load_data()")
self.trades = []
self.equity_curve = []
print(f"\n🚀 Début du backtest: capital initial = {self.initial_capital}")
print(f" Période: {self.data.index[0]} → {self.data.index[-1]}\n")
for idx in range(len(self.data) - 1):
self.current_idx = idx
self.timestamp = self.data.index[idx]
# Signaux de la stratégie
signals = strategy.generate_signals(
self.data.iloc[:idx + 1],
self.timestamp
)
# Exécution des signaux
for signal in signals:
if signal == "BUY" and not self.positions:
# Stratégie demande d'acheter
price = self._get_current_price()
quantity = (self.cash * 0.95) / price # 95% du capital
self.open_position(PositionSide.LONG, quantity)
elif signal == "SELL" and self.positions:
# Stratégie demande de vendre
self.close_position()
# Mise à jour de l'equity curve
current_equity = self.cash
if self.positions:
current_price = self._get_current_price()
current_equity += self.positions[0].market_value + \
self.positions[0].unrealized_pnl(current_price)
self.equity_curve.append(current_equity)
# Fermer les positions restantes
while self.positions:
self.close_position()
return self.get_metrics()
def get_metrics(self) -> Dict:
"""Calcule les métriques de performance"""
if not self.trades:
return {
"total_trades": 0,
"win_rate": 0.0,
"avg_profit": 0.0,
"avg_loss": 0.0,
"profit_factor": 0.0,
"max_drawdown": 0.0,
"sharpe_ratio": 0.0
}
winning_trades = [t for t in self.trades if t.pnl > 0]
losing_trades = [t for t in self.trades if t.pnl <= 0]
total_wins = sum(t.pnl for t in winning_trades) if winning_trades else 0
total_losses = abs(sum(t.pnl for t in losing_trades)) if losing_trades else 0
# Drawdown maximum
equity_series = pd.Series(self.equity_curve)
running_max = equity_series.expanding().max()
drawdowns = (equity_series - running_max) / running_max
max_drawdown = abs(drawdowns.min()) * 100
# Sharpe Ratio (simplifié)
returns = equity_series.pct_change().dropna()
sharpe = returns.mean() / returns.std() * np.sqrt(252) if returns.std() > 0 else 0
return {
"total_trades": len(self.trades),
"winning_trades": len(winning_trades),
"losing_trades": len(losing_trades),
"win_rate": len(winning_trades) / len(self.trades) * 100,
"total_profit": total_wins,
"total_loss": total_losses,
"net_profit": self.cash - self.initial_capital,
"profit_factor": total_wins / total_losses if total_losses > 0 else float('inf'),
"avg_profit": total_wins / len(winning_trades) if winning_trades else 0,
"avg_loss": total_losses / len(losing_trades) if losing_trades else 0,
"max_drawdown": max_drawdown,
"sharpe_ratio": sharpe,
"final_equity": self.cash
}
print("✅ Moteur de backtesting chargé!")Stratégie d'exemple : Moyenne Mobile
#!/usr/bin/env python3
"""
Stratégie de croisement de moyennes mobiles
Simple mais efficace pour comprendre le backtesting
"""
import pandas as pd
import numpy as np
class MovingAverageCrossoverStrategy:
"""
Stratégie basée sur le croisement de deux moyennes mobiles
- ACHAT quand la MA courte croise la MA longue à la hausse
- VENTE quand la MA courte croise la MA longue à la baisse
"""
def __init__(
self,
short_window: int = 10,
long_window: int = 50
):
self.short_window = short_window
self.long_window = long_window
self.position = None # None, 'long', 'short'
def generate_signals(
self,
data: pd.DataFrame,
current_time: pd.Timestamp
) -> list:
"""
Génère les signaux de trading
Args:
data: DataFrame avec les prix historiques
current_time: Timestamp actuel
Returns:
Liste de signaux ['BUY', 'SELL', 'HOLD']
"""
if len(data) < self.long_window:
return []
# Calculer les moyennes mobiles
data = data.copy()
data["ma_short"] = data["Close"].rolling(window=self.short_window).mean()
data["ma_long"] = data["Close"].rolling(window=self.long_window).mean()
# Récupérer les valeurs actuelles et précédentes
current_row = data.iloc[-1]
prev_row = data.iloc[-2]
signals = []
# Croisement haussier (MA courte passe au-dessus de MA longue)
if prev_row["ma_short"] <= prev_row["ma_long"] and \
current_row["ma_short"] > current_row["ma_long"]:
if self.position != "long":
signals.append("BUY")
self.position = "long"
# Croisement baissier (MA courte passe en-dessous de MA longue)
elif prev_row["ma_short"] >= prev_row["ma_long"] and \
current_row["ma_short"] < current_row["ma_long"]:
if self.position == "long":
signals.append("SELL")
self.position = None
return signals
Test de la stratégie
if __name__ == "__main__":
from backtest.engine import BacktestEngine
# Créer des données de test
np.random.seed(42)
dates = pd.date_range("2026-01-01", periods=200, freq="1h")
prices = 40000 + np.cumsum(np.random.randn(200) * 100)
test_data = pd.DataFrame({
"Open": prices,
"High": prices * 1.01,
"Low": prices * 0.99,
"Close": prices * 1.002,
"Volume": np.random.randint(100, 1000, 200)
}, index=dates)
# Initialiser la stratégie
strategy = MovingAverageCrossoverStrategy(
short_window=10,
long_window=30
)
# Lancer le backtest
engine = BacktestEngine(initial_capital=10000)
engine.load_data(test_data)
metrics = engine.run(strategy)
print("\n" + "="*50)
print("📊 RÉSULTATS DU BACKTEST")
print("="*50)
for key, value in metrics.items():
print(f" {key}: {value}")Optimisation et gestion des erreurs
Lors de mes premiers backtests, j'ai rencontré de nombreux problèmes qui m'ont fait perdre du temps précieux. Voici les meilleures pratiques que j'ai apprises :
Validation des données
#!/usr/bin/env python3
"""
Module de validation des données
Évite les erreurs de backtesting dues à des données corrompues
"""
import pandas as pd
import numpy as np
def validate_kline_data(df: pd.DataFrame) -> dict:
"""
Valide la qualité des données K-line
Returns:
Dictionary avec les résultats de validation
"""
results = {
"is_valid": True,
"warnings": [],
"errors": []
}
# Vérifier les colonnes requises
required_cols = ["Open", "High", "Low", "Close", "Volume"]
missing_cols = [col for col in required_cols if col not in df.columns]
if missing_cols:
results["is_valid"] = False
results["errors"].append(f"Colonnes manquantes: {missing_cols}")
# Vérifier les NaN
nan_counts = df[required_cols].isna().sum()
if nan_counts.any():
for col, count in nan_counts.items():
if count > 0:
results["warnings"].append(f"{col}: {count} valeurs NaN")
# Vérifier les cohérences OHLC
invalid_high = df[df["High"] < df["Low"]]
if len(invalid_high) > 0:
results["is_valid"] = False
results["errors"].append(f"High < Low dans {len(invalid_high)} lignes")
invalid_open = df[(df["Open"] > df["High"]) | (df["Open"] < df["Low"])]
if len(invalid_open) > 0:
results["warnings"].append(f"Open hors range [Low, High] dans {len(invalid_open)} lignes")
invalid_close = df[(df["Close"] > df["High"]) | (df["Close"] < df["Low"])]
if len(invalid_close) > 0:
results["warnings"].append(f"Close hors range [Low, High] dans {len(invalid_close)} lignes")
# Vérifier les valeurs négatives
for col in ["Open", "High", "Low", "Close", "Volume"]:
if (df[col] <= 0).any():
results["is_valid"] = False
results["errors"].append(f"{col}: valeurs négatives ou nulles détectées")
# Vérifier les gaps temporels
if isinstance(df.index, pd.DatetimeIndex):
time_diff = df.index.to_series().diff()
median_diff = time_diff.median()
large_gaps = time_diff[time_diff > median_diff * 5]
if len(large_gaps) > 0:
results["warnings"].append(
f"⚠️ {len(large_gaps)} gaps temporels détectés (>5x intervalle médian)"
)
return results
def clean_data(df: pd.DataFrame) -> pd.DataFrame:
"""
Nettoie les données problématiques
Args:
df: DataFrame à nettoyer
Returns:
DataFrame nettoyé