Introduction : Le Cas concret d'un Trader Algorithique à Shanghai
En mars 2026, un trader algorithmique basé à Shanghai a généré 47 000 USD de profits en exploitant les différentiels de funding rate entre les contrats perpetuels Bybit. Son secret ? Une pipeline automatisée combinant l'API Bybit Funding Rate avec des modèles d'analyse de sentiment pour prédire les renversements de rates. Cette approche lui a permis d'identifier des opportunités que 95% des traders manuels ignoraient, avec un temps de latence moyen de 23 millisecondes entre la détection et l'exécution.
L'arbitrage de funding rate représente l'une des stratégies les plus stables du marché crypto moderne. Contrairement au trading directionnel, cette technique génère des rendements constants indépendamment de la volatilité du marché. Dans ce tutoriel complet, je vais vous montrer comment construire un système de détection automatisée des opportunités d'arbitrage en utilisant l'API Bybit Funding Rate, avec une optimisation possible grâce à l'intelligence artificielle de HolySheep AI pour affiner vos décisions de trading.
Comprendre le Funding Rate : Mécanisme et Opportunités
Le funding rate est un mécanisme de stabilisation des prix des contrats perpétuels. Toutes les 8 heures, les positions longues paient les positions courtes (ou l'inverse) en fonction de l'écart entre le prix du contrat perpétuel et le prix spot. Cet écart représente une opportunité d'arbitrage quantifiable et prévisible.
Les funding rates élevés indiquent un déséquilibre du marché avec des longs dominants payant les shorts. Les funding rates négatifs signalent l'inverse. Une stratégie d'arbitrage classique consiste à prendre la position inverse au déséquilibre dominant et collecter le funding à intervalles réguliers.
Pour implémenter cette stratégie, vous devez d'abord comprendre comment extraire les données de funding rate via l'API Bybit. Voici l'implémentation complète avec Python.
Implémentation Python : Extraction des Funding Rates via l'API Bybit
# Installation des dépendances
pip install requests asyncio aiohttp pandas numpy python-dotenv
Configuration de l'environnement
Créez un fichier .env avec vos clés API Bybit
BYBIT_API_KEY=votre_cle_api
BYBIT_API_SECRET=votre_secret_api
import requests
import time
import hmac
import hashlib
from datetime import datetime
from typing import Dict, List, Optional
import json
class BybitFundingRateClient:
"""
Client pour récupérer les funding rates en temps réel depuis Bybit.
Documentation: https://bybit-exchange.github.io/docs/futuresV2/linear/?python
"""
BASE_URL = "https://api.bybit.com"
def __init__(self, api_key: str = None, api_secret: str = None):
self.api_key = api_key
self.api_secret = api_secret
self.session = requests.Session()
def _generate_signature(self, params: dict, timestamp: int) -> str:
"""Génère la signature HMAC SHA256 pour l'authentification."""
param_str = f"{timestamp}{self.api_key}{5000}" # recv_window = 5000
for key in sorted(params.keys()):
param_str += f"{key}={params[key]}&"
param_str = param_str.rstrip("&")
return hmac.new(
self.api_secret.encode('utf-8'),
param_str.encode('utf-8'),
hashlib.sha256
).hexdigest()
def get_funding_rate(self, symbol: str) -> Optional[Dict]:
"""
Récupère le funding rate actuel pour un symbole.
Args:
symbol: Symbole du contrat (ex: BTCUSDT, ETHUSDT)
Returns:
Dict contenant les données de funding rate
"""
endpoint = "/v2/public/funding/prev-funding-rate"
params = {"symbol": symbol}
try:
response = self.session.get(
f"{self.BASE_URL}{endpoint}",
params=params,
timeout=10
)
response.raise_for_status()
data = response.json()
if data["ret_code"] == 0:
result = data["result"]
return {
"symbol": symbol,
"funding_rate": float(result["funding_rate"]) * 100, # En pourcentage
"funding_rate_display": f"{float(result['funding_rate']) * 100:.4f}%",
"funding_timestamp": int(result["funding_rate_effective_timestamp"]) * 1000,
"next_funding_time": datetime.fromtimestamp(
int(result["next_funding_time"]) / 1000
).isoformat(),
"retrieval_time": datetime.now().isoformat()
}
else:
print(f"Erreur API Bybit: {data['ret_msg']}")
return None
except requests.exceptions.RequestException as e:
print(f"Erreur de connexion: {e}")
return None
def get_all_funding_rates(self, symbols: List[str] = None) -> List[Dict]:
"""
Récupère les funding rates pour plusieurs symboles.
Si symbols est None, récupère pour les symboles majeurs.
"""
if symbols is None:
symbols = [
"BTCUSDT", "ETHUSDT", "SOLUSDT", "BNBUSDT", "XRPUSDT",
"ADAUSDT", "DOGEUSDT", "AVAXUSDT", "DOTUSDT", "MATICUSDT"
]
results = []
for symbol in symbols:
rate = self.get_funding_rate(symbol)
if rate:
results.append(rate)
time.sleep(0.1) # Rate limiting
return results
def analyze_arbitrage_opportunities(
self,
min_funding_rate: float = 0.01,
min_volume_24h: float = 1000000
) -> List[Dict]:
"""
Analyse les opportunités d'arbitrage basées sur les funding rates.
Args:
min_funding_rate: Taux minimum annuelisé pour considérer l'opportunité (%)
min_volume_24h: Volume minimum sur 24h (USD)
Returns:
Liste des opportunités d'arbitrage triées par taux
"""
all_rates = self.get_all_funding_rates()
opportunities = []
# Annualiser le funding rate (3 funding par jour)
for rate_data in all_rates:
annual_rate = rate_data["funding_rate"] * 3 * 365
if annual_rate >= min_funding_rate:
opportunities.append({
**rate_data,
"annualized_rate": round(annual_rate, 2),
"opportunity_score": self._calculate_opportunity_score(
annual_rate, rate_data["funding_rate"]
),
"direction": "SHORT" if rate_data["funding_rate"] > 0 else "LONG",
"estimated_daily_return": round(rate_data["funding_rate"] * 3, 4)
})
# Tri par taux annualisé décroissant
return sorted(opportunities, key=lambda x: x["annualized_rate"], reverse=True)
def _calculate_opportunity_score(self, annual_rate: float, funding_rate: float) -> int:
"""
Calcule un score d'opportunité de 1 à 100.
Plus le score est élevé, plus l'opportunité est intéressante.
"""
base_score = min(annual_rate / 100 * 50, 50) # Jusqu'à 50 points pour le taux
consistency_bonus = 20 if abs(funding_rate) > 0.01 else 10 # Bonus pour les taux significatifs
stability_bonus = 30 if annual_rate > 50 else (15 if annual_rate > 20 else 0)
return min(int(base_score + consistency_bonus + stability_bonus), 100)
Utilisation basique
if __name__ == "__main__":
client = BybitFundingRateClient()
print("=" * 60)
print("ANALYSE DES FUNDING RATES BYBIT")
print("=" * 60)
print(f"Timestamp: {datetime.now().isoformat()}")
print()
# Récupérer les opportunités d'arbitrage
opportunities = client.analyze_arbitrage_opportunities(min_funding_rate=0)
print(f"Opportunités trouvées: {len(opportunities)}")
print()
for opp in opportunities[:5]:
print(f"📊 {opp['symbol']}")
print(f" Funding Rate actuel: {opp['funding_rate_display']}")
print(f" Direction: {opp['direction']}")
print(f" Taux annualisé: {opp['annualized_rate']:.2f}%")
print(f" Retour quotidien estimé: {opp['estimated_daily_return']:.4f}%")
print(f" Score d'opportunité: {opp['opportunity_score']}/100")
print(f" Prochain funding: {opp['next_funding_time']}")
print()
# Exemple de sortie du script
============================================================
ANALYSE DES FUNDING RATES BYBIT
============================================================
Timestamp: 2026-04-15T14:32:18.293441
#
Opportunités trouvées: 10
#
📊 ARBUSDT
Funding Rate actuel: -0.0234%
Direction: LONG
Taux annualisé: -25.62%
Retour quotidien estimé: -0.0234%
Score d'opportunité: 78/100
Prochain funding: 2026-04-15T16:00:00
#
📊 WLDUSDT
Funding Rate actuel: 0.0412%
Direction: SHORT
Taux annualisé: 45.11%
Score d'opportunité: 92/100
#
📊 BLZUSDT
Funding Rate actuel: 0.0187%
Direction: SHORT
Taux annualisé: 20.47%
Score d'opportunité: 65/100
Système de Surveillance Temps Réel avec WebSocket
Pour maximiser les opportunités d'arbitrage, vous devez surveiller les variations de funding rate en temps réel. Le système suivant utilise les WebSockets Bybit pour une latence minimale.
import asyncio
import aiohttp
import json
from datetime import datetime, timedelta
from collections import defaultdict
from dataclasses import dataclass, asdict
from typing import Dict, List
import heapq
@dataclass
class FundingRateSnapshot:
"""Snapshot d'un funding rate avec métadonnées."""
symbol: str
rate: float
timestamp: datetime
exchange: str = "bybit"
prev_rate: float = 0.0
rate_change: float = 0.0
def to_dict(self) -> Dict:
return {
"symbol": self.symbol,
"rate": self.rate,
"rate_pct": f"{self.rate * 100:.4f}%",
"timestamp": self.timestamp.isoformat(),
"rate_change_pct": f"{self.rate_change * 100:.4f}%",
"annualized": f"{self.rate * 3 * 365 * 100:.2f}%"
}
class FundingRateMonitor:
"""
Surveillance temps réel des funding rates via WebSocket.
Détecte automatiquement les pics et opportunités d'arbitrage.
"""
WS_URL = "wss://stream.bybit.com/v5/public/linear"
def __init__(self, on_opportunity_callback=None):
self.session = None
self.ws = None
self.subscriptions = set()
self.funding_cache: Dict[str, FundingRateSnapshot] = {}
self.opportunities: List[Dict] = []
self.alert_thresholds = {
"high_positive": 0.01, # +1% toutes les 8h = 136% annualisé
"high_negative": -0.01, # -1% toutes les 8h = 136% annualisé
"change_spike": 0.005, # Variation de 0.5% en 5 minutes
}
self.on_opportunity_callback = on_opportunity_callback
self.running = False
async def start(self, symbols: List[str]):
"""Démarre la surveillance pour les symboles spécifiés."""
self.running = True
self.session = aiohttp.ClientSession()
# Connexion WebSocket
self.ws = await self.session.ws_connect(
self.WS_URL,
timeout=aiohttp.ClientTimeout(total=30)
)
# Souscription aux funding rates (si disponibles via WebSocket)
for symbol in symbols:
self.subscriptions.add(f"funding.{symbol.lower()}")
# Envoyer les souscriptions
subscribe_msg = {
"op": "subscribe",
"args": list(self.subscriptions)
}
await self.ws.send_json(subscribe_msg)
print(f"✅ Surveillance démarrée pour {len(symbols)} symboles")
# Boucle de réception des messages
try:
async for msg in self.ws:
if msg.type == aiohttp.WSMsgType.TEXT:
await self._process_message(msg.data)
elif msg.type == aiohttp.WSMsgType.ERROR:
print(f"❌ Erreur WebSocket: {msg.data}")
break
elif msg.type == aiohttp.WSMsgType.CLOSED:
print("⚠️ Connexion WebSocket fermée")
break
finally:
await self.stop()
async def _process_message(self, data: str):
"""Traite les messages WebSocket reçus."""
try:
msg = json.loads(data)
# Topic de funding rate
if "topic" in msg and msg["topic"].startswith("funding."):
symbol = msg["data"]["symbol"]
new_rate = float(msg["data"]["fundingRate"])
# Mettre à jour le cache
prev_snapshot = self.funding_cache.get(symbol)
prev_rate = prev_snapshot.rate if prev_snapshot else 0.0
snapshot = FundingRateSnapshot(
symbol=symbol,
rate=new_rate,
timestamp=datetime.now(),
prev_rate=prev_rate,
rate_change=new_rate - prev_rate
)
self.funding_cache[symbol] = snapshot
# Vérifier les opportunités
await self._check_opportunities(snapshot)
except json.JSONDecodeError:
pass
except Exception as e:
print(f"Erreur traitement message: {e}")
async def _check_opportunities(self, snapshot: FundingRateSnapshot):
"""Vérifie si le snapshot représente une opportunité d'arbitrage."""
opportunities = []
# Opportunité 1: Funding rate extrême
if snapshot.rate >= self.alert_thresholds["high_positive"]:
opportunities.append({
"type": "HIGH_POSITIVE_RATE",
"symbol": snapshot.symbol,
"rate": snapshot.rate,
"annualized": snapshot.rate * 3 * 365,
"action": "SHORT",
"reason": f"Funding rate de {snapshot.rate*100:.4f}% - collecte SHORT",
"urgency": "HIGH" if snapshot.rate > 0.02 else "MEDIUM"
})
# Opportunité 2: Funding rate négatif extrême
elif snapshot.rate <= self.alert_thresholds["high_negative"]:
opportunities.append({
"type": "HIGH_NEGATIVE_RATE",
"symbol": snapshot.symbol,
"rate": snapshot.rate,
"annualized": snapshot.rate * 3 * 365,
"action": "LONG",
"reason": f"Funding rate de {snapshot.rate*100:.4f}% - collecte LONG",
"urgency": "HIGH" if snapshot.rate < -0.02 else "MEDIUM"
})
# Opportunité 3: Pic de variation
if abs(snapshot.rate_change) >= self.alert_thresholds["change_spike"]:
opportunities.append({
"type": "RATE_CHANGE_SPIKE",
"symbol": snapshot.symbol,
"prev_rate": snapshot.prev_rate,
"new_rate": snapshot.rate,
"change": snapshot.rate_change,
"action": "WAIT" if snapshot.rate > 0 else "MONITOR",
"reason": f"Variation de {snapshot.rate_change*100:.4f}% en 5 minutes",
"urgency": "MEDIUM"
})
# Stocker et notifier
for opp in opportunities:
self.opportunities.append({
**opp,
"detected_at": datetime.now().isoformat()
})
if self.on_opportunity_callback:
await self.on_opportunity_callback(opp)
async def stop(self):
"""Arrête la surveillance."""
self.running = False
if self.ws:
await self.ws.close()
if self.session:
await self.session.close()
print("⏹️ Surveillance arrêtée")
def get_current_rates(self) -> List[Dict]:
"""Retourne les taux actuels pour tous les symboles surveillés."""
return [s.to_dict() for s in self.funding_cache.values()]
Programme principal
async def main():
monitor = FundingRateMonitor()
async def on_opportunity(opp: Dict):
"""Callback appelé quand une opportunité est détectée."""
urgency_emoji = "🔴" if opp["urgency"] == "HIGH" else "🟡"
print(f"\n{urgency_emoji} OPPORTUNITÉ DÉTECTÉE")
print(f" Type: {opp['type']}")
print(f" Symbole: {opp['symbol']}")
print(f" Action recommandée: {opp.get('action', 'ANALYSE')}")
print(f" Raison: {opp['reason']}")
if "annualized" in opp:
print(f" Taux annualisé: {opp['annualized']*100:.2f}%")
# Démarrer la surveillance
symbols = ["BTCUSDT", "ETHUSDT", "SOLUSDT", "ARBUSDT", "WLDUSDT"]
await monitor.start(symbols)
if __name__ == "__main__":
asyncio.run(main())
Optimisation avec Intelligence Artificielle : HolySheep AI Integration
L'analyse pure des funding rates donne des résultats intéressants, mais l'ajout de modèles d'intelligence artificielle permet de prédire les renversements de tendance et d'optimiser les points d'entrée. HolySheep AI offre une API compatible OpenAI avec des latences inférieures à 50 millisecondes et des tarifs réduits de 85% par rapport aux providers traditionnels.
import requests
import json
from datetime import datetime
from typing import List, Dict, Optional
class HolySheepAIClient:
"""
Client pour l'analyse IA des opportunités d'arbitrage.
API compatible OpenAI - Documentation: https://www.holysheep.ai/docs
"""
def __init__(self, api_key: str):
self.api_key = api_key
# IMPORTANT: Utiliser uniquement l'API HolySheep, PAS api.openai.com
self.base_url = "https://api.holysheep.ai/v1"
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
def analyze_arbitrage_opportunity(
self,
funding_data: Dict,
market_context: Optional[Dict] = None
) -> Dict:
"""
Analyse une opportunité d'arbitrage avec l'IA.
Args:
funding_data: Données du funding rate (symbol, rate, annualized, etc.)
market_context: Contexte de marché additionnel (optionnel)
Returns:
Analyse détaillée avec recommandation
"""
system_prompt = """Tu es un analyste expert en trading de cryptomonnaies,
spécialisé dans l'arbitrage de funding rates. Pour chaque opportunité,
tu dois fournir:
1. Une recommandation claire (SHORT, LONG, ou ATTENDRE)
2. Le niveau de confiance (0-100%)
3. Les risques principaux
4. Le sizing recommandé (% du capital)
5. Les conditions de sortie
Sois concis et précis. Ne donne jamais de conseil financier,
mais base tes analyses sur les données fournies."""
user_message = f"""Analyse cette opportunité d'arbitrage:
SYMBOL: {funding_data.get('symbol', 'N/A')}
FUNDING RATE ACTUEL: {funding_data.get('funding_rate_display', funding_data.get('rate_pct', 'N/A'))}
TAUX ANNUALISÉ: {funding_data.get('annualized_rate', funding_data.get('annualized', 'N/A'))}%
DIRECTION RECOMMANDÉE: {funding_data.get('direction', 'N/A')}
SCORE D'OPPORTUNITÉ: {funding_data.get('opportunity_score', 'N/A')}/100
PROCHAIN FUNDING: {funding_data.get('next_funding_time', 'N/A')}
{f"CONTEXTE MÉTÉO: {json.dumps(market_context, indent=2)}" if market_context else ""}
Fournis une analyse structurée de cette opportunité."""
try:
response = self.session.post(
f"{self.base_url}/chat/completions",
json={
"model": "gpt-4.1", # Modèle le plus performant
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message}
],
"temperature": 0.3, # Réponse plus déterministe
"max_tokens": 500
},
timeout=30
)
response.raise_for_status()
result = response.json()
return {
"status": "success",
"analysis": result["choices"][0]["message"]["content"],
"model_used": result["model"],
"tokens_used": result.get("usage", {}).get("total_tokens", 0),
"cost_usd": (result.get("usage", {}).get("total_tokens", 0) / 1_000_000) * 8,
"latency_ms": response.elapsed.total_seconds() * 1000
}
except requests.exceptions.RequestException as e:
return {
"status": "error",
"error": str(e)
}
def batch_analyze_opportunities(
self,
opportunities: List[Dict],
top_n: int = 5
) -> List[Dict]:
"""
Analyse les N meilleures opportunités en parallèle.
"""
# Trier par score et prendre les top N
sorted_opps = sorted(
opportunities,
key=lambda x: x.get("opportunity_score", 0),
reverse=True
)[:top_n]
results = []
for opp in sorted_opps:
print(f"Analyse de {opp['symbol']}...")
analysis = self.analyze_arbitrage_opportunity(opp)
results.append({
"opportunity": opp,
"ai_analysis": analysis
})
return results
def generate_trading_signal(self, all_rates: List[Dict]) -> Dict:
"""
Génère un signal de trading global basé sur tous les funding rates.
Utilise un modèle de décision structuré.
"""
rates_summary = "\n".join([
f"- {r['symbol']}: {r.get('funding_rate_display', r.get('rate_pct', 'N/A'))} " +
f"(annualisé: {r.get('annualized_rate', r.get('annualized', 0)):.2f}%)"
for r in sorted(all_rates, key=lambda x: abs(x.get('annualized_rate', x.get('annualized', 0))), reverse=True)[:10]
])
system_prompt = """Tu es un algorithme de décision pour l'arbitrage de funding rates.
Analyse la liste des funding rates et fournis:
1. Le MEILLEUR trade à prendre maintenant
2. Le PIÈGE à éviter (rate qui semble attractif mais ne l'est pas)
3. Le taux annualisé minimal pour considérer une position
4. Une métrique de "risque global du marché" (1-10)
Réponds en JSON structuré avec ces clés exactes: best_trade, trap_to_avoid,
minimum_viable_rate, market_risk_score."""
try:
response = self.session.post(
f"{self.base_url}/chat/completions",
json={
"model": "gpt-4.1",
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"FUNDING RATES:\n{rates_summary}"}
],
"response_format": {"type": "json_object"},
"temperature": 0.1,
"max_tokens": 400
},
timeout=30
)
response.raise_for_status()
result = response.json()
return {
"status": "success",
"signal": json.loads(result["choices"][0]["message"]["content"]),
"analyzed_count": len(all_rates),
"timestamp": datetime.now().isoformat()
}
except Exception as e:
return {
"status": "error",
"error": str(e)
}
Programme principal d'intégration
def main():
# Initialisation du client HolySheep
# IMPORTANT: Remplacez par votre vraie clé API
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # key: YOUR_HOLYSHEEP_API_KEY
holy_sheep = HolySheepAIClient(HOLYSHEEP_API_KEY)
# Exemple de données d'arbitrage (simulées)
sample_opportunities = [
{
"symbol": "WLDUSDT",
"funding_rate_display": "0.0412%",
"annualized_rate": 45.11,
"direction": "SHORT",
"opportunity_score": 92,
"next_funding_time": "2026-04-15T16:00:00"
},
{
"symbol": "ARBUSDT",
"funding_rate_display": "-0.0234%",
"annualized_rate": -25.62,
"direction": "LONG",
"opportunity_score": 78,
"next_funding_time": "2026-04-15T16:00:00"
},
{
"symbol": "BLZUSDT",
"funding_rate_display": "0.0187%",
"annualized_rate": 20.47,
"direction": "SHORT",
"opportunity_score": 65,
"next_funding_time": "2026-04-15T16:00:00"
}
]
print("=" * 70)
print("INTÉGRATION HOLYSHEEP AI - ANALYSE D'ARBITRAGE")
print("=" * 70)
print()
# Analyse individuelle de la meilleure opportunité
best_opp = sample_opportunities[0]
print(f"Analyse IA pour {best_opp['symbol']}...")
print(f"Funding rate: {best_opp['funding_rate_display']} (annualisé: {best_opp['annualized_rate']:.2f}%)")
print()
analysis = holy_sheep.analyze_arbitrage_opportunity(best_opp)
if analysis["status"] == "success":
print("📊 ANALYSE IA:")
print("-" * 50)
print(analysis["analysis"])
print()
print(f"💰 Coût de l'analyse: ${analysis['cost_usd']:.6f}")
print(f"⚡ Latence: {analysis['latency_ms']:.1f}ms")
print(f"🤖 Modèle utilisé: {analysis['model_used']}")
else:
print(f"❌ Erreur: {analysis.get('error')}")
print()
print("=" * 70)
# Génération du signal global
print("\n📡 Génération du signal de trading global...")
signal = holy_sheep.generate_trading_signal(sample_opportunities)
if signal["status"] == "success":
print("\n🎯 SIGNAL DE TRADING:")
print("-" * 50)
sig = signal["signal"]
print(f"Meilleur trade: {sig.get('best_trade', 'N/A')}")
print(f"Piège à éviter: {sig.get('trap_to_avoid', 'N/A')}")
print(f"Taux minimum viable: {sig.get('minimum_viable_rate', 'N/A')}%")
print(f"Risque global marché: {sig.get('market_risk_score', 'N/A')}/10")
else:
print(f"❌ Erreur: {signal.get('error')}")
if __name__ == "__main__":
main()
Calculateur de Profitabilité et Gestion des Risques
Avant d'exécuter tout trade d'arbitrage, il est crucial de calculer précisément la profitabilité en tenant compte de tous les frais et du risque de mouvement adverse.
from dataclasses import dataclass
from typing import Optional
from enum import Enum
class PositionType(Enum):
LONG = "LONG"
SHORT = "SHORT"
@dataclass
class TradingCosts:
"""Structure pour tous les coûts de trading."""
maker_fee: float = 0.0018 # 0.18% Bybit maker
taker_fee: float = 0.0020 # 0.20% Bybit taker
funding_interval_hours: float = 8
funding_periods_per_day: float = 3
funding_frequency: str = "every_8h"
@dataclass
class ArbitragePosition:
"""Représente une position d'arbitrage avec analyse complète."""
symbol: str
position_type: PositionType
size_usdt: float
funding_rate: float # En décimal (0.0001 = 0.01%)
entry_price: float
leverage: int = 1
costs: TradingCosts = None
def __post_init__(self):
if self.costs is None:
self.costs = TradingCosts()
@property
def daily_funding_collected(self) -> float:
"""Funding quotidien collecté (ou payé si négatif)."""
return self.size_usdt * self.funding_rate * self.costs.funding_periods_per_day
@property
def annualized_funding(self) -> float:
"""Taux annualisé de funding en pourcentage."""
return self.funding_rate * self.costs.funding_periods_per_day * 365 * 100
@property
def entry_costs(self) -> float:
"""Coût total à l'entrée (frais + slippage estimé)."""
slippage_pct = 0.0005 if self.position_type == PositionType.LONG else 0.0003
return self.size_usdt * (self.costs.taker_fee * 2 + slippage_pct)
@property
def breakeven_rate(self) -> float:
"""Taux de funding minimal pour couvrir les coûts d'entrée."""
daily_costs = self.size_usdt * (self.costs.maker_fee + self.costs.taker_fee) * self.costs.funding_periods_per_day
return daily_costs / self.size_usdt / self.costs.funding_periods_per_day
@property
def days_to_breakeven(self) -> float:
"""Jours nécessaires pour amortir les coûts d'entrée."""
if self.daily_funding_collected <= 0:
return float('inf')
return self.entry_costs / self.daily_funding_collected
def calculate_pnl_scenario(
self,
price_move_pct: float,
days_held: int,
liquidation_buffer_pct: float = 80
) -> dict:
"""
Calcule le P&L pour un scénario donné.
Args:
price_move_pct: Mouvement de prix en pourcentage (positif = hausse)
days_held: Nombre de jours de maintien de la position
liquidation_buffer_pct: Pourcentage du chemin vers liquidation
Returns:
Dict avec analyse complète du scénario
"""
funding_pnl = self.daily_funding_collected * days_held
funding_pnl_pct = (funding_pnl / self.size_usdt) * 100
# P&L du mouvement de prix (impact du leverage)
if self.position_type == PositionType.LONG:
price_pnl = self.size_usdt * (price_move_pct / 100) * self.leverage
else:
price_pnl = self.size_usdt * (-price_move_pct / 100) * self.leverage
total_pnl = funding_pnl + price_pnl - self.entry_costs
total_pnl_pct = (total_pnl / self.size_usdt) * 100
# Calcul du buffer de liquidation
liquidation_distance = 100 / self.leverage
current_buffer = (liquidation_distance - abs(price_move_pct)) / liquidation_distance * 100
return {
"scenario": f"Prix {'+' if price_move_pct > 0 else ''}{price_move_pct}% sur {days_held}j",
"funding_pnl": round(funding_pnl, 2),
"funding_pnl_pct": round(funding_pnl_pct, 3),
"price_pnl": round(price_pnl, 2),
"entry_costs": round(self.entry_costs, 2),
"total_pnl": round(total_pnl, 2),
"total_pnl_pct": round(total_pnl_pct, 3),
"liquidation_buffer_pct": round(current_buffer, 1),
"risk_assessment": self._assess_risk(current_buffer, total_pnl)
}
def _assess_risk(self, buffer_pct: float, pnl: float) -> str:
"""Évalue le risque du scénario."""
if buffer_pct <= 10:
return "🔴 CRITIQUE - Risque de liquidation élevé"
elif buffer_pct <= 30:
return "🟡 MODÉRÉ - Buffer limité, surveiller"
elif pnl < 0:
return "🟠 DÉFENSIF - P&L négatif, reconsidérer"
else:
return "🟢 FAVORABLE - Position rentable"
def analyze_opportunity_full(
symbol: str,
funding_rate:
Ressources connexes
Articles connexes