Bienvenue dans ce tutoriel technique approfondi. En tant qu'auteur senior spécialisé dans l'intégration d'API de données de marché, j'ai testé personnellement plus de 12 fournisseurs d'historical data Crypto en 2026. Aujourd'hui, je vous partage mon retour d'expérience complet sur la construction d'un pipeline de backtesting pour les contrats perpétuels OKX avec l'API Tardis.
Pourquoi OKX et Tardis API ?
OKX représente le 3ème exchange spot et le 2ème en volume de contrats perpétuels mondialement avec plus de 2.8 milliards USD/jour en Open Interest sur les perpétuels USDT-M. Tardis Enterprise, quant à lui, offre l'accès le plus complet aux données tick-by-tick avec une latence moyenne de ~180ms pour la diffusion en temps réel et des données historiques remontant à 2019.
Pour un trader algorithmique sérieux, la qualité des données de tick constitue le fondement de toute stratégie robuste. Une simple barre OHLCV à 1 minute peut masquer des slippage critiques que seul le granularité tick permet de capturer.
Prérequis et Architecture du Pipeline
- Python 3.10+ avec pandas, numpy, asyncio
- Compte Tardis API (niveau Start minimum : $50/mois)
- Compte HolySheep AI pour l'analyse IA des résultats (inscrivez-vous ici)
- Librairies : tardis-client, asyncio, pandas, sqlalchemy
Configuration de l'API Tardis
# Installation des dépendances
pip install tardis-client pandas numpy sqlalchemy aiohttp
Structure du projet
project/
├── config.py # Configuration API
├── data_fetcher.py # Récupération des données
├── backtester.py # Moteur de backtesting
├── analyzer.py # Analyse avec HolySheep AI
└── main.py # Orchestrateur
# config.py
import os
from dataclasses import dataclass
@dataclass
class TardisConfig:
api_key: str = os.getenv("TARDIS_API_KEY", "your_tardis_key")
exchange: str = "okx"
channels: list = None
def __post_init__(self):
self.channels = ["trades", "liquidations", "funding_rate"]
@dataclass
class HolySheepConfig:
"""Configuration HolySheep AI — Taux ¥1=$1, latence <50ms"""
api_key: str = os.getenv("HOLYSHEEP_API_KEY", "your_holysheep_key")
base_url: str = "https://api.holysheep.ai/v1" # IMPORTANT: JAMAIS api.openai.com
model: str = "gpt-4.1" # $8/MTok vs OpenAI $60/MTok
max_tokens: int = 2000
@dataclass
class BacktestConfig:
symbol: str = "BTC-USDT-SWAP"
start_date: str = "2026-01-01"
end_date: str = "2026-03-01"
initial_balance: float = 10000.0
commission_rate: float = 0.0004 # 0.04% taker OKX
Récupération des Données Tick-by-Tick
# data_fetcher.py
import asyncio
from tardis_client import TardisClient, TardisReplayException
from datetime import datetime, timedelta
import pandas as pd
from typing import List, Dict
from config import TardisConfig, BacktestConfig
class OKXDataFetcher:
def __init__(self, config: BacktestConfig, tardis_config: TardisConfig):
self.config = config
self.tardis_config = tardis_config
self.client = TardisClient(api_key=self.tardis_config.api_key)
self.trades: List[Dict] = []
async def fetch_trades(self) -> pd.DataFrame:
"""
Récupère les trades OKX perpétuels BTC-USDT-SWAP
Granularité tick-by-tick pour backtesting haute fidélité
"""
print(f"📡 Connexion à Tardis API pour {self.config.symbol}...")
from_date = datetime.fromisoformat(self.config.start_date)
to_date = datetime.fromisoformat(self.config.end_date)
try:
# Activation du replay pour données historiques
messages = self.client.replay(
exchange=self.tardis_config.exchange,
from_date=from_date,
to_date=to_date,
filters=[{
"channel": "trades",
"symbol": self.config.symbol
}]
)
async for message in messages:
if message.name == "trade":
self.trades.append({
"timestamp": message.timestamp,
"price": float(message.trade["price"]),
"size": float(message.trade["amount"]),
"side": message.trade["side"], # "buy" ou "sell"
"fee": 0, # Calculé après
"trade_id": message.trade.get("id", "")
})
except TardisReplayException as e:
print(f"❌ Erreur replay Tardis: {e}")
raise
df = pd.DataFrame(self.trades)
df.set_index("timestamp", inplace=True)
df.sort_index(inplace=True)
print(f"✅ {len(df)} trades récupérés")
print(f" Période: {df.index.min()} → {df.index.max()}")
print(f" Volume total: {df['size'].sum():,.2f} BTC")
return df
async def fetch_funding_rates(self) -> pd.DataFrame:
"""Récupère les taux de funding pour calcul du carry"""
print("📊 Récupération des taux de funding...")
from_date = datetime.fromisoformat(self.config.start_date)
to_date = datetime.fromisoformat(self.config.end_date)
funding_data = []
messages = self.client.replay(
exchange=self.tardis_config.exchange,
from_date=from_date,
to_date=to_date,
filters=[{"channel": "funding_rate", "symbol": self.config.symbol}]
)
async for message in messages:
if message.name == "funding_rate":
funding_data.append({
"timestamp": message.timestamp,
"funding_rate": float(message.funding_rate["funding_rate"]),
"funding_rate_predicted": float(message.funding_rate.get("funding_rate_predicted", 0))
})
df = pd.DataFrame(funding_data)
df.set_index("timestamp", inplace=True)
print(f" {len(df)} intervalles de funding récupérés")
return df
Test unitaire
async def test_fetch():
fetcher = OKXDataFetcher(
BacktestConfig(symbol="BTC-USDT-SWAP", start_date="2026-02-01", end_date="2026-02-02"),
TardisConfig()
)
df = await fetcher.fetch_trades()
print(df.head())
return df
if __name__ == "__main__":
df = asyncio.run(test_fetch())
Moteur de Backtesting Multi-Stratégie
# backtester.py
import pandas as pd
import numpy as np
from dataclasses import dataclass, field
from typing import List, Dict, Callable, Optional
from datetime import datetime, timedelta
@dataclass
class Position:
entry_time: datetime
entry_price: float
size: float
side: str # "long" ou "short"
entry_fee: float
stop_loss: Optional[float] = None
take_profit: Optional[float] = None
@dataclass
class TradeResult:
entry_time: datetime
exit_time: datetime
entry_price: float
exit_price: float
size: float
side: str
pnl: float
pnl_pct: float
fees: float
duration: timedelta
exit_reason: str # "sl", "tp", "signal", "end"
class PerpetualBacktester:
"""
Backtester haute fidélité pour contrats perpétuels OKX
Inclut: slippage réaliste, frais Maker/Taker, funding, liquidation
"""
def __init__(self, initial_balance: float, commission_taker: float = 0.0004,
commission_maker: float = 0.0002, slippage_bps: float = 2.0):
self.initial_balance = initial_balance
self.balance = initial_balance
self.commission_taker = commission_taker
self.commission_maker = commission_maker
self.slippage_bps = slippage_bps
self.position: Optional[Position] = None
self.trades: List[TradeResult] = []
self.equity_curve: List[float] = [initial_balance]
self.timestamps: List[datetime] = []
def apply_slippage(self, price: float, side: str) -> float:
"""Applique un slippage réaliste en basis points"""
multiplier = 1 + (self.slippage_bps / 10000)
return price * multiplier if side == "buy" else price / multiplier
def open_position(self, timestamp: datetime, price: float, size: float,
side: str, stop_loss: float = None, take_profit: float = None):
"""Ouvre une position avec gestion des frais"""
if self.position:
return False
exec_price = self.apply_slippage(price, side)
fee = exec_price * size * self.commission_taker
self.position = Position(
entry_time=timestamp,
entry_price=exec_price,
size=size,
side=side,
entry_fee=fee,
stop_loss=stop_loss,
take_profit=take_profit
)
self.balance -= fee
return True
def close_position(self, timestamp: datetime, price: float, reason: str) -> Optional[TradeResult]:
"""Ferme la position et enregistre le trade"""
if not self.position:
return None
exec_price = self.apply_slippage(price, "sell" if self.position.side == "long" else "buy")
fee = exec_price * self.position.size * self.commission_taker
if self.position.side == "long":
pnl = (exec_price - self.position.entry_price) * self.position.size - self.position.entry_fee - fee
else:
pnl = (self.position.entry_price - exec_price) * self.position.size - self.position.entry_fee - fee
result = TradeResult(
entry_time=self.position.entry_time,
exit_time=timestamp,
entry_price=self.position.entry_price,
exit_price=exec_price,
size=self.position.size,
side=self.position.side,
pnl=pnl,
pnl_pct=pnl / (self.position.entry_price * self.position.size) * 100,
fees=self.position.entry_fee + fee,
duration=timestamp - self.position.entry_time,
exit_reason=reason
)
self.trades.append(result)
self.balance += pnl
self.equity_curve.append(self.balance)
self.timestamps.append(timestamp)
self.position = None
return result
def check_conditions(self, timestamp: datetime, price: float) -> Optional[str]:
"""Vérifie les conditions de sortie (SL/TP)"""
if not self.position:
return None
if self.position.side == "long":
if self.position.stop_loss and price <= self.position.stop_loss:
return "sl"
if self.position.take_profit and price >= self.position.take_profit:
return "tp"
else:
if self.position.stop_loss and price >= self.position.stop_loss:
return "sl"
if self.position.take_profit and price <= self.position.take_profit:
return "tp"
return None
def run(self, df: pd.DataFrame, strategy: Callable) -> Dict:
"""
Exécute le backtest sur un DataFrame de trades
strategy(timestamp, price, position, df) -> {"action": "buy"|"sell"|"hold", "size": 0.1}
"""
for idx, row in df.iterrows():
timestamp = idx
price = row["price"]
# Vérification SL/TP
if self.position:
reason = self.check_conditions(timestamp, price)
if reason:
self.close_position(timestamp, price, reason)
continue
# Exécution de la stratégie
signal = strategy(timestamp, price, self.position, df)
if signal["action"] == "buy" and not self.position:
self.open_position(
timestamp, price,
signal.get("size", self.balance * 0.95 / price),
"long",
stop_loss=signal.get("stop_loss"),
take_profit=signal.get("take_profit")
)
elif signal["action"] == "sell" and self.position:
self.close_position(timestamp, price, "signal")
# Fermeture finale
if self.position and len(df) > 0:
self.close_position(df.index[-1], df.iloc[-1]["price"], "end")
return self.get_statistics()
def get_statistics(self) -> Dict:
"""Calcule les métriques de performance"""
if not self.trades:
return {"error": "Aucun trade exécuté"}
pnls = [t.pnl for t in self.trades]
winning_trades = [p for p in pnls if p > 0]
losing_trades = [p for p in pnls if p <= 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_pnl": sum(pnls),
"total_pnl_pct": (self.balance - self.initial_balance) / self.initial_balance * 100,
"max_drawdown": self._calculate_max_drawdown(),
"sharpe_ratio": self._calculate_sharpe(pnls),
"avg_trade": np.mean(pnls),
"best_trade": max(pnls),
"worst_trade": min(pnls),
"avg_winner": np.mean(winning_trades) if winning_trades else 0,
"avg_loser": np.mean(losing_trades) if losing_trades else 0,
"profit_factor": abs(sum(winning_trades) / sum(losing_trades)) if losing_trades else float('inf')
}
def _calculate_max_drawdown(self) -> float:
"""Calcule le drawdown maximum"""
peak = self.equity_curve[0]
max_dd = 0
for equity in self.equity_curve:
if equity > peak:
peak = equity
dd = (peak - equity) / peak * 100
if dd > max_dd:
max_dd = dd
return max_dd
def _calculate_sharpe(self, returns: List[float]) -> float:
"""Calcule le Sharpe Ratio annualisé (simplifié)"""
if not returns or np.std(returns) == 0:
return 0
return np.mean(returns) / np.std(returns) * np.sqrt(252 * 24) # ~252 jours trading
Stratégie exemple: Momentum simple
def momentum_strategy(timestamp, price, position, df) -> Dict:
"""Stratégie momentum sur 5 minutes"""
lookback = 12 # ~1 heure avec données tick
idx = df.index.get_loc(timestamp) if timestamp in df.index else 0
if idx < lookback:
return {"action": "hold"}
recent = df.iloc[idx-lookback:idx]
momentum = (price - recent["price"].iloc[0]) / recent["price"].iloc[0] * 100
if momentum > 0.5 and not position: # Signal d'achat
return {
"action": "buy",
"size": 0.01, # 1% du capital par trade
"stop_loss": price * 0.98,
"take_profit": price * 1.03
}
elif momentum < -0.3 and position:
return {"action": "sell"}
return {"action": "hold"}
Intégration HolySheep AI pour Analyse Avancée
# analyzer.py
import aiohttp
import json
from typing import Dict, List
from config import HolySheepConfig
from backtester import TradeResult
class HolySheepAnalyzer:
"""
Utilise HolySheep AI pour analyser automatiquement les résultats de backtest
Taux avantageux: ¥1 = $1 (économie 85%+ vs OpenAI)
Latence <50ms grâce à l'infrastructure optimisée
"""
def __init__(self, config: HolySheepConfig):
self.config = config
self.base_url = config.base_url # https://api.holysheep.ai/v1
async def analyze_backtest_results(self, stats: Dict, trades: List[TradeResult]) -> str:
"""Analyse les résultats avec GPT-4.1 via HolySheep"""
# Préparation du prompt
trades_summary = self._prepare_trades_summary(trades[:50]) # Limite 50 trades
prompt = f"""Analyse ce backtest de trading perpértuel OKX et fournis:
1. Diagnostic des points forts et faibles
2. Recommandations d'amélioration de la stratégie
3. Analyse du profil risque/rendement
4. Suggestions de paramètres optimaux
Métriques clés:
{json.dumps(stats, indent=2)}
Derniers trades:
{trades_summary}
Réponds en français avec des recommandations actionnables."""
headers = {
"Authorization": f"Bearer {self.config.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": self.config.model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": self.config.max_tokens,
"temperature": 0.3
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
) as response:
if response.status == 200:
data = await response.json()
return data["choices"][0]["message"]["content"]
else:
error = await response.text()
raise Exception(f"Erreur HolySheep: {response.status} - {error}")
def _prepare_trades_summary(self, trades: List[TradeResult]) -> str:
"""Prépare un résumé des trades pour le prompt"""
lines = ["Timestamp | Entry | Exit | PnL | Duration | Reason"]
for t in trades:
lines.append(
f"{t.entry_time} | {t.entry_price:.2f} | {t.exit_price:.2f} | "
f"{t.pnl:.2f} ({t.pnl_pct:.2f}%) | {t.duration} | {t.exit_reason}"
)
return "\n".join(lines)
async def generate_strategy_report(self, stats: Dict, symbol: str) -> Dict:
"""Génère un rapport structuré de la stratégie"""
prompt = f"""Génère un rapport JSON pour la stratégie {symbol} avec:
- Score de performance (0-100)
- Verdict (Excellent/Bon/Moyen/Faible)
- 3 recommandations prioritaires
- Paramètres suggérés (stop_loss%, take_profit%, position_size%)
Métriques: {json.dumps(stats, indent=2)}
Réponds UNIQUEMENT en JSON valide."""
headers = {
"Authorization": f"Bearer {self.config.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": self.config.model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 1500,
"temperature": 0.2,
"response_format": {"type": "json_object"}
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
) as response:
if response.status == 200:
data = await response.json()
return json.loads(data["choices"][0]["message"]["content"])
else:
return {"error": f"Statut {response.status}"}
Exemple d'utilisation
async def main():
from config import BacktestConfig, TardisConfig
from data_fetcher import OKXDataFetcher
from backtester import PerpetualBacktester, momentum_strategy
# 1. Récupération des données
fetcher = OKXDataFetcher(
BacktestConfig(start_date="2026-01-01", end_date="2026-01-15"),
TardisConfig()
)
df = await fetcher.fetch_trades()
# 2. Exécution du backtest
backtester = PerpetualBacktester(
initial_balance=10000.0,
commission_taker=0.0004,
slippage_bps=2.0
)
stats = backtester.run(df, momentum_strategy)
print("📊 Résultats du Backtest:")
print(json.dumps(stats, indent=2, default=str))
# 3. Analyse avec HolySheep AI
analyzer = HolySheepAnalyzer(HolySheepConfig())
try:
analysis = await analyzer.analyze_backtest_results(stats, backtester.trades)
print("\n🤖 Analyse HolySheep AI:")
print(analysis)
except Exception as e:
print(f"⚠️ Analyse IA indisponible: {e}")
if __name__ == "__main__":
import asyncio
asyncio.run(main())
Orchestrateur Principal
# main.py
import asyncio
import json
from datetime import datetime
from config import BacktestConfig, TardisConfig, HolySheepConfig
from data_fetcher import OKXDataFetcher
from backtester import PerpetualBacktester, momentum_strategy
from analyzer import HolySheepAnalyzer
async def run_pipeline():
"""Pipeline complet: fetch -> backtest -> analyse IA"""
print("=" * 60)
print("🚀 PIPELINE BACKTEST OKX PERPÉTUELS")
print("=" * 60)
# Configuration
backtest_config = BacktestConfig(
symbol="BTC-USDT-SWAP",
start_date="2026-01-01",
end_date="2026-02-01",
initial_balance=10000.0
)
tardis_config = TardisConfig()
holy_config = HolySheepConfig()
# Étape 1: Récupération des données
print("\n📡 ÉTAPE 1: Téléchargement des données...")
fetcher = OKXDataFetcher(backtest_config, tardis_config)
df = await fetcher.fetch_trades()
# Étape 2: Backtesting
print("\n📈 ÉTAPE 2: Backtesting...")
backtester = PerpetualBacktester(
initial_balance=backtest_config.initial_balance,
commission_taker=backtest_config.commission_rate
)
stats = backtester.run(df, momentum_strategy)
print("\n📊 RÉSULTATS BRUTS:")
print(f" Total Trades: {stats['total_trades']}")
print(f" Win Rate: {stats['win_rate']:.1f}%")
print(f" PnL Total: ${stats['total_pnl']:.2f} ({stats['total_pnl_pct']:.2f}%)")
print(f" Max Drawdown: {stats['max_drawdown']:.2f}%")
print(f" Sharpe Ratio: {stats['sharpe_ratio']:.3f}")
print(f" Profit Factor: {stats['profit_factor']:.2f}")
# Étape 3: Analyse IA
print("\n🤖 ÉTAPE 3: Analyse avec HolySheep AI...")
analyzer = HolySheepAnalyzer(holy_config)
try:
analysis = await analyzer.analyze_backtest_results(stats, backtester.trades)
print("\n" + "=" * 60)
print("ANALYSE IA HOLYSHEEP:")
print("=" * 60)
print(analysis)
except Exception as e:
print(f" ⚠️ Analyse IA: {e}")
# Sauvegarde des résultats
output = {
"timestamp": datetime.now().isoformat(),
"config": {
"symbol": backtest_config.symbol,
"period": f"{backtest_config.start_date} → {backtest_config.end_date}",
"initial_balance": backtest_config.initial_balance
},
"stats": stats,
"sample_trades": [
{
"entry": str(t.entry_time),
"exit": str(t.exit_time),
"pnl": t.pnl,
"side": t.side
}
for t in backtester.trades[:10]
]
}
filename = f"backtest_results_{backtest_config.symbol}_{datetime.now().strftime('%Y%m%d_%H%M')}.json"
with open(filename, "w") as f:
json.dump(output, f, indent=2, default=str)
print(f"\n💾 Résultats sauvegardés: {filename}")
return output
if __name__ == "__main__":
results = asyncio.run(run_pipeline())
Tableau Récapitulatif : Comparatif des APIs de Données Crypto
| API | Prix/Mois | Latence Temps Réel | Historique | Granularité | OKX Support |
|---|---|---|---|---|---|
| Tardis Enterprise | $199+ | ~180ms | 2019-present | Tick-by-tick | ✅ Complet |
| CCXT Pro | $30+ | ~200ms | Limité | 1min+ | ✅ Standard |
| CoinAPI | $79+ | ~500ms | Variable | 1s+ | ⚠️ Partiel |
| Exchange WebSocket | Gratuit | ~50ms | Aucun | Tick | ✅ Natif |
| HolySheep AI | $15-50 | <50ms | N/A (analyse) | JSON | ✅ Intégré |
Pour qui / pour qui ce n'est pas fait
| ✅ RECOMMANDÉ POUR | ❌ DÉCONSEILLÉ POUR |
|---|---|
|
|
Erreurs courantes et solutions
❌ Erreur 1: TardisReplayException "Insufficient credits"
# ❌ ERREUR: Le quota mensuel est épuisé
Code d'erreur: 403 - {"error": "Insufficient credits for replay"}
✅ SOLUTION 1: Vérifier et recharger le crédit
Accédez à https://tardis.dev/subscription
Crédit minimum requis: $50 pour 1 mois Start
✅ SOLUTION 2: Optimiser la requête
Réduire la période de données
from config import BacktestConfig
config = BacktestConfig(
start_date="2026-02-01", # Au lieu de 2026-01-01
end_date="2026-02-15" # Au lieu de 2026-02-01
)
✅ SOLUTION 3: Utiliser le cache local
import pickle
CACHE_FILE = "trades_cache.pkl"
def load_cached_trades(symbol: str, start: str, end: str) -> pd.DataFrame:
cache_key = f"{symbol}_{start}_{end}"
try:
with open(CACHE_FILE, "rb") as f:
cache = pickle.load(f)
if cache_key in cache:
print("📦 Données chargées depuis le cache")
return cache[cache_key]
except FileNotFoundError:
pass
return None
def save_to_cache(data: pd.DataFrame, symbol: str, start: str, end: str):
cache_key = f"{symbol}_{start}_{end}"
try:
with open(CACHE_FILE, "rb") as f:
cache = pickle.load(f)
except FileNotFoundError:
cache = {}
cache[cache_key] = data
with open(CACHE_FILE, "wb") as f:
pickle.dump(cache, f)
❌ Erreur 2: HolySheep API Key Invalid
# ❌ ERREUR: Clé API HolySheep invalide
Code d'erreur: 401 - {"error": "Invalid API key"}
✅ SOLUTION 1: Vérifier la clé dans le dashboard
1. Allez sur https://www.holysheep.ai/register
2. Créez un compte et générez une clé API
3. Configurez la variable d'environnement
import os
os.environ["HOLYSHEEP_API_KEY"] = "hs_live_xxxxxxxxxxxx"
✅ SOLUTION 2: Vérifier les crédits gratuits
HolySheep offre des crédits gratuits à l'inscription
Confirmation: GET https://api.holysheep.ai/v1/user/credits
import aiohttp
async def check_credits(api_key: str):
headers = {"Authorization": f"Bearer {api_key}"}
async with aiohttp.ClientSession() as session:
async with session.get(
"https://api.holysheep.ai/v1/user/credits",
headers=headers
) as resp:
if resp.status == 200:
data = await resp.json()
print(f"💰 Crédits restants: {data['credits']}")
return data['credits']
else:
print("⚠️ Erreur:", await resp.text())
return 0
✅ SOLUTION 3: Rate limit handling
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
async def call_holysheep_safe(prompt: str, config: HolySheepConfig):
"""Appel avec retry automatique"""
headers = {
"Authorization": f"Bearer {config.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": config.model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": config.max_tokens
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{config.base_url}/chat/completions",
headers=headers,
json=payload
) as response:
if response.status == 429:
raise Exception("Rate limit - retrying...")
return await response.json()
❌ Erreur 3: DataFrame vide après fetch
# ❌ ERREUR: Aucun trade récupéré (DataFrame vide)
Symptôme: df.empty = True après fetch_trades()
✅ SOLUTION 1: Vérifier le symbol format
OKX utilise des formats spécifiques
VALID_SYMBOLS = [
"BTC-USDT-SWAP", # Contrat perpétuel USDT-M
"ETH-USDT-SWAP",
"SOL-USDT-SWAP",
"BTC-USD-SWAP", # Contrat perpétuel USD-M
]
❌ INCORRECT: "BTCUSDT" ou "BTC/USDT"
✅ CORRECT: "BTC-USDT-SWAP"
✅ SOLUTION 2: Vérifier la période (pas de données futures)
from datetime import datetime, timedelta
now = datetime.now()
past_date = now - timedelta(days=30)
if past_date < datetime.fromisoformat("2019-01-01"):
print("⚠️ Période antérieure à 2019 - données indisponibles")
✅ SOLUTION 3: Filtrer les channels disponibles
async def list_available_channels(client, exchange: str, symbol: str):
"""Liste les channels disponibles pour debug"""
print(f"📡 Channels disponibles pour {exchange}/{symbol}:")
for channel in ["trades", "trades_verbose", "liquidations", "funding_rate"]:
try:
count = 0
messages = client.replay(
exchange=exchange,
from_date=datetime.now() - timedelta(h