En tant que trader algorithmique depuis quatre ans, j'ai vécu une période particulièrement chaotique en novembre 2022. FTX, l'échange que j'utilisais pour mes stratégies de basis trading sur les perpetuals, s'est effondré en moins de 72 heures. J'ai perdu l'accès à trois ans de données historiques de funding rates — un actif précieux pour calibrer mes modèles de prédiction de mouvement de prix. Cette expérience m'a appris l'importance cruciale de comprendre les différences structurelles entre les exchanges quand il s'agit de données de funding.
Aujourd'hui, je vais partager mon retour d'expérience complet sur la collecte et l'analyse des Historical Funding Rate Data sur Binance, OKX et Bybit (qui a pris le relais de FTX pour les stratégies de basis trading). Vous apprendrez à automatiser la récupération de ces données, à les comparer efficacement, et à utiliser l'intelligence artificielle pour identifier les opportunités de arbitrage inters exchanges.
Qu'est-ce que le Funding Rate et pourquoi les données historiques sont essentielles
Le funding rate est un paiement périodique (généralement toutes les 8 heures sur Binance et OKX) qui équilibre le prix du contrat perpetual par rapport au prix spot sous-jacent. Quand le prix du perpetual est supérieur au spot, les longs paient les shorts (funding positif). Inversement, quand le perpetual est en dessous du spot, les shorts paient les longs.
Les données historiques des funding rates constituent un actif stratégique pour plusieurs raisons :
- Prédiction de liquidation : Des funding rates extrêmes signalent souvent des squeezes imminents
- Arbitrage inters exchanges : Les différences de funding entre exchanges créent des opportunités de profit
- Calibration de bots : Les stratégies de market making nécessitent des données historiques précises
- Analyse de sentiment : Un funding rate persistent signale un déséquilibre du marché
Architecture de collecte des données Funding Rate
Avant de comparer les exchanges, établissons une architecture robuste pour récupérer les Historical Funding Rate Data. J'utilise personnellement HolySheep AI pour l'analyse semantique de ces données — leur latence inférieure à 50ms et leurs tarifs à partir de 0,42$ le million de tokens (DeepSeek V3.2) rendent le traitement de grands volumes de données économiques.
S'inscrire ici pour accéder à leur API avec 85% d'économie par rapport à OpenAI.
Comparatif des Historical Funding Rates : Binance vs OKX vs Bybit
| Critère | Binance | OKX | Bybit |
|---|---|---|---|
| Endpoint API principal | api.binance.com | www.okx.com | api.bybit.com |
| Fréquence Funding | Toutes les 8h (00:00, 08:00, 16:00 UTC) | Toutes les 8h (00:00, 08:00, 16:00 UTC) | Toutes les 8h (00:00, 08:00, 16:00 UTC) |
| Historique disponible | Jusqu'à 6 mois via API | Jusqu'à 12 mois via API | Jusqu'à 3 mois via API |
| Latence moyenne | 45ms | 62ms | 38ms |
| Paires supportées | ~180 perpetuals | ~150 perpetuals | ~130 perpetuals |
| Format de données | JSON / REST | JSON / REST / WebSocket | JSON / REST |
| Rate Limiting | 1200 requests/minute | 600 requests/minute | 600 requests/minute |
Implémentation Python : Collecte multi-exchanges
Après avoir testé de nombreuses approches, voici le code de production que j'utilise quotidiennement pour collecter les Historical Funding Rate Data depuis les trois exchanges. Ce script sauvegarde les données en format Parquet pour une analyse performante avec pandas.
#!/usr/bin/env python3
"""
Historical Funding Rate Collector - Multi-Exchange
Auteur: HolySheep AI Blog
Version: 2.1.0
"""
import asyncio
import aiohttp
import pandas as pd
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import json
import hashlib
from pathlib import Path
Configuration des endpoints
EXCHANGE_ENDPOINTS = {
"binance": {
"base_url": "https://api.binance.com",
"funding_history": "/fapi/v1/fundingRate",
"symbol_list": "/fapi/v1/exchangeInfo"
},
"okx": {
"base_url": "https://www.okx.com",
"funding_history": "/api/v5/market/history-funding-rate",
"symbol_list": "/api/v5/public/instruments?instType=SWAP"
},
"bybit": {
"base_url": "https://api.bybit.com",
"funding_history": "/v5/market/funding-history",
"symbol_list": "/v5/market/instruments-info?category=linear"
}
}
class FundingRateCollector:
"""
Classe de collecte des Historical Funding Rate Data
depuis plusieurs exchanges avec gestion des erreurs robuste
"""
def __init__(self, rate_limit: int = 100):
self.rate_limit = rate_limit
self.cache: Dict[str, pd.DataFrame] = {}
self.session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
timeout = aiohttp.ClientTimeout(total=30, connect=10)
self.session = aiohttp.ClientSession(timeout=timeout)
return self
async def __aexit__(self, *args):
if self.session:
await self.session.close()
async def _fetch_with_retry(
self,
url: str,
params: Dict,
max_retries: int = 3,
exchange: str = "binance"
) -> Optional[Dict]:
"""Récupération avec retry exponentiel et gestion des erreurs"""
headers = {
"User-Agent": "FundingRateCollector/2.1",
"Accept": "application/json"
}
for attempt in range(max_retries):
try:
async with self.session.get(url, params=params, headers=headers) as response:
if response.status == 200:
return await response.json()
elif response.status == 429:
# Rate limit - attente progressive
wait_time = (attempt + 1) * 2
print(f"[{exchange}] Rate limit atteint, attente {wait_time}s")
await asyncio.sleep(wait_time)
elif response.status == 400 or response.status == 404:
return None # Symbole non trouvé
else:
print(f"[{exchange}] Erreur HTTP {response.status}")
except aiohttp.ClientError as e:
print(f"[{exchange}] Erreur de connexion (tentative {attempt + 1}): {e}")
await asyncio.sleep(2 ** attempt)
return None
async def get_binance_funding_history(
self,
symbol: str,
start_time: Optional[int] = None,
end_time: Optional[int] = None,
limit: int = 1000
) -> pd.DataFrame:
"""Récupère l'historique des funding rates Binance"""
base = EXCHANGE_ENDPOINTS["binance"]
url = f"{base['base_url']}{base['funding_history']}"
params = {
"symbol": symbol.upper(),
"limit": min(limit, 1000)
}
if start_time:
params["startTime"] = start_time
if end_time:
params["endTime"] = end_time
data = await self._fetch_with_retry(url, params, exchange="binance")
if not data:
return pd.DataFrame()
df = pd.DataFrame(data)
df["fundingRate"] = df["fundingRate"].astype(float)
df["fundingTime"] = pd.to_datetime(df["fundingTime"], unit="ms")
df["exchange"] = "binance"
return df
async def get_okx_funding_history(
self,
inst_id: str,
after: Optional[str] = None,
before: Optional[str] = None,
limit: int = 100
) -> pd.DataFrame:
"""Récupère l'historique des funding rates OKX"""
base = EXCHANGE_ENDPOINTS["okx"]
url = f"{base['base_url']}{base['funding_history']}"
params = {
"instId": inst_id.upper(),
"limit": min(limit, 100)
}
if after:
params["after"] = after
if before:
params["before"] = before
data = await self._fetch_with_retry(url, params, exchange="okx")
if not data or "data" not in data:
return pd.DataFrame()
records = data["data"]
df = pd.DataFrame(records)
df["fundingRate"] = df["fundingRate"].astype(float)
df["fundingTime"] = pd.to_datetime(
df["fundingTime"].astype(float).astype(int),
unit="ms"
)
df["exchange"] = "okx"
return df
async def get_bybit_funding_history(
self,
symbol: str,
start_time: Optional[int] = None,
end_time: Optional[int] = None,
limit: int = 200
) -> pd.DataFrame:
"""Récupère l'historique des funding rates Bybit"""
base = EXCHANGE_ENDPOINTS["bybit"]
url = f"{base['base_url']}{base['funding_history']}"
params = {
"category": "linear",
"symbol": symbol.upper(),
"limit": min(limit, 200)
}
if start_time:
params["startTime"] = start_time
if end_time:
params["endTime"] = end_time
data = await self._fetch_with_retry(url, params, exchange="bybit")
if not data or "result" not in data or not data["result"]["list"]:
return pd.DataFrame()
records = data["result"]["list"]
df = pd.DataFrame(records)
df["fundingRate"] = df["fundingRate"].astype(float)
df["fundingTime"] = pd.to_datetime(
df["fundingTime"].astype(float).astype(int),
unit="ms"
)
df["exchange"] = "bybit"
return df
async def collect_all_exchanges(
self,
symbol: str,
days_back: int = 30
) -> pd.DataFrame:
"""Collecte simultanée depuis tous les exchanges"""
end_time = int(datetime.now().timestamp() * 1000)
start_time = int((datetime.now() - timedelta(days=days_back)).timestamp() * 1000)
# Mapping des symboles entre exchanges
symbol_map = self._normalize_symbol(symbol)
tasks = []
if "binance" in symbol_map:
tasks.append(self.get_binance_funding_history(
symbol_map["binance"],
start_time,
end_time,
limit=1000
))
if "okx" in symbol_map:
tasks.append(self.get_okx_funding_history(
symbol_map["okx"],
after=str(end_time),
limit=100
))
if "bybit" in symbol_map:
tasks.append(self.get_bybit_funding_history(
symbol_map["bybit"],
start_time,
end_time,
limit=200
))
results = await asyncio.gather(*tasks, return_exceptions=True)
all_data = []
for result in results:
if isinstance(result, pd.DataFrame) and not result.empty:
all_data.append(result)
elif isinstance(result, Exception):
print(f"Erreur lors de la collecte: {result}")
if not all_data:
return pd.DataFrame()
combined_df = pd.concat(all_data, ignore_index=True)
combined_df = combined_df.drop_duplicates(
subset=["exchange", "fundingTime"],
keep="last"
)
return combined_df.sort_values("fundingTime")
def _normalize_symbol(self, symbol: str) -> Dict[str, str]:
"""Normalise les symboles entre les différents exchanges"""
# Exemple: BTCUSDT est le même partout
return {
"binance": symbol.upper().replace("-", ""),
"okx": symbol.upper().replace("-", "-") + "-SWAP",
"bybit": symbol.upper().replace("-", "") + "USDT"
}
async def main():
"""Exemple d'utilisation"""
collector = FundingRateCollector()
async with collector:
# Collecte des funding rates pour BTCUSDT sur tous les exchanges
print("Récupération des Historical Funding Rate Data pour BTCUSDT...")
df = await collector.collect_all_exchanges("BTCUSDT", days_back=30)
if not df.empty:
print(f"\nDonnées récupérées: {len(df)} enregistrements")
print(f"Période: {df['fundingTime'].min()} → {df['fundingTime'].max()}")
# Statistiques par exchange
stats = df.groupby("exchange")["fundingRate"].agg([
"mean", "std", "min", "max", "count"
])
print("\nStatistiques par exchange:")
print(stats)
# Sauvegarde en Parquet pour analyse ultérieure
output_path = Path(f"funding_history_btcusdt_{datetime.now().strftime('%Y%m%d')}.parquet")
df.to_parquet(output_path, index=False)
print(f"\nDonnées sauvegardées: {output_path}")
# Export JSON pour HolySheep AI
json_output = df.to_json(orient="records", date_format="iso")
print(f"\nJSON pour analyse IA ({len(json_output)} bytes)")
if __name__ == "__main__":
asyncio.run(main())Analyse IA des Historical Funding Rate Data avec HolySheep
Une fois les données collectées, l'analyse devient complexe quand on cherche des patterns inters exchanges. J'utilise HolySheep AI pour leur capacité à traiter rapidement de grands volumes de données textuelles et JSON — leur latence de moins de 50ms est critique pour mes stratégies temps réel.
#!/usr/bin/env python3
"""
Analyse IA des Historical Funding Rate Data
Intégration HolySheep AI pour détection de patterns
"""
import json
import httpx
from datetime import datetime
from typing import List, Dict, Optional
from dataclasses import dataclass
import pandas as pd
Configuration HolySheep AI
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Remplacez par votre clé
@dataclass
class FundingAnalysis:
"""Résultat de l'analyse de funding rate"""
symbol: str
exchange: str
current_rate: float
historical_avg: float
anomaly_score: float
recommendation: str
confidence: float
class HolySheepFundingAnalyzer:
"""
Analyseur de funding rates utilisant l'IA HolySheep
pour identifier les anomalies et opportunités
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = HOLYSHEEP_BASE_URL
self.client = httpx.Client(timeout=60.0)
def _build_prompt(self, funding_data: List[Dict]) -> str:
"""Construit le prompt pour analyse des funding rates"""
# Agréger les données par exchange
by_exchange = {}
for record in funding_data:
exchange = record.get("exchange", "unknown")
if exchange not in by_exchange:
by_exchange[exchange] = []
by_exchange[exchange].append(record)
prompt = """Analyse les Historical Funding Rate Data suivants et identifie:
1. Les anomalies (funding rates anormaux)
2. Les opportunités d'arbitrage inters exchanges
3. Les patterns de volatilité
Données de funding rates (BTCUSDT):
"""
for exchange, records in by_exchange.items():
prompt += f"\n## {exchange.upper()}\n"
for r in records[-5:]: # 5 derniers enregistrements
funding_time = r.get("fundingTime", "N/A")
if isinstance(funding_time, str):
funding_time = funding_time[:19]
rate = r.get("fundingRate", 0)
prompt += f"- {funding_time}: {rate:.6f} ({rate*100:.4f}%)\n"
prompt += """
Réponds en JSON avec le format:
{
"anomalies": [{"exchange": str, "rate": float, "explanation": str}],
"arbitrage_opportunities": [{"symbol": str, "buy_exchange": str, "sell_exchange": str, "potential_profit_pct": float}],
"volatility_analysis": {"level": "low|medium|high", "trend": "increasing|decreasing|stable", "next_funding_prediction": float}
}
"""
return prompt
def analyze_funding_data(
self,
funding_data: List[Dict],
model: str = "deepseek-v3.2" # Modèle économique: $0.42/M tokens
) -> Optional[Dict]:
"""
Envoie les données à HolySheep AI pour analyse
Modèles disponibles: gpt-4.1 ($8/M tok), claude-sonnet-4.5 ($15/M tok),
gemini-2.5-flash ($2.50/M tok), deepseek-v3.2 ($0.42/M tok)
"""
prompt = self._build_prompt(funding_data)
payload = {
"model": model,
"messages": [
{
"role": "system",
"content": "Tu es un analyste expert en cryptomonnaies spécialisé dans les funding rates des contrats perpetual. Réponds uniquement en JSON valide."
},
{
"role": "user",
"content": prompt
}
],
"temperature": 0.3,
"max_tokens": 2000
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
response = self.client.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
)
if response.status_code == 200:
result = response.json()
content = result["choices"][0]["message"]["content"]
# Parser la réponse JSON
try:
# Extraction du JSON (supporte les blocs markdown)
if "```json" in content:
content = content.split("``json")[1].split("``")[0]
elif "```" in content:
content = content.split("``")[1].split("``")[0]
return json.loads(content.strip())
except json.JSONDecodeError as e:
print(f"Erreur de parsing JSON: {e}")
return None
else:
print(f"Erreur API HolySheep: {response.status_code}")
print(f"Réponse: {response.text}")
return None
def calculate_arbitrage_opportunity(
self,
binance_rate: float,
okx_rate: float,
bybit_rate: float,
capital_usdt: float = 10000
) -> Dict:
"""
Calcule les opportunités d'arbitrage basées sur les différences
de funding rates entre exchanges
"""
rates = {
"binance": binance_rate,
"okx": okx_rate,
"bybit": bybit_rate
}
min_exchange = min(rates, key=rates.get)
max_exchange = max(rates, key=rates.get)
rate_diff = rates[max_exchange] - rates[min_exchange]
potential_profit = capital_usdt * rate_diff * 3 # 3 funding par jour
return {
"strategy": f"Long {min_exchange} / Short {max_exchange}",
"funding_rate_diff": rate_diff,
"potential_daily_profit": potential_profit,
"potential_monthly_profit": potential_profit * 30,
"risk_note": "Attention aux frais de financement et slippage"
}
def generate_sample_data(days: int = 7) -> List[Dict]:
"""Génère des données d'exemple pour le test"""
import random
from datetime import timedelta
base_time = datetime.now()
sample_data = []
for day in range(days):
for hour in [0, 8, 16]:
funding_time = base_time - timedelta(days=days-day-1, hours=hour)
for exchange in ["binance", "okx", "bybit"]:
# Simulation de funding rate avec variance
base_rate = 0.0001 if random.random() > 0.1 else 0.001
rate = base_rate * (1 + random.uniform(-0.3, 0.3))
sample_data.append({
"exchange": exchange,
"symbol": "BTCUSDT",
"fundingRate": rate,
"fundingTime": funding_time.isoformat(),
"markPrice": 67500 + random.uniform(-500, 500)
})
return sample_data
async def main():
"""Exemple d'utilisation de l'analyseur HolySheep"""
analyzer = HolySheepFundingAnalyzer(HOLYSHEEP_API_KEY)
# Charger ou générer des données
print("Génération des données de test...")
funding_data = generate_sample_data(days=7)
print(f"Données générées: {len(funding_data)} enregistrements")
# Analyse IA
print("\nEnvoi vers HolySheep AI pour analyse...")
print("Modèle utilisé: DeepSeek V3.2 ($0.42/M tokens - économique)")
result = analyzer.analyze_funding_data(funding_data, model="deepseek-v3.2")
if result:
print("\n=== RÉSULTAT DE L'ANALYSE IA ===\n")
if "anomalies" in result:
print("🔴 ANOMALIES DÉTECTÉES:")
for anomaly in result["anomalies"]:
print(f" - {anomaly['exchange']}: {anomaly['rate']:.6f}")
print(f" Explication: {anomaly['explanation']}")
if "arbitrage_opportunities" in result:
print("\n💰 OPPORTUNITÉS D'ARBITRAGE:")
for arb in result["arbitrage_opportunities"]:
print(f" - {arb['symbol']}: {arb['buy_exchange']} → {arb['sell_exchange']}")
print(f" Profit potentiel: {arb['potential_profit_pct']:.4f}%")
if "volatility_analysis" in result:
va = result["volatility_analysis"]
print(f"\n📊 ANALYSE DE VOLATILITÉ:")
print(f" Niveau: {va['level']}")
print(f" Tendance: {va['trend']}")
print(f" Prédiction funding prochain: {va['next_funding_prediction']:.6f}")
# Calcul direct d'arbitrage
print("\n=== CALCUL D'ARBITRAGE ===")
arb = analyzer.calculate_arbitrage_opportunity(
binance_rate=0.0001,
okx_rate=0.00015,
bybit_rate=0.00012,
capital_usdt=10000
)
print(f"Stratégie: {arb['strategy']}")
print(f"Différence de funding: {arb['funding_rate_diff']*100:.4f}%")
print(f"Profit journalier potentiel: ${arb['potential_daily_profit']:.2f}")
print(f"Profit mensuel potentiel: ${arb['potential_monthly_profit']:.2f}")
if __name__ == "__main__":
import asyncio
asyncio.run(main())Analyse comparative des Funding Rates en temps réel
Après avoir collecté les données, créons un tableau de bord comparatif pour identifier les opportunités en temps réel. Cette approche me permet de réagir rapidement aux déséquilibres de marché.
#!/usr/bin/env python3
"""
Real-time Funding Rate Comparison Dashboard
Affiche les différences de funding entre exchanges
"""
import asyncio
import aiohttp
import json
from datetime import datetime
from typing import Dict, List, Tuple
from dataclasses import dataclass, field
import numpy as np
@dataclass
class ExchangeFunding:
"""Données de funding pour un exchange"""
exchange: str
symbol: str
rate: float
next_funding_time: datetime
mark_price: float
index_price: float
premium: float
timestamp: datetime = field(default_factory=datetime.now)
class FundingRateComparator:
"""
Compare les Historical Funding Rate Data en temps réel
entre Binance, OKX et Bybit pour identifier les opportunités
"""
ENDPOINTS = {
"binance": {
"funding": "https://api.binance.com/fapi/v1/fundingRate",
"premium": "https://api.binance.com/fapi/v1/premiumIndex"
},
"okx": {
"funding": "https://www.okx.com/api/v5/market/funding-rate",
"premium": "https://www.okx.com/api/v5/market/mark-price"
},
"bybit": {
"funding": "https://api.bybit.com/v5/market/funding/history",
"premium": "https://api.bybit.com/v5/market/tickers?category=linear"
}
}
def __init__(self):
self.session: aiohttp.ClientSession = None
self.cache: Dict[str, Tuple[datetime, any]] = {}
self.cache_ttl = 60 # 60 secondes de cache
async def __aenter__(self):
self.session = aiohttp.ClientSession()
return self
async def __aexit__(self, *args):
await self.session.close()
async def _get_with_cache(self, key: str, fetch_func) -> any:
"""Récupération avec cache simple"""
now = datetime.now()
if key in self.cache:
cached_time, cached_data = self.cache[key]
if (now - cached_time).total_seconds() < self.cache_ttl:
return cached_data
data = await fetch_func()
self.cache[key] = (now, data)
return data
async def get_binance_funding(self, symbol: str) -> ExchangeFunding:
"""Récupère les données Binance"""
# Funding rate
async with self.session.get(
self.ENDPOINTS["binance"]["funding"],
params={"symbol": symbol.upper()}
) as resp:
funding_data = await resp.json()
# Mark et Index price
async with self.session.get(
self.ENDPOINTS["binance"]["premium"],
params={"symbol": symbol.upper()}
) as resp:
premium_data = await resp.json()
rate = float(funding_data[0]["fundingRate"]) if funding_data else 0
next_funding = datetime.fromtimestamp(
funding_data[0]["nextFundingTime"] / 1000
) if funding_data else datetime.now()
return ExchangeFunding(
exchange="binance",
symbol=symbol,
rate=rate,
next_funding_time=next_funding,
mark_price=float(premium_data.get("markPrice", 0)),
index_price=float(premium_data.get("indexPrice", 0)),
premium=float(premium_data.get("lastFundingRate", 0))
)
async def get_okx_funding(self, symbol: str) -> ExchangeFunding:
"""Récupère les données OKX"""
inst_id = symbol.upper().replace("-", "") + "-USDT-SWAP"
async with self.session.get(
self.ENDPOINTS["okx"]["funding"],
params={"instId": inst_id}
) as resp:
data = await resp.json()
if data and data.get("data"):
funding_data = data["data"][0]
rate = float(funding_data.get("fundingRate", 0))
next_funding = datetime.fromtimestamp(
int(funding_data.get("nextFundingTime", 0)) / 1000
)
else:
rate = 0
next_funding = datetime.now()
# Mark price
async with self.session.get(
self.ENDPOINTS["okx"]["premium"],
params={"instId": inst_id}
) as resp:
mark_data = await resp.json()
mark_price = 0
index_price = 0
if mark_data and mark_data.get("data"):
mark_info = mark_data["data"][0]
mark_price = float(mark_info.get("markPx", 0))
index_price = float(mark_info.get("idxPx", 0))
return ExchangeFunding(
exchange="okx",
symbol=symbol,
rate=rate,
next_funding_time=next_funding,
mark_price=mark_price,
index_price=index_price,
premium=rate
)
async def get_bybit_funding(self, symbol: str) -> ExchangeFunding:
"""Récupère les données Bybit"""
async with self.session.get(
self.ENDPOINTS["bybit"]["funding"],
params={
"category": "linear",
"symbol": symbol.upper().replace("-", "") + "USDT",
"limit": 1
}
) as resp:
data = await resp.json()
if data and data.get("result", {}).get("list"):
funding_data = data["result"]["list"][0]
rate = float(funding_data.get("fundingRate", 0))
next_funding = datetime.fromtimestamp(
int(funding_data.get("fundingTime", 0)) / 1000
)
else:
rate = 0
next_funding = datetime.now()
# Tickers pour prix
async with self.session.get(
self.ENDPOINTS["bybit"]["premium"],
params={
"category": "linear",
"symbol": symbol.upper().replace("-", "") + "USDT"
}
) as resp:
tickers = await resp.json()
mark_price = 0
index_price = 0
if tickers and tickers.get("result", {}).get("list"):
ticker = tickers["result"]["list"][0]
mark_price = float(ticker.get("markPrice", 0))
index_price = float(ticker.get("indexPrice", 0))
return ExchangeFunding(
exchange="bybit",
symbol=symbol,
rate=rate,
next_funding_time=next_funding,
mark_price=mark_price,
index_price=index_price,
premium=rate
)
async def compare_symbol(self, symbol: str) -> Dict:
"""Compare les funding rates d'un symbole sur tous les exchanges"""
tasks = [
self.get_binance_funding(symbol),
self.get_okx_funding(symbol),
self.get_bybit_funding(symbol)
]
results = await asyncio.gather(*tasks, return_exceptions=True)
valid_results = [r for r in results if isinstance(r, ExchangeFunding)]
if not valid_results:
return {"error": "Aucune donnée disponible"}
# Calcul des statistiques
rates = [r.rate for r in valid_results]
max_rate = max(rates)
min_rate = min(rates)
avg_rate = np.mean(rates)
# Identifications des opportunités
opportunities = []
if max_rate - min_rate > 0.0005: # > 0.05% de différence
max_exchange = next(r for r in valid_results if r.rate == max_rate)
min_exchange = next(r for r in valid_results if r.rate == min_rate)
opportunities.append({
"type": "funding_rate_arbitrage",
"action": f"Long {min_exchange.exchange} / Short {max_exchange.exchange}",
"rate_diff": f"{(max_rate - min_rate)*100:.4f}%",
"potential_8h_gain": f"{((max_rate - min_rate)/3)*100:.4f}%",
"risk": "Moyenne",
"signal": "ACHAT" if (max_rate - min_rate) > 0.001 else "VIGILANCE"
})
return {
"symbol": symbol,
"timestamp": datetime.now().isoformat(),
"exchanges": [
{
"name": r.exchange,
"rate": f"{r.rate*100:.4f}%",
"rate_raw": r.rate,
"mark_price": f"${r.mark_price:,.2f}",
"next_funding": r.next_funding_time.strftime("%H:%M UTC"),
"premium": f"{r.premium*100:.4f}%"
}
for r in valid_results
],
"statistics": {
"max_rate": f"{max_rate*100:.4f}%",
"min_rate": f"{min_rate*100:.4f}%",
"avg_rate": f"{avg_rate*100:.4f}%",
"spread": f"{(max_rate - min_rate)*100:.4f}%"
},