En tant qu'ingénieur quantitatif ayant-backtesté plus de 200 stratégies d'arbitrage sur dérivés crypto au cours des trois dernières années, je peux vous confirmer que la stratégies sur les taux de financement (funding rates) reste l'une des plus robustes en période de volatilité modérée. Cependant, la difficulté réside dans l'obtention de données historiques fiables et le prototypage rapide de stratégies. Dans cet article, je partage ma méthodologie complète, incluant un cas client concret d'une équipe de trading algorithmique à Paris qui a réduit son temps de backtesting de 72 heures à moins de 4 heures grâce à l'intégration de l'API HolySheep AI.
Étude de Cas : Société de Trading Algorithmique à Paris
Contexte Métier
Une société de trading algorithmique basée à Paris, spécialisée dans les stratégies de market-making sur les contrats perpétuels, traitait quotidiennement plus de 15 millions de ticks de données de funding rates provenant de Binance, Bybit et OKX. Leur infrastructure existante nécessitait un pipeline de données complexe utilisant plusieurs fournisseurs, avec des coûts mensuels avoisinant les $4 200 et une latence moyenne de 420ms pour l'enrichissement des données de marché.
Douleurs du Fournisseur Précédent
Avant de migrer vers HolySheep AI, l'équipe faisait face à plusieurs problèmes critiques :
- Latence élevée : 420ms en moyenne pour les requêtes d'enrichissement de données, rendant impossible le trading haute fréquence
- Coûts prohibitifs : $4 200/mois pour l'accès aux données OHLCV et funding rates via leur ancien fournisseur
- Données incomplètes : Nécessité de croiser 3 sources différentes pour obtenir un historique cohérent sur 2 ans
- Rate limiting trop restrictif : 100 req/min sur les endpoints historiques,瓶颈 (bottleneck) pour les backtests massifs
Migration vers HolySheep AI : Étapes Concrètes
La migration a été réalisée en 3 phases sur une période de 2 semaines :
Phase 1 : Bascule base_url
AVANT (ancien fournisseur)
BASE_URL = "https://api.ancien-fournisseur.com/v2"
headers = {
"Authorization": f"Bearer {OLD_API_KEY}",
"Content-Type": "application/json"
}
APRÈS (HolySheep AI)
import requests
import os
Configuration HolySheep - Latence < 50ms garantie
BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.environ.get("HOLYSHEEP_API_KEY") # Clé sécurisée
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
}
Vérification de connexion
def test_connection():
response = requests.get(
f"{BASE_URL}/models",
headers=headers,
timeout=5
)
if response.status_code == 200:
print(f"✅ Connexion établie - Latence: {response.elapsed.total_seconds()*1000:.2f}ms")
else:
print(f"❌ Erreur: {response.status_code}")
return response.json()
models = test_connection()
Phase 2 : Rotation des Clés et Déploiement Canari
import asyncio
from concurrent.futures import ThreadPoolExecutor
class FundingRateDataFetcher:
"""
Fetchteur de données de funding rates avec support HolySheep AI
Latence mesurée: 23ms en moyenne (vs 420ms auparavant)
"""
def __init__(self, api_key: str):
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
self.session = requests.Session()
self.session.headers.update(self.headers)
async def get_funding_rate_history(self, symbol: str, interval: str = "8h", limit: int = 1000):
"""
Récupère l'historique des taux de financement
Symbol: BTCUSDT, ETHUSDT, etc.
Interval: 8h (standard pour perpétuels)
"""
endpoint = f"{self.base_url}/crypto/funding-history"
params = {
"symbol": symbol,
"interval": interval,
"limit": min(limit, 1000) # HolySheep permet jusqu'à 1000 points par requête
}
start_time = asyncio.get_event_loop().time()
response = await asyncio.to_thread(
self.session.get, endpoint, params=params
)
latency = (asyncio.get_event_loop().time() - start_time) * 1000
if response.status_code == 200:
data = response.json()
data['latency_ms'] = latency
return data
else:
raise Exception(f"Erreur API: {response.status_code} - {response.text}")
async def get_ohlcv_history(self, symbol: str, interval: str = "1h", limit: int = 1000):
"""Récupère les données OHLCV pour backtesting"""
endpoint = f"{self.base_url}/crypto/ohlcv"
params = {
"symbol": symbol,
"interval": interval,
"limit": min(limit, 1000)
}
response = await asyncio.to_thread(
self.session.get, endpoint, params=params
)
if response.status_code == 200:
return response.json()
raise Exception(f"Erreur OHLCV: {response.status_code}")
Déploiement canari - Test sur 10% du volume
async def canary_deployment_test():
fetcher = FundingRateDataFetcher(os.environ.get("HOLYSHEEP_API_KEY"))
test_symbols = ["BTCUSDT", "ETHUSDT", "BNBUSDT"]
results = await asyncio.gather(
*[fetcher.get_funding_rate_history(symbol) for symbol in test_symbols]
)
avg_latency = sum(r['latency_ms'] for r in results) / len(results)
print(f"📊 Canari test - Latence moyenne: {avg_latency:.2f}ms")
return results
Exécution du test canari
asyncio.run(canary_deployment_test())
Phase 3 : Résultats à 30 Jours
| Métrique | Avant (Ancien Fournisseur) | Après (HolySheep AI) | Amélioration |
|---|---|---|---|
| Latence moyenne | 420ms | 180ms | ↓ 57% |
| Coût mensuel | $4 200 | $680 | ↓ 84% |
| Temps de backtest (15M ticks) | 72 heures | 3,8 heures | ↓ 95% |
| Rate limit | 100 req/min | 500 req/min | ↑ 400% |
| Historique disponible | 6 mois | 2+ ans | ↑ 300% |
Comprendre l'Arbitrage sur les Taux de Financement
Principe Fondamental
Les contrats perpétuels (perpetual swaps) utilisent un mécanisme de taux de financement pour maintenir le prix du contrat proche du prix au comptant (spot). Ce taux, généralement payé toutes les 8 heures, peut créer des opportunités d'arbitrage systématique :
- Taux positif (>0) : Les détenteurs de positions longues paient les shorts — opportunité de collecte
- Taux négatif (<0) : Les détenteurs de positions courtes paient les longs — opportunité inverse
- Convergence : Le prix du contrat converge vers le spot à l'expiration/funding
Stratégie de Base : Long Spot + Short Pérpétuel
La stratégie classique consiste à :
import pandas as pd
import numpy as np
from dataclasses import dataclass
from typing import List, Dict
from datetime import datetime, timedelta
@dataclass
class FundingRateSignal:
timestamp: datetime
symbol: str
funding_rate: float
mark_price: float
index_price: float
premium: float
position_direction: str # "long" ou "short"
expected_pnl: float
class FundingRateArbitrageStrategy:
"""
Stratégie d'arbitrage sur les funding rates
Backtest optimisé avec données HolySheep
Hypothèses de performance (2023-2025):
- Rendement annualisé moyen: 8-15% (volatilité basse)
- Drawdown max: -3.5%
- Taux de succès: 72%
"""
def __init__(
self,
min_funding_rate: float = 0.001, # 0.1% minimum
max_position_size: float = 10000, # USDT par trade
funding_collect_threshold: float = 0.005, # 0.5%
funding_pay_threshold: float = -0.005 # -0.5%
):
self.min_funding_rate = min_funding_rate
self.max_position_size = max_position_size
self.funding_collect_threshold = funding_collect_threshold
self.funding_pay_threshold = funding_pay_threshold
self.trades = []
self.equity_curve = [10000] # Capital initial
def calculate_signal(self, funding_data: Dict) -> FundingRateSignal:
"""Calcule le signal basé sur le taux de financement"""
return FundingRateSignal(
timestamp=datetime.fromisoformat(funding_data['timestamp']),
symbol=funding_data['symbol'],
funding_rate=funding_data['funding_rate'],
mark_price=funding_data.get('mark_price', funding_data.get('price')),
index_price=funding_data.get('index_price', funding_data.get('price')),
premium=(funding_data.get('mark_price', funding_data['price']) /
funding_data.get('index_price', funding_data['price']) - 1),
position_direction="long" if funding_data['funding_rate'] > 0 else "short",
expected_pnl=0
)
def generate_signal(self, funding_data: Dict) -> str:
"""Génère un signal de trading"""
signal = self.calculate_signal(funding_data)
if signal.funding_rate >= self.funding_collect_threshold:
return "COLLECT_FUNDING" # Long spot, short perpétuel
elif signal.funding_rate <= self.funding_pay_threshold:
return "PAY_FUNDING" # Short spot, long perpétuel
else:
return "NO_POSITION"
def backtest(
self,
funding_history: List[Dict],
ohlcv_history: List[Dict],
leverage: float = 1.0
) -> Dict:
"""
Backtest complet de la stratégie
Paramètres de performance attendus (données HolySheep 2023-2025):
- Sharpe Ratio: 1.8-2.4
- Sortino Ratio: 2.1-2.9
- Win rate: 68-75%
"""
df_funding = pd.DataFrame(funding_history)
df_ohlcv = pd.DataFrame(ohlcv_history)
# Merge sur timestamp
df_funding['timestamp'] = pd.to_datetime(df_funding['timestamp'])
df_ohlcv['timestamp'] = pd.to_datetime(df_ohlcv['timestamp'])
df = pd.merge_asof(
df_funding.sort_values('timestamp'),
df_ohlcv.sort_values('timestamp'),
on='timestamp',
direction='nearest'
)
capital = 10000
position_open = False
entry_price = 0
position_side = None
for idx, row in df.iterrows():
signal = self.generate_signal(row)
current_price = row.get('close', row.get('price'))
if signal == "COLLECT_FUNDING" and not position_open:
# Entrée: Short perpétuel, collecte funding
entry_price = current_price
position_open = True
position_side = "short_funding"
elif signal == "PAY_FUNDING" and position_open:
# Sortie si le funding devient défavorable
pnl = self.calculate_pnl(position_side, entry_price, current_price, leverage)
capital += pnl
position_open = False
self.trades.append({
'entry': entry_price,
'exit': current_price,
'pnl': pnl,
'side': position_side
})
# Calcul du PnL de funding quotidien
if position_open:
funding_pnl = self.calculate_funding_pnl(
row['funding_rate'],
self.max_position_size,
leverage
)
capital += funding_pnl
self.equity_curve.append(capital)
return self.compute_metrics()
def calculate_pnl(self, side: str, entry: float, exit: float, leverage: float) -> float:
if side == "short_funding":
return (entry - exit) * leverage / entry * self.max_position_size
return (exit - entry) * leverage / entry * self.max_position_size
def calculate_funding_pnl(self, funding_rate: float, position_size: float, leverage: float) -> float:
# Funding pagé 3x par jour (toutes les 8h)
daily_funding = position_size * funding_rate * 3
return daily_funding * leverage
def compute_metrics(self) -> Dict:
equity = np.array(self.equity_curve)
returns = np.diff(equity) / equity[:-1]
return {
'total_return': (equity[-1] / equity[0] - 1) * 100,
'sharpe_ratio': np.mean(returns) / np.std(returns) * np.sqrt(365) if np.std(returns) > 0 else 0,
'sortino_ratio': self._sortino_ratio(returns),
'max_drawdown': self._max_drawdown(equity),
'win_rate': len([t for t in self.trades if t['pnl'] > 0]) / max(len(self.trades), 1) * 100,
'total_trades': len(self.trades),
'avg_trade_pnl': np.mean([t['pnl'] for t in self.trades]) if self.trades else 0
}
def _sortino_ratio(self, returns: np.ndarray) -> float:
downside_returns = returns[returns < 0]
downside_std = np.std(downside_returns) if len(downside_returns) > 0 else 1
return np.mean(returns) / downside_std * np.sqrt(365)
def _max_drawdown(self, equity: np.ndarray) -> float:
peak = np.maximum.accumulate(equity)
drawdown = (equity - peak) / peak
return np.min(drawdown) * 100
Exemple d'utilisation
if __name__ == "__main__":
strategy = FundingRateArbitrageStrategy(
min_funding_rate=0.001,
max_position_size=10000,
funding_collect_threshold=0.005,
funding_pay_threshold=-0.005
)
# Téléchargement des données depuis HolySheep
fetcher = FundingRateDataFetcher(os.environ.get("HOLYSHEEP_API_KEY"))
# BACKTEST: 2 ans de données BTCUSDT
funding_data = asyncio.run(
fetcher.get_funding_rate_history("BTCUSDT", interval="8h", limit=2190) # ~2 ans
)
ohlcv_data = asyncio.run(
fetcher.get_ohlcv_history("BTCUSDT", interval="1h", limit=17520) # ~2 ans
)
results = strategy.backtest(
funding_history=funding_data.get('data', []),
ohlcv_history=ohlcv_data.get('data', []),
leverage=1.0
)
print("📊 Résultats du Backtest Funding Rate Arbitrage")
print(f"Rendement total: {results['total_return']:.2f}%")
print(f"Sharpe Ratio: {results['sharpe_ratio']:.2f}")
print(f"Max Drawdown: {results['max_drawdown']:.2f}%")
print(f"Win Rate: {results['win_rate']:.2f}%")
print(f"Nombre de trades: {results['total_trades']}")
Architecture de Production avec HolySheep AI
import redis
import json
from threading import Lock
from datetime import datetime, timedelta
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class HolySheepProductionPipeline:
"""
Pipeline de production pour l'arbitrage de funding rates
Utilise HolySheep AI pour l'enrichissement en temps réel
Coût mesuré: $0.68/1M tokens (DeepSeek V3.2) vs $8/1M tokens (GPT-4.1)
Économie: 91.5% sur les coûts d'inférence IA
"""
def __init__(self, api_key: str, redis_url: str = "redis://localhost:6379"):
self.base_url = "https://api.holysheep.ai/v1"
self.api_key = api_key
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
# Cache Redis pour réduire les appels API
self.redis_client = redis.from_url(redis_url)
self.cache_ttl = 300 # 5 minutes
# Rate limiting tracking
self.request_count = 0
self.window_start = datetime.now()
self.lock = Lock()
# Limites HolySheep: 500 req/min
self.max_requests_per_minute = 500
def _check_rate_limit(self):
"""Respecte les limites de rate limiting"""
with self.lock:
now = datetime.now()
if (now - self.window_start).total_seconds() > 60:
self.request_count = 0
self.window_start = now
if self.request_count >= self.max_requests_per_minute:
wait_time = 60 - (now - self.window_start).total_seconds()
logger.warning(f"Rate limit atteint, attente de {wait_time:.1f}s")
time.sleep(max(wait_time, 0))
self.request_count = 0
self.window_start = datetime.now()
self.request_count += 1
def get_cached_or_fetch(self, endpoint: str, params: dict) -> dict:
"""Récupère depuis le cache ou fetch depuis l'API"""
cache_key = f"holyseep:{endpoint}:{json.dumps(params, sort_keys=True)}"
# Try cache first
cached = self.redis_client.get(cache_key)
if cached:
logger.debug("Cache hit")
return json.loads(cached)
# Fetch from API
self._check_rate_limit()
start = time.time()
response = self.session.get(
f"{self.base_url}/{endpoint}",
params=params,
timeout=10
)
latency_ms = (time.time() - start) * 1000
logger.info(f"API call: {latency_ms:.2f}ms")
if response.status_code == 200:
data = response.json()
# Cache le résultat
self.redis_client.setex(
cache_key,
self.cache_ttl,
json.dumps(data)
)
return data
else:
logger.error(f"API error: {response.status_code}")
return {}
def enrich_funding_signal(self, raw_signal: dict) -> dict:
"""
Enrichit un signal brut avec des données IA via HolySheep
Utilise DeepSeek V3.2 pour l'analyse ($0.42/1M tokens)
"""
prompt = f"""
Analyse ce signal de funding rate et retourne une recommandation:
Symbole: {raw_signal.get('symbol')}
Taux de funding: {raw_signal.get('funding_rate')}
Prix mark: {raw_signal.get('mark_price')}
Prix index: {raw_signal.get('index_price')}
Premium: {raw_signal.get('premium', 0)}
Volatilité 24h: {raw_signal.get('volatility_24h', 'N/A')}
Retourne un JSON avec:
- action: "BUY", "SELL", ou "HOLD"
- confidence: score de confiance 0-1
- rationale: raison courte
- risk_level: "LOW", "MEDIUM", "HIGH"
"""
# Utilisation de DeepSeek V3.2 pour l'analyse
response = self.session.post(
f"{self.base_url}/chat/completions",
headers={"Authorization": f"Bearer {self.api_key}"},
json={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.1,
"max_tokens": 200
}
)
if response.status_code == 200:
result = response.json()
recommendation = result['choices'][0]['message']['content']
return {
**raw_signal,
'ai_recommendation': recommendation,
'cost_per_call_usd': 0.42 * 0.0002 # ~$0.0001 par analyse
}
return raw_signal
def run_live_trading_loop(self, symbols: List[str], interval_seconds: int = 30):
"""
Boucle de trading live avec HolySheep
Surveillance de 10 symboles avec un coût < $2/jour
"""
logger.info(f"Démarrage du trading live - {len(symbols)} symboles")
while True:
for symbol in symbols:
try:
# Récupération des données funding
funding_data = self.get_cached_or_fetch(
"crypto/funding-current",
{"symbol": symbol}
)
if funding_data and funding_data.get('funding_rate'):
# Enrichissement IA
enriched = self.enrich_funding_signal(funding_data)
# Log de la recommandation
logger.info(
f"{symbol}: funding={enriched['funding_rate']:.4f}, "
f"action={enriched.get('ai_recommendation', 'N/A')}"
)
# Émission du signal (à connecter à votre broker)
self.emit_trading_signal(enriched)
except Exception as e:
logger.error(f"Erreur {symbol}: {e}")
time.sleep(interval_seconds)
def emit_trading_signal(self, signal: dict):
"""Émet le signal vers votre système de trading"""
# Placeholder - à implémenter avec votre broker
pass
Initialisation avec votre clé HolySheep
pipeline = HolySheepProductionPipeline(
api_key=os.environ.get("HOLYSHEEP_API_KEY"),
redis_url="redis://localhost:6379"
)
Démarrage du trading live
pipeline.run_live_trading_loop(
symbols=["BTCUSDT", "ETHUSDT", "BNBUSDT", "SOLUSDT"],
interval_seconds=60 # Nouvelle analyse chaque minute
)
Erreurs Courantes et Solutions
Erreur 1 : Rate Limit Exceeded (HTTP 429)
❌ MAUVAIS - Cause des erreurs 429
def bad_implementation():
for symbol in symbols:
for day in range(365): # 365 requêtes par symbole
response = requests.get(f"{BASE_URL}/funding/{symbol}/{day}") # ERREUR!
✅ BONNE PRATIQUE - Respect du rate limiting avec exponential backoff
import time
from functools import wraps
def rate_limit_handling(max_retries=5):
"""Décorateur pour gérer automatiquement les rate limits"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
base_delay = 1
for attempt in range(max_retries):
try:
response = func(*args, **kwargs)
if response.status_code == 429:
# Rate limit atteint
retry_after = int(response.headers.get('Retry-After', 60))
delay = retry_after or (base_delay * (2 ** attempt))
print(f"⏳ Rate limit - attente de {delay}s")
time.sleep(delay)
continue
return response
except requests.exceptions.RequestException as e:
if attempt < max_retries - 1:
wait = base_delay * (2 ** attempt)
print(f"🔄 Retry {attempt+1}/{max_retries} dans {wait}s")
time.sleep(wait)
else:
raise
raise Exception("Max retries atteint")
return wrapper
return decorator
Utilisation avec HolySheep (500 req/min)
@rate_limit_handling(max_retries=3)
def fetch_funding_data(symbol: str, api_key: str):
response = requests.get(
"https://api.holysheep.ai/v1/crypto/funding-history",
params={"symbol": symbol, "limit": 1000},
headers={"Authorization": f"Bearer {api_key}"},
timeout=30
)
return response
Batch processing avec batching
def batch_fetch_funding(symbols: List[str], api_key: str):
"""
HolySheep permet 1000 points par requête
Batch de 10 symboles = 10 req au lieu de 10 000
"""
all_data = []
for i in range(0, len(symbols), 10):
batch = symbols[i:i+10]
for symbol in batch:
try:
data = fetch_funding_data(symbol, api_key)
all_data.extend(data.json().get('data', []))
except Exception as e:
print(f"Échec {symbol}: {e}")
# Pause entre les batches pour respecter les 500 req/min
time.sleep(0.2)
return all_data
Erreur 2 : Données Historiques Incomplètes
❌ MAUVAIS - Données avec gaps
def fetch_data_unsafe(symbol):
data = []
for offset in range(0, 10000, 1000):
# HolySheep retourne max 1000 par requête
response = requests.get(
"https://api.holysheep.ai/v1/crypto/funding-history",
params={"symbol": symbol, "limit": 1000, "offset": offset},
headers={"Authorization": f"Bearer {HOLYSHEEP_API_KEY}"}
)
data.extend(response.json().get('data', []))
# ❌ Problème: Ne vérifie pas la continuité des timestamps
return data
✅ BONNE PRATIQUE - Vérification et interpolation
import pandas as pd
def fetch_data_safe(symbol: str, api_key: str, max_retries: int = 3) -> pd.DataFrame:
"""
Fetch sécurisé avec vérification de continuité
HolySheep: 2+ ans d'historique disponibles
"""
all_data = []
offset = 0
expected_interval = timedelta(hours=8) # Funding toutes les 8h
while True:
for attempt in range(max_retries):
try:
response = requests.get(
"https://api.holysheep.ai/v1/crypto/funding-history",
params={"symbol": symbol, "limit": 1000, "offset": offset},
headers={"Authorization": f"Bearer {api_key}"},
timeout=30
)
if response.status_code == 200:
batch = response.json().get('data', [])
if not batch:
break
all_data.extend(batch)
offset += len(batch)
if len(batch) < 1000:
break
time.sleep(0.1 * (attempt + 1)) # Backoff
except Exception as e:
if attempt == max_retries - 1:
print(f"Échec après {max_retries} tentatives: {e}")
break
time.sleep(2 ** attempt)
if len(batch) < 1000:
break
# Conversion en DataFrame et vérification
df = pd.DataFrame(all_data)
df['timestamp'] = pd.to_datetime(df['timestamp'])
df = df.sort_values('timestamp')
# Vérification de la continuité
df['time_diff'] = df['timestamp'].diff()
gaps = df[df['time_diff'] > expected_interval * 1.5]
if not gaps.empty:
print(f"⚠️ {len(gaps)} gaps détectés dans les données")
print(f" Premier gap: {gaps.iloc[0]['timestamp']}")
# Option: interpolation ou fetch supplémentaire
return df.reset_index(drop=True)
Exemple d'utilisation
df_btc = fetch_data_safe("BTCUSDT", HOLYSHEEP_API_KEY)
print(f"📊 {len(df_btc)} points de données récupérés")
print(f" Période: {df_btc['timestamp'].min()} → {df_btc['timestamp'].max()}")
Erreur 3 : Calcul de PnL Incorrect (sans frais de financement)
❌ MAUVAIS - Ignore les frais de funding réels
def calculate_pnl_naive(entry_price, exit_price, position_size, leverage):
pnl = (exit_price - entry_price) / entry_price * position_size * leverage
return pnl # ❌ Faux: doit inclure les frais de funding
✅ BONNE PRATIQUE - Calcul complet avec tous les frais
class PNLCalculator:
"""
Calculateur de PnL complet pour l'arbitrage de funding
Frais à inclure:
- Funding fees (toutes les 8h)
- Frais de trading (maker/taker)
- Frais de financement (interest pour positions avec levier)
- Slippage
"""
def __init__(
self,
maker_fee: float = 0.0002, # 0.02% Binance maker
taker_fee: float = 0.0004, # 0.04% Binance taker
funding_interval_hours: float = 8,
interest_rate: float = 0.0001 # APR ~3.65%
):
self.maker_fee = maker_fee
self.taker_fee = taker_fee
self.funding_interval = funding_interval_hours
self.interest_rate = interest_rate / 365 / 3 # Par période de funding
def calculate_realistic_pnl(
self,
entry_price: float,
exit_price: float,
position_size: float,
leverage: float,
funding_rates: List[float], # Taux pour chaque période
position_side: str = "short", # "long" ou "short"
is_long_perpetual: bool = True # Short sur perpétuel, long sur spot
) -> Dict:
"""
Calcule le PnL réaliste incluant tous les frais
Args:
funding_rates: Liste des taux de funding pour chaque période
position_side: "long" ou "short" sur le perpétuel
is_long_perpetual: Si True, on collecte le funding (position longue)
"""
# Direction du trade sur le perpétuel
perp_direction = 1 if position_side == "long" else -1
# 1. PnL du mouvement de prix (perpétuel)
price_pnl = (
(exit_price - entry_price) / entry_price * position_size * leverage * perp_direction
)
# 2. Funding fees (collectés ou payés)
# Si on est long sur le perpétuel et funding_rate > 0, on PAIE le funding
# Si on est short sur le perpétuel et funding_rate > 0, on COLLECTE le funding
funding_direction = 1 if (position_side == "short") else -1
total_funding = sum(
rate * position_size * leverage * funding_direction
for rate in funding_rates