Introduction : Pourquoi Combiner Tardis et HolySheep pour la Recherche Quantitative
Dans l'écosystème crypto de 2026, la précision des données de funding rate et des ticks dérivés représente un avantage compétitif considérable. Tardis offre un accès brut aux carnets d'ordres et aux données de funding des exchanges majeurs (Binance, Bybit, OKX, Deribit), tandis que HolySheep AI constitue la passerelle unifiée permettant de traiter ces flux massifs avec une latence inférieure à 50ms et un taux de change ¥1=$1 générant des économies de 85% sur les coûts d'inférence IA.
Ce guide s'adresse aux ingénieurs quantitatifs souhaitant construire des pipelines de données robustes, capable de ingérer des millions de ticks par seconde tout en exécutant des modèles de prédiction en temps réel via des APIs IA optimisées.
Architecture du Pipeline de Données
Vue d'ensemble du flux
┌─────────────────┐ ┌──────────────────┐ ┌────────────────────┐
│ Tardis API │────▶│ Data Collector │────▶│ HolySheep AI API │
│ (Raw Tick Data) │ │ (Normalisation) │ │ (ML Inference) │
└─────────────────┘ └──────────────────┘ └────────────────────┘
│ │ │
▼ ▼ ▼
┌─────────────────┐ ┌──────────────────┐ ┌────────────────────┐
│ Funding Rates │ │ Feature Store │ │ Trade Signals │
│ Spot/Derivatives│ │ (Time-series) │ │ (Real-time) │
└─────────────────┘ └──────────────────┘ └────────────────────┘Composants clés
- Tardis Historical : Ingestion des ticks de marché archivés avec latence de réplication ~100ms
- Tardis Real-time : WebSocket stream des carnets d'ordres et trades
- HolySheep AI : Backend IA pour feature engineering et prédiction de funding rate
- Kafka/Redis : Buffers de streaming pour découplage des composants
Configuration Initiale du Projet
Installation des dépendances
# requirements.txt
tardis-client==2.1.4
websocket-client==1.8.0
httpx==0.27.2
pandas==2.2.3
numpy==1.26.4
asyncio-aiohttp==3.10.5
redis==5.0.8
msgpack==1.1.0
Installation
pip install -r requirements.txtConfiguration de l'environnement
# .env
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
TARDIS_API_KEY=your_tardis_api_key
TARDIS_WS_URL=wss://tardis.dev/stream
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_DB=0
Configuration trading
SYMBOLS=BTCUSDT,ETHUSDT,SOLUSDT
TIMEFRAME=1mImplémentation du Collecteur de Données Tardis
Client WebSocket Temps Réel
import asyncio
import json
import logging
from datetime import datetime
from typing import Dict, List, Optional
import httpx
import pandas as pd
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class TardisCollector:
"""
Collecteur de données temps réel depuis Tardis API.
Gère le funding rate et les ticks dérivés avec reconnexion automatique.
"""
def __init__(
self,
api_key: str,
base_url: str = "https://api.tardis.dev/v1",
symbols: List[str] = None,
exchanges: List[str] = None
):
self.api_key = api_key
self.base_url = base_url
self.symbols = symbols or ["BTCUSDT", "ETHUSDT"]
self.exchanges = exchanges or ["binance", "bybit"]
self.ws_url = "wss://tardis.dev/stream"
self.buffer: Dict[str, List] = {s: [] for s in self.symbols}
self.funding_rates: Dict[str, float] = {}
self._running = False
async def connect_websocket(self) -> None:
"""Connexion WebSocket avec gestion du heartbeat."""
import websocket
def on_message(ws, message):
data = json.loads(message)
self._process_message(data)
def on_error(ws, error):
logger.error(f"WebSocket error: {error}")
def on_close(ws, close_status_code, close_msg):
logger.warning(f"WebSocket closed: {close_status_code}")
if self._running:
asyncio.create_task(self._reconnect())
def on_open(ws):
logger.info("WebSocket connected to Tardis")
subscribe_msg = {
"type": "subscribe",
"channels": ["funding_rate", "trades", "orderbook_snapshot"],
"symbols": self.symbols,
"exchanges": self.exchanges
}
ws.send(json.dumps(subscribe_msg))
self.ws = websocket.WebSocketApp(
self.ws_url,
on_message=on_message,
on_error=on_error,
on_close=on_close,
on_open=on_open,
header={"Authorization": f"Bearer {self.api_key}"}
)
def _process_message(self, data: dict) -> None:
"""Traitement des messages selon le type."""
msg_type = data.get("type")
if msg_type == "funding_rate":
self.funding_rates[data["symbol"]] = {
"rate": float(data["rate"]),
"timestamp": data.get("timestamp"),
"next_funding": data.get("next_funding_time"),
"exchange": data.get("exchange")
}
logger.debug(f"Funding update: {data['symbol']} = {data['rate']}")
elif msg_type == "trade":
self.buffer[data["symbol"]].append({
"price": float(data["price"]),
"size": float(data["size"]),
"side": data.get("side", "unknown"),
"timestamp": data["timestamp"],
"trade_id": data.get("id")
})
elif msg_type == "orderbook_snapshot":
# Normalisation du carnet d'ordres
self._process_orderbook(data)
def _process_orderbook(self, data: dict) -> None:
"""Normalisation et stockage du carnet d'ordres."""
symbol = data["symbol"]
bids = [(float(p), float(s)) for p, s in data.get("bids", [])[:20]]
asks = [(float(p), float(s)) for p, s in data.get("asks", [])[:20]]
# Calcul du mid-price et spread
if bids and asks:
mid_price = (bids[0][0] + asks[0][0]) / 2
spread = (asks[0][0] - bids[0][0]) / mid_price
logger.debug(
f"{symbol} | Mid: {mid_price:.4f} | "
f"Spread: {spread*100:.4f}% | "
f"Bid depth: {sum(s for _, s in bids):.2f} | "
f"Ask depth: {sum(s for _, s in asks):.2f}"
)
async def _reconnect(self, delay: int = 5) -> None:
"""Reconnexion avec backoff exponentiel."""
import time
attempt = 1
max_attempts = 10
while attempt <= max_attempts:
wait_time = min(delay * (2 ** attempt), 60)
logger.info(f"Reconnecting in {wait_time}s (attempt {attempt}/{max_attempts})")
await asyncio.sleep(wait_time)
try:
await self.connect_websocket()
self.ws.run_forever()
break
except Exception as e:
logger.error(f"Reconnection failed: {e}")
attempt += 1
async def start(self) -> None:
"""Démarrage du collecteur."""
self._running = True
await self.connect_websocket()
self.ws.run_forever(ping_interval=30, ping_timeout=10)
Utilisation
if __name__ == "__main__":
collector = TardisCollector(
api_key="your_tardis_api_key",
symbols=["BTCUSDT", "ETHUSDT", "SOLUSDT"],
exchanges=["binance", "bybit", "okx"]
)
asyncio.run(collector.start())Service de Requêtes Historiques
import httpx
import pandas as pd
from datetime import datetime, timedelta
from typing import Optional, List, Dict
import logging
logger = logging.getLogger(__name__)
class TardisHistoricalClient:
"""
Client pour récupérer l'historique des données de funding et ticks.
Optimisé pour les gros volumes de données avec pagination automatique.
"""
BASE_URL = "https://api.tardis.dev/v1"
CHUNK_SIZE = 10000 # Limite par requête
def __init__(self, api_key: str):
self.api_key = api_key
self.client = httpx.Client(
timeout=httpx.Timeout(60.0, connect=10.0),
headers={"Authorization": f"Bearer {api_key}"}
)
def get_funding_rates(
self,
exchange: str,
symbol: str,
start_date: datetime,
end_date: Optional[datetime] = None,
limit: int = 1000
) -> pd.DataFrame:
"""
Récupère l'historique des funding rates.
Args:
exchange: Exchange cible (binance, bybit, okx, deribit)
symbol: Symbole de trading
start_date: Date de début
end_date: Date de fin (défaut: maintenant)
limit: Nombre maximum de records par requête
Returns:
DataFrame avec colonnes: timestamp, rate, predicted_rate, next_funding_time
"""
end_date = end_date or datetime.utcnow()
url = f"{self.BASE_URL}/funding-rates"
params = {
"exchange": exchange,
"symbol": symbol,
"start_time": int(start_date.timestamp() * 1000),
"end_time": int(end_date.timestamp() * 1000),
"limit": min(limit, self.CHUNK_SIZE)
}
all_data = []
has_more = True
while has_more:
response = self.client.get(url, params=params)
response.raise_for_status()
data = response.json()
all_data.extend(data.get("data", []))
has_more = data.get("has_more", False)
if has_more and "next_cursor" in data:
params["cursor"] = data["next_cursor"]
logger.info(
f"{symbol} {exchange}: {len(all_data)} records récupérés"
)
df = pd.DataFrame(all_data)
if not df.empty:
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
df["rate"] = df["rate"].astype(float)
return df
def get_trades(
self,
exchange: str,
symbol: str,
start_date: datetime,
end_date: Optional[datetime] = None
) -> pd.DataFrame:
"""
Récupère l'historique des trades avec métadonnées.
Returns:
DataFrame avec: timestamp, price, size, side, id, fee
"""
end_date = end_date or datetime.utcnow()
url = f"{self.BASE_URL}/trades"
params = {
"exchange": exchange,
"symbol": symbol,
"start_time": int(start_date.timestamp() * 1000),
"end_time": int(end_date.timestamp() * 1000),
"limit": self.CHUNK_SIZE,
"include_fee": True
}
all_trades = []
has_more = True
while has_more:
response = self.client.get(url, params=params)
response.raise_for_status()
data = response.json()
all_trades.extend(data.get("data", []))
has_more = data.get("has_more", False)
if has_more and "next_cursor" in data:
params["cursor"] = data["next_cursor"]
df = pd.DataFrame(all_trades)
if not df.empty:
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
df["price"] = df["price"].astype(float)
df["size"] = df["size"].astype(float)
return df
def get_orderbook_snapshots(
self,
exchange: str,
symbol: str,
start_date: datetime,
frequency: str = "1m"
) -> pd.DataFrame:
"""
Récupère les snapshots du carnet d'ordres à intervalles réguliers.
Args:
frequency: Fréquence d'échantillonnage (1s, 1m, 5m, 1h)
"""
end_date = datetime.utcnow()
url = f"{self.BASE_URL}/orderbook-snapshots"
params = {
"exchange": exchange,
"symbol": symbol,
"start_time": int(start_date.timestamp() * 1000),
"end_time": int(end_date.timestamp() * 1000),
"frequency": frequency,
"limit": 5000
}
all_snapshots = []
has_more = True
while has_more:
response = self.client.get(url, params=params)
response.raise_for_status()
data = response.json()
all_snapshots.extend(data.get("data", []))
has_more = data.get("has_more", False)
if has_more and "next_cursor" in data:
params["cursor"] = data["next_cursor"]
# Transformation en features exploitables
records = []
for snapshot in all_snapshots:
bids = snapshot.get("bids", [])
asks = snapshot.get("asks", [])
if bids and asks:
best_bid = float(bids[0][0])
best_ask = float(asks[0][0])
mid_price = (best_bid + best_ask) / 2
spread = (best_ask - best_bid) / mid_price
bid_volume = sum(float(b[1]) for b in bids[:10])
ask_volume = sum(float(a[1]) for a in asks[:10])
imbalance = (bid_volume - ask_volume) / (bid_volume + ask_volume)
records.append({
"timestamp": pd.to_datetime(snapshot["timestamp"], unit="ms"),
"symbol": symbol,
"mid_price": mid_price,
"spread_bps": spread * 10000,
"bid_depth_10": bid_volume,
"ask_depth_10": ask_volume,
"imbalance": imbalance
})
return pd.DataFrame(records)
def batch_get_funding_multi_symbols(
self,
symbols: List[str],
days_back: int = 90
) -> Dict[str, pd.DataFrame]:
"""
Récupération par lot pour optimisation des appels API.
Returns:
Dict[symbol, DataFrame] avec tous les symbols demandés
"""
start_date = datetime.utcnow() - timedelta(days=days_back)
results = {}
for symbol in symbols:
logger.info(f"Récupération funding rate pour {symbol}")
try:
df = self.get_funding_rates(
exchange="binance",
symbol=symbol,
start_date=start_date
)
results[symbol] = df
except Exception as e:
logger.error(f"Erreur pour {symbol}: {e}")
results[symbol] = pd.DataFrame()
return results
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.client.close()
Benchmark de performance
if __name__ == "__main__":
# Test de performance avec 30 jours de données
import time
start = time.time()
with TardisHistoricalClient("your_api_key") as client:
df = client.get_funding_rates(
exchange="binance",
symbol="BTCUSDT",
start_date=datetime.utcnow() - timedelta(days=30)
)
elapsed = time.time() - start
print(f"=== BENCHMARK TARDIS HISTORICAL ===")
print(f"Records récupérés: {len(df)}")
print(f"Temps total: {elapsed:.2f}s")
print(f"Taux: {len(df)/elapsed:.1f} records/s")Intégration avec HolySheep AI pour l'Inférence
Client HolySheep pour Feature Engineering
import httpx
import json
import asyncio
from typing import List, Dict, Optional, Any
from dataclasses import dataclass
from datetime import datetime
import logging
logger = logging.getLogger(__name__)
@dataclass
class HolySheepConfig:
"""Configuration du client HolySheep."""
api_key: str
base_url: str = "https://api.holysheep.ai/v1"
model: str = "deepseek-v3.2" # $0.42/MTok - meilleur rapport qualité/prix
max_tokens: int = 2048
temperature: float = 0.1
timeout: float = 30.0
class HolySheepQuantClient:
"""
Client pour l'inférence IA sur les données quantitatives via HolySheep.
Optimisé pour:
- Feature engineering automatisé
- Classification du funding rate (bull/bear/neutral)
- Détection d'anomalies dans les ticks
- Génération de signaux de trading
"""
def __init__(self, config: HolySheepConfig):
self.config = config
self.client = httpx.Client(
timeout=httpx.Timeout(config.timeout),
headers={
"Authorization": f"Bearer {config.api_key}",
"Content-Type": "application/json"
}
)
self.metrics = {
"requests": 0,
"tokens_used": 0,
"total_cost_usd": 0.0,
"avg_latency_ms": 0.0
}
self._latencies = []
def analyze_funding_rate(
self,
funding_history: List[Dict],
market_context: Dict
) -> Dict[str, Any]:
"""
Analyse le funding rate et génère des insights.
Args:
funding_history: Liste des derniers funding rates avec timestamps
market_context: Contexte de marché (prix, volume, open interest)
Returns:
Analyse structurée avec signaux et recommandations
"""
prompt = f"""Analyse quantitative du funding rate pour {market_context.get('symbol', 'UNKNOWN')}.
Contexte de marché actuel:
- Prix: ${market_context.get('price', 0):,.2f}
- Volume 24h: ${market_context.get('volume_24h', 0):,.0f}
- Open Interest: ${market_context.get('open_interest', 0):,.0f}
- Volatilité: {market_context.get('volatility', 0):.2f}%
Historique des funding rates (8h):
{json.dumps(funding_history[-12:], indent=2)}
Tâches:
1. Classifier le funding (bearish/neutral/bullish)
2. Identifier les tendances anormales
3. Estimer la probabilité de squeeze de liquidité
4. Générer un score de confiance (0-100)
Répondre en JSON structuré uniquement."""
start = asyncio.get_event_loop().time()
response = self._make_request(prompt)
latency = (asyncio.get_event_loop().time() - start) * 1000
self._update_metrics(response, latency)
return response
def generate_trading_signals(
self,
tick_data: List[Dict],
orderbook_data: Dict,
funding_data: Dict
) -> Dict[str, Any]:
"""
Génère des signaux de trading multi-facteurs.
Combine:
- Analyse des ticks (momentum, volume profile)
- Structure du carnet d'ordres (imbalance, wall detection)
- Funding rate (sentiment perpétuel)
"""
prompt = f"""Génération de signaux de trading pour marché perpétuel.
DONNÉES DE TICK (30 dernières secondes):
{json.dumps(tick_data[-20:], indent=2)}
CARNET D'ORDRES:
Bids (top 5): {json.dumps(orderbook_data.get('bids', [])[:5], indent=2)}
Asks (top 5): {json.dumps(orderbook_data.get('asks', [])[:5], indent=2)}
Imbalance: {orderbook_data.get('imbalance', 0):.4f}
FUNDING RATE:
Taux actuel: {funding_data.get('rate', 0):.6f}
Prochain funding: {funding_data.get('next_funding')}
Historique: {json.dumps(funding_data.get('history', [])[-5:], indent=2)}
Analyser et retourner en JSON:
{{
"signal": "long|short|neutral",
"confidence": 0-100,
"entry_price": float,
"stop_loss": float,
"take_profit": float,
"risk_reward": float,
"reasons": ["reason1", "reason2"],
"warnings": ["warning1"]
}}"""
start = asyncio.get_event_loop().time()
response = self._make_request(prompt)
latency = (asyncio.get_event_loop().time() - start) * 1000
self._update_metrics(response, latency)
return response
def detect_anomalies(
self,
price_series: List[float],
volume_series: List[float],
threshold: float = 2.5
) -> List[Dict]:
"""
Détecte les anomalies statistiques dans les séries de données.
Utilise Z-score et analyse de Bollinger Bands pour identifier
les mouvements de prix anormaux.
"""
prompt = f"""Détection d'anomalies dans données de marché crypto.
Prix (derniers 100 ticks): {price_series[-100:]}
Volume (derniers 100 ticks): {volume_series[-100:]}
Seuils statistiques: {threshold} sigma
Retourner en JSON:
{{
"anomalies": [
{{
"index": int,
"type": "price_spike|volume_spike|funding_squeeze",
"severity": "low|medium|high|critical",
"z_score": float,
"timestamp": "ISO string"
}}
],
"summary": {{
"total_anomalies": int,
"most_common_type": string,
"recommendation": string
}}
}}"""
start = asyncio.get_event_loop().time()
response = self._make_request(prompt)
latency = (asyncio.get_event_loop().time() - start) * 1000
self._update_metrics(response, latency)
return response
def _make_request(self, prompt: str) -> Dict:
"""Exécution de la requête API avec gestion des erreurs."""
payload = {
"model": self.config.model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": self.config.max_tokens,
"temperature": self.config.temperature
}
try:
response = self.client.post(
f"{self.config.base_url}/chat/completions",
json=payload
)
response.raise_for_status()
result = response.json()
content = result["choices"][0]["message"]["content"]
# Parsing JSON de la réponse
try:
return json.loads(content)
except json.JSONDecodeError:
return {"raw_response": content, "parse_error": True}
except httpx.HTTPStatusError as e:
logger.error(f"HTTP Error {e.response.status_code}: {e.response.text}")
raise
except httpx.TimeoutException:
logger.error("Request timeout")
raise
def _update_metrics(self, response: Dict, latency_ms: float) -> None:
"""Mise à jour des métriques de performance."""
self._latencies.append(latency_ms)
if len(self._latencies) > 1000:
self._latencies = self._latencies[-1000:]
usage = response.get("usage", {})
tokens = usage.get("total_tokens", 0)
# Calcul du coût basé sur le modèle
price_map = {
"deepseek-v3.2": 0.42,
"gpt-4.1": 8.0,
"claude-sonnet-4.5": 15.0,
"gemini-2.5-flash": 2.50
}
price_per_mtok = price_map.get(self.config.model, 0.42)
cost_usd = (tokens / 1_000_000) * price_per_mtok
self.metrics["requests"] += 1
self.metrics["tokens_used"] += tokens
self.metrics["total_cost_usd"] += cost_usd
self.metrics["avg_latency_ms"] = sum(self._latencies) / len(self._latencies)
def get_metrics(self) -> Dict:
"""Retourne les métriques de performance."""
return {
**self.metrics,
"p50_latency_ms": sorted(self._latencies)[len(self._latencies)//2] if self._latencies else 0,
"p99_latency_ms": sorted(self._latencies)[int(len(self._latencies)*0.99)] if self._latencies else 0
}
async def batch_analyze(
self,
items: List[Dict],
analysis_type: str = "funding"
) -> List[Dict]:
"""
Analyse par lot pour optimiser les coûts et la latence.
Traite plusieurs symboles en parallèle tout en respectant
les limites de rate limiting.
"""
semaphore = asyncio.Semaphore(5) # Max 5 requêtes concurrentes
async def process_single(item: Dict) -> Dict:
async with semaphore:
if analysis_type == "funding":
return self.analyze_funding_rate(
item.get("history", []),
item.get("context", {})
)
elif analysis_type == "anomaly":
return self.detect_anomalies(
item.get("prices", []),
item.get("volumes", [])
)
return {}
tasks = [process_single(item) for item in items]
return await asyncio.gather(*tasks)
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.client.close()
Test et benchmark
if __name__ == "__main__":
config = HolySheepConfig(
api_key="YOUR_HOLYSHEEP_API_KEY",
model="deepseek-v3.2" # $0.42/MTok
)
with HolySheepQuantClient(config) as client:
# Test avec données simulées
mock_funding_history = [
{"timestamp": f"2026-05-0{i}T{(j%3)*8:02d}:00:00Z", "rate": 0.0001 * (1 + j*0.1)}
for i in range(1, 6) for j in range(3)
]
mock_context = {
"symbol": "BTCUSDT",
"price": 67432.50,
"volume_24h": 28_500_000_000,
"open_interest": 18_200_000_000,
"volatility": 3.2
}
result = client.analyze_funding_rate(mock_funding_history, mock_context)
print(f"Résultat analyse: {json.dumps(result, indent=2)}")
print(f"Métriques: {client.get_metrics()}")Contrôle de Concurrence et Gestion des Erreurs
Gestionnaire de Rate Limiting
import asyncio
import time
from typing import Optional, Callable, Any
from dataclasses import dataclass, field
from collections import deque
import logging
logger = logging.getLogger(__name__)
@dataclass
class RateLimiter:
"""
Rate limiter token bucket avec burst support.
Implémente un algorithme token bucket permettant:
- Rate limiting configurable (req/s ou req/min)
- Burst capacity pour pics de charge
- Délais automatiques lors de dépassement
- Métriques de monitoring
"""
rate: float # Requêtes par seconde
burst: int = 10 # Capacité de burst
_tokens: float = field(init=False)
_last_update: float = field(init=False)
_lock: asyncio.Lock = field(default_factory=asyncio.Lock)
_request_times: deque = field(default_factory=lambda: deque(maxlen=1000))
def __post_init__(self):
self._tokens = float(self.burst)
self._last_update = time.monotonic()
async def acquire(self, timeout: Optional[float] = 30.0) -> bool:
"""
Acquiert un token pour exécuter une requête.
Args:
timeout: Temps maximum d'attente (secondes)
Returns:
True si token acquis, False si timeout
"""
start_time = time.monotonic()
while True:
async with self._lock:
now = time.monotonic()
elapsed = now - self._last_update
# Régénération des tokens
self._tokens = min(
self.burst,
self._tokens + elapsed * self.rate
)
self._last_update = now
if self._tokens >= 1:
self._tokens -= 1
self._request_times.append(now)
return True
# Attente avant retry
wait_time = (1 - self._tokens) / self.rate
if timeout and (time.monotonic() - start_time + wait_time) > timeout:
logger.warning(f"Rate limit timeout after {time.monotonic() - start_time:.2f}s")
return False
await asyncio.sleep(min(wait_time, 0.1))
def get_stats(self) -> dict:
"""Retourne les statistiques d'utilisation."""
now = time.monotonic()
recent_requests = [
t for t in self._request_times
if now - t < 60
]
return {
"current_tokens": self._tokens,
"requests_last_minute": len(recent_requests),
"requests_per_minute_actual": len(recent_requests),
"utilization": len(recent_requests) / (self.rate * 60) * 100
}
class HolySheepRateLimiter(RateLimiter):
"""
Rate limiter spécifique pour HolySheep avec gestion des erreurs 429.
"""
def __init__(self):
# HolySheep: 1000 req/min pour la plupart des endpoints
super().__init__(rate=16.67, burst=20) # ~1000/min
self._retry_after = 0
async def handle_429(self, retry_after: int) -> None:
"""Gestion des réponses 429 avec backoff."""
logger.warning(f"Rate limited. Retry-After: {retry_after}s")
self._retry_after = retry_after
await asyncio.sleep(retry_after)
class TardisRateLimiter(RateLimiter):
"""
Rate limiter pour Tardis avec limites spécifiques par plan.
"""
def __init__(self, plan: str = "starter"):
limits = {
"starter": {"rate": 5, "burst": 10},
"pro": {"rate": 20, "burst": 30},
"enterprise": {"rate": 100, "burst": 100}
}
config = limits.get(plan, limits["starter"])
super().__init__(**config)
self.plan = plan
class CircuitBreaker:
"""
Circuit breaker pattern pour resilient API calls.
États:
- CLOSED: Fonctionnement normal
- OPEN: Failures détectées, requêtes bloquées
- HALF_OPEN: Test de récupération
"""
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: int = 60,
expected_exception: type = Exception
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.expected_exception = expected_exception
self._failure_count = 0
self._last_failure_time: Optional[float] = None
self._state = "CLOSED" # CLOSED, OPEN, HALF_OPEN
self._lock = asyncio.Lock()
@property
def state(self) -> str:
if self._state == "OPEN":
if time.monotonic() - self._last_failure_time > self.recovery_timeout:
self._state = "HALF_OPEN"
return self._state
async def call(self, func: Callable, *args, **kwargs) -> Any:
"""Exécute la fonction avec protection circuit breaker."""
async with self._lock:
if self.state == "OPEN":
raise CircuitBreakerOpenError(
f"Circuit breaker OPEN. Retry after {self.recovery_timeout}s"
)
try:
result = await func(*args, **kwargs)
await self._record_success()
return result
except self.expected_exception as e:
await self._record_failure()
raise
async def _record_success(self) -> None:
async with self._lock:
self._failure_count = 0
self._state = "CLOSED"
async def _record_failure(self) -> None:
async with self._lock:
self._failure_count += 1
self._last_failure_time = time.monotonic()
if self._failure_count >= self.failure_threshold:
logger.error(
f"Circuit breaker OPENED after {self._failure_count} failures"
)
self._state = "OPEN"
def get_stats(self) -> dict:
return {
"state": self.state,
"failure_count": self._failure_count,
"last_failure": self._last_failure_time,
"time_until_retry": max(0, self.recovery_timeout - (time.monotonic() - self._last_failure_time))
if self._last_failure_time else 0
}
class CircuitBreakerOpenError(Exception):
"""Exception levée quand le circuit breaker est ouvert."""
pass
Pipeline intégré avec tous les mécanismes de rés