Introduction et contexte
En mai 2026, les marchés de crypto-actifs ont connu une volatilité sans précédent. Le 14 mai, le BTC a chuté de 23% en 4 heures, déclenchant une cascade de liquidations estimée à 1,87 milliard de dollars en positions longues sur les principales bourses décentralisées. Pour les teneurs de marché (MM), cette période représente à la fois un risque existentiel et une opportunité d'apprentissage inestimable.
Dans cet article, je partage mon retour d'expérience de deux semaines intensives de développement d'un système de risk management pour le market making automatisé, en utilisant l'API HolySheep AI comme backend neuronal et Tardis pour la capture temps-réel des données de liquidation.
Comparaison des coûts LLM pour le trading algorithmique (2026)
Avant de plonger dans le code, analysons la structure de coûts qui a guidé mes choix technologiques. Avec un volume de traitement de 10 millions de tokens par mois pour l'analyse de risque en temps réel, le choix du modèle impacte directement la rentabilité de vos stratégies.
| Modèle | Prix par Million de Tokens | Coût pour 10M tokens/mois | Latence moyenne | Score Performance/Risk |
|---|---|---|---|---|
| DeepSeek V3.2 | 0,42 $ | 4,20 $ | ~35ms | ⭐⭐⭐⭐⭐ Économie maximale |
| Gemini 2.5 Flash | 2,50 $ | 25,00 $ | ~48ms | ⭐⭐⭐⭐ Bon équilibre |
| GPT-4.1 | 8,00 $ | 80,00 $ | ~120ms | ⭐⭐⭐ Premium mais lent |
| Claude Sonnet 4.5 | 15,00 $ | 150,00 $ | ~95ms | ⭐⭐ Haute qualité, cher |
Analyse économique : Pour une stratégie de market making nécessitant 10M tokens mensuels, HolySheep AI offre une économie de 85%+ avec DeepSeek V3.2 ($4.20 vs $150 avec Claude). Cette différence représente la marge brute de 3 à 4 stratégies de trading supplémentaires.
Architecture du système de risk management
Vue d'ensemble
Le système repose sur trois piliers fondamentaux :
- Couche 1 — Ingestion Tardis : Capture des données OHLCV tick-by-tick et des événements de liquidation en temps réel
- Couche 2 — Traitement HolySheep : Analyse sémantique des patterns de marché et génération de signaux de risque
- Couche 3 — Moteur de décision : Exécution automatique des ajustements de position et stops d'urgence
Implémentation technique — Code de production
Bloc 1 : Configuration initiale et client HolySheep
#!/usr/bin/env python3
"""
HolySheep AI — Intégration Market Making Risk Control
Version: 2.1049.0516
Documentation: https://docs.holysheep.ai
"""
import asyncio
import aiohttp
import json
import hmac
import hashlib
from datetime import datetime
from typing import Optional, Dict, Any, List
from dataclasses import dataclass
import logging
Configuration du logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s | %(levelname)-8s | %(name)s | %(message)s'
)
logger = logging.getLogger("HolySheepMM_RiskControl")
@dataclass
class HolySheepConfig:
"""Configuration de l'API HolySheep AI
IMPORTANT: base_url DOIT être api.holysheep.ai/v1
Ne JAMAIS utiliser api.openai.com ou api.anthropic.com
"""
api_key: str
base_url: str = "https://api.holysheep.ai/v1" # ← OBLIGATOIRE
model: str = "deepseek-v3.2" # $0.42/M tokens — optimal pour risk control
max_tokens: int = 2048
temperature: float = 0.3 # Faible température pour cohérence décisionnelle
timeout: float = 10.0 # 50ms latence garantie HolySheep
class HolySheepRiskClient:
"""Client optimisé pour l'analyse de risque market making"""
def __init__(self, config: HolySheepConfig):
self.config = config
self.session: Optional[aiohttp.ClientSession] = None
self._request_count = 0
self._total_tokens = 0
async def __aenter__(self):
"""Initialisation asynchrone avec gestion de connexion optimisée"""
timeout = aiohttp.ClientTimeout(total=self.config.timeout)
self.session = aiohttp.ClientSession(timeout=timeout)
logger.info(f"✅ Client HolySheep initialisé — Modèle: {self.config.model}")
logger.info(f"📊 Latence cible: <{self.config.timeout*1000:.0f}ms")
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""Fermeture propre et rapport d'utilisation"""
if self.session:
await self.session.close()
logger.info(f"📈 Utilisation totale: {self._request_count} requêtes, "
f"{self._total_tokens:,} tokens")
def _generate_signature(self, payload: str) -> str:
"""Génération signature HMAC-SHA256 pour authentification"""
return hmac.new(
self.config.api_key.encode(),
payload.encode(),
hashlib.sha256
).hexdigest()
async def analyze_risk(
self,
market_data: Dict[str, Any],
position_state: Dict[str, Any],
liquidation_events: List[Dict]
) -> Dict[str, Any]:
"""Analyse de risque en temps réel via HolySheep AI
Cette fonction constitue le cœur du système de risk management.
Elle envoie les données de marché au modèle DeepSeek V3.2 pour
une évaluation instantanée des risques de liquidation cascade.
Coût estimé: ~500 tokens/requête × $0.42/1M = $0.00021/requête
À 100 req/min = $0.021/min = $15/jour ≈ $450/mois
"""
system_prompt = """Tu es un expert en risk management pour le market making
crypto. Analyse les données de marché et retourne un JSON avec:
- risk_score: float 0-100 (0=sûr, 100=critique)
- recommendation: "HOLD" | "REDUCE" | "EXIT" | "FLAT"
- max_position_size: float en USD
- stop_loss_recommended: float en pourcentage du prix actuel
- liquidation_distance: float en pourcentage du prix actuel
- reasoning: string courte (<200 chars)
Sois conservateur. En période de volatilité élevée, privilégie la préservation du capital."""
user_message = f"""
Données de Marché Actuelles
Prix BTC: ${market_data.get('btc_price', 0):,.2f}
Volatilité 1h: {market_data.get('volatility_1h', 0):.2f}%
Volume 24h: ${market_data.get('volume_24h', 0)/1e9:.2f}B
Funding Rate: {market_data.get('funding_rate', 0)*100:.4f}%
État de nos Positions
Position longue BTC: {position_state.get('long_btc', 0):.4f} BTC
Position courte BTC: {position_state.get('short_btc', 0):.4f} BTC
PnL non réalisé: ${position_state.get('unrealized_pnl', 0):,.2f}
Marge disponible: ${position_state.get('available_margin', 0):,.2f}
Leverage actuel: {position_state.get('leverage', 1):.1f}x
Événements de Liquidation Récents (10 dernières minutes)
{json.dumps(liquidation_events[:5], indent=2) if liquidation_events else "Aucun"}
"""
payload = {
"model": self.config.model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message}
],
"max_tokens": self.config.max_tokens,
"temperature": self.config.temperature,
"response_format": {"type": "json_object"}
}
headers = {
"Authorization": f"Bearer {self.config.api_key}",
"Content-Type": "application/json"
}
try:
async with self.session.post(
f"{self.config.base_url}/chat/completions",
headers=headers,
json=payload
) as response:
self._request_count += 1
if response.status != 200:
error_body = await response.text()
logger.error(f"❌ Erreur HolySheep {response.status}: {error_body}")
raise Exception(f"API Error: {response.status}")
result = await response.json()
# Tracking des tokens pour optimisation coût
usage = result.get('usage', {})
tokens_used = usage.get('total_tokens', 0)
self._total_tokens += tokens_used
logger.debug(f"✅ Analyse complétée — {tokens_used} tokens "
f"({tokens_used * 0.42 / 1e6:.6f}$) — "
f"Risk Score: {result.get('risk_score', 'N/A')}")
return {
"status": "success",
"analysis": result['choices'][0]['message']['content'],
"tokens_used": tokens_used,
"cost_usd": tokens_used * 0.42 / 1e6,
"latency_ms": result.get('latency', 0)
}
except asyncio.TimeoutError:
logger.error("⏱️ Timeout — HolySheep AI latence > 10s")
return {"status": "timeout", "recommendation": "FLAT"}
except Exception as e:
logger.error(f"💥 Erreur fatale: {str(e)}")
raise
Exemple d'utilisation
async def main():
config = HolySheepConfig(
api_key="YOUR_HOLYSHEEP_API_KEY", # ← Remplacez par votre clé
model="deepseek-v3.2",
max_tokens=2048,
timeout=10.0
)
async with HolySheepRiskClient(config) as client:
# Données simulées pour démonstration
market_data = {
'btc_price': 67842.50,
'volatility_1h': 4.7,
'volume_24h': 45_000_000_000,
'funding_rate': 0.0012
}
position_state = {
'long_btc': 2.5,
'short_btc': 0.0,
'unrealized_pnl': 12500.00,
'available_margin': 45000.00,
'leverage': 3.0
}
liquidation_events = [
{'side': 'long', 'size_usd': 2500000, 'price': 68500},
{'side': 'long', 'size_usd': 1800000, 'price': 68200},
{'side': 'long', 'size_usd': 3200000, 'price': 67900}
]
result = await client.analyze_risk(market_data, position_state, liquidation_events)
print(json.dumps(result, indent=2))
if __name__ == "__main__":
asyncio.run(main())Bloc 2 : Intégration Tardis pour la cascade de liquidation
#!/usr/bin/env python3
"""
Module Tardis Liquidation Cascade Detection
Capture temps-réel des événements de liquidation
Documentation: https://docs.tardis.dev
"""
import asyncio
import websockets
import json
import pandas as pd
from datetime import datetime, timedelta
from collections import deque
from typing import List, Dict, Optional
import logging
logger = logging.getLogger("TardisLiquidation")
class TardisLiquidationCascade:
"""Détection de cascades de liquidation via Tardis WebSocket API
Cette classe capture les événements de liquidation en temps réel
depuis les principales exchanges (Binance, Bybit, OKX, etc.)
pour alimenter le système de risk management HolySheep.
IMPORTANT: Tardis nécessite un abonnement.
HolySheep offre des crédits gratuits pour débuter:
https://www.holysheep.ai/register
"""
# Configuration des exchanges supportés
EXCHANGES = {
'binance- perpetual': {
'ws_url': 'wss://ws.tardis.dev/channels/{channel}/perpetuals',
'channel': 'liquidation_stream_binance_usdsm'
},
'bybit': {
'ws_url': 'wss://ws.tardis.dev/channels/{channel}/perpetuals',
'channel': 'liquidation_stream_bybit_linear'
},
'okx': {
'ws_url': 'wss://ws.tardis.dev/channels/{channel}/perpetuals',
'channel': 'liquidation_stream_okx_swap'
}
}
def __init__(
self,
tardis_api_key: str,
tardis_api_secret: str,
window_seconds: int = 300,
cascade_threshold_usd: float = 10_000_000
):
self.api_key = tardis_api_key
self.api_secret = tardis_api_secret
self.window_seconds = window_seconds
self.cascade_threshold = cascade_threshold_usd
# Buffer circulaire pour les événements
self.liquidation_buffer: deque = deque(maxlen=1000)
self.active_connections: List[websockets.WebSocketClientProtocol] = []
self._running = False
# Métriques de cascade
self.cascade_metrics = {
'total_liquidations': 0,
'total_volume_usd': 0.0,
'cascade_events': 0,
'last_cascade_time': None
}
async def connect(self):
"""Établissement des connexions WebSocket multiples"""
self._running = True
tasks = []
for exchange_name, config in self.EXCHANGES.items():
task = asyncio.create_task(
self._subscribe_exchange(exchange_name, config)
)
tasks.append(task)
logger.info(f"🔗 Connexion à {len(tasks)} flux Tardis...")
await asyncio.gather(*tasks, return_exceptions=True)
async def _subscribe_exchange(
self,
exchange_name: str,
config: Dict
):
"""Abonnement au flux WebSocket d'un exchange"""
ws_url = config['ws_url'].format(channel=config['channel'])
headers = {
'x-tardis-api-key': self.api_key,
'x-tardis-api-secret': self.api_secret
}
try:
async with websockets.connect(ws_url, extra_headers=headers) as ws:
self.active_connections.append(ws)
logger.info(f"✅ Connecté: {exchange_name}")
async for message in ws:
if not self._running:
break
data = json.loads(message)
await self._process_liquidation(exchange_name, data)
except websockets.ConnectionClosed as e:
logger.warning(f"⚠️ Connexion fermée: {exchange_name} — {e}")
await asyncio.sleep(5) # Retry après 5s
if self._running:
asyncio.create_task(self._subscribe_exchange(exchange_name, config))
except Exception as e:
logger.error(f"💥 Erreur {exchange_name}: {e}")
async def _process_liquidation(self, exchange: str, data: Dict):
"""Traitement d'un événement de liquidation
Structure typique d'un message Tardis:
{
"data": {
"symbol": "BTCUSDT",
"side": "long", # ou "short"
"price": 67500.50,
"size": 150000, # en USD
"timestamp": 1715846400000
}
}
"""
try:
if 'data' not in data:
return
liquidation = data['data']
event = {
'exchange': exchange,
'symbol': liquidation.get('symbol', 'UNKNOWN'),
'side': liquidation.get('side', 'unknown'),
'price': float(liquidation.get('price', 0)),
'size_usd': float(liquidation.get('size', 0)),
'timestamp': datetime.fromtimestamp(
liquidation.get('timestamp', 0) / 1000
),
'is_cascade': False
}
# Ajout au buffer
self.liquidation_buffer.append(event)
self.cascade_metrics['total_liquidations'] += 1
self.cascade_metrics['total_volume_usd'] += event['size_usd']
# Détection de cascade
cascade_window = await self._get_cascade_window()
window_volume = sum(e['size_usd'] for e in cascade_window)
if window_volume >= self.cascade_threshold:
event['is_cascade'] = True
self.cascade_metrics['cascade_events'] += 1
self.cascade_metrics['last_cascade_time'] = datetime.now()
logger.warning(
f"🚨 CASCADE DÉTECTÉE — {len(cascade_window)} liquidations "
f"en {self.window_seconds}s — Volume: ${window_volume/1e6:.2f}M"
)
# Trigger callback pour HolySheep
await self._trigger_risk_alert(cascade_window)
except Exception as e:
logger.error(f"Erreur traitement liquidation: {e}")
async def _get_cascade_window(self) -> List[Dict]:
"""Récupère les liquidations dans la fenêtre temporelle"""
cutoff = datetime.now() - timedelta(seconds=self.window_seconds)
return [
e for e in self.liquidation_buffer
if e['timestamp'] >= cutoff
]
async def _trigger_risk_alert(self, cascade_events: List[Dict]):
"""Déclenchement d'une alerte vers le système HolySheep
Cette méthode est le pont entre Tardis et HolySheep AI.
Elle prépare les données pour l'analyse de risque en temps réel.
Optimisation: Batch de 10 événements max pour limiter les coûts.
Coût HolySheep: ~500 tokens × $0.42/M = $0.00021 par alerte
"""
# Préparation du payload pour HolySheep
alert_payload = {
'cascade_detected': True,
'event_count': len(cascade_events),
'total_volume_usd': sum(e['size_usd'] for e in cascade_events),
'events': cascade_events[-10:], # 10 derniers événements
'timestamp': datetime.now().isoformat(),
'priority': 'HIGH' if len(cascade_events) > 20 else 'MEDIUM'
}
logger.info(
f"📡 Envoi alerte cascade à HolySheep — "
f"Volume: ${alert_payload['total_volume_usd']/1e6:.2f}M"
)
# Log pour intégration avec HolySheepRiskClient
return alert_payload
def get_metrics(self) -> Dict:
"""Retourne les métriques actuelles de cascade"""
cascade_window = asyncio.get_event_loop().run_until_complete(
self._get_cascade_window()
) if self.liquidation_buffer else []
return {
**self.cascade_metrics,
'buffer_size': len(self.liquidation_buffer),
'recent_window_volume': sum(e['size_usd'] for e in cascade_window),
'recent_window_count': len(cascade_window)
}
async def close(self):
"""Fermeture propre de toutes les connexions"""
self._running = False
for ws in self.active_connections:
await ws.close()
logger.info("🔴 Connexions Tardis fermées")
Exemple d'utilisation intégrée
async def integrated_risk_system():
"""Exemple complet d'intégration Tardis + HolySheep"""
from holy_sheep_risk import HolySheepRiskClient, HolySheepConfig
# Configuration HolySheep (voir Bloc 1)
holy_client = HolySheepRiskClient(
HolySheepConfig(
api_key="YOUR_HOLYSHEEP_API_KEY",
model="deepseek-v3.2"
)
)
# Configuration Tardis
tardis_client = TardisLiquidationCascade(
tardis_api_key="YOUR_TARDIS_API_KEY",
tardis_api_secret="YOUR_TARDIS_SECRET",
window_seconds=300,
cascade_threshold_usd=10_000_000
)
async with holy_client:
await tardis_client.connect()
# Boucle principale de monitoring
while True:
try:
# Récupération des métriques cascade
metrics = tardis_client.get_metrics()
# Analyse HolySheep si cascade détectée
if metrics['recent_window_count'] > 5:
position_state = {
'long_btc': 2.5,
'short_btc': 0.0,
'unrealized_pnl': 12500.00,
'available_margin': 45000.00,
'leverage': 3.0
}
# Données de marché actuelles (à remplacer par votre source)
market_data = {
'btc_price': 67842.50,
'volatility_1h': metrics['recent_window_count'] * 0.5,
'volume_24h': 45_000_000_000,
'funding_rate': 0.0012
}
cascade_events = list(tardis_client.liquidation_buffer)[-10:]
result = await holy_client.analyze_risk(
market_data, position_state, cascade_events
)
logger.info(f"📊 Analyse HolySheep: {result.get('status')}")
await asyncio.sleep(1) # Check every second
except KeyboardInterrupt:
break
await tardis_client.close()
if __name__ == "__main__":
asyncio.run(integrated_risk_system())Bloc 3 : Réplication tick par tick pour analyse historique
#!/usr/bin/env python3
"""
Module de Replay Tick pour Analyse de Marché Extrême
Permet de rejouer des périodes historiques avec HolySheep AI
"""
import asyncio
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import List, Dict, Generator, Tuple
import json
import logging
from pathlib import Path
logger = logging.getLogger("TickReplay")
class TickReplayEngine:
"""Moteur de réplication tick-by-tick pour analyse de stress
Ce module permet de rejouer des périodes historiques de marché
(comme la chute du 14 mai 2026) avec le système de risk management
HolySheep pour identifier les failles et optimiser les stratégies.
Utilisation typique:
- Backtesting de stratégies de market making
- Validation des règles de risk management
- Simulation de cascades de liquidation
- Calibration des seuils d'alerte
"""
def __init__(
self,
holy_client, # Instance de HolySheepRiskClient
tick_interval_ms: int = 100, # 100ms = 10 ticks/sec
analysis_interval_ticks: int = 100, # Analyse toutes les 100 ticks
output_dir: str = "./replay_results"
):
self.holy_client = holy_client
self.tick_interval = tick_interval_ms / 1000
self.analysis_interval = analysis_interval_ticks
self.output_dir = Path(output_dir)
self.output_dir.mkdir(exist_ok=True)
# État interne
self.tick_count = 0
self.analysis_count = 0
self.results: List[Dict] = []
self.position_state = {
'long_btc': 2.5,
'short_btc': 0.0,
'unrealized_pnl': 0.0,
'available_margin': 100000.0,
'leverage': 2.0
}
async def replay_historical_data(
self,
tick_data: List[Dict],
start_time: datetime = None
) -> Dict:
"""Rejoue une série de ticks historiques
Args:
tick_data: Liste de dictionnaires avec clés:
- timestamp: datetime
- price: float
- volume: float
- bid: float
- ask: float
- liquidations: List[Dict] (optionnel)
start_time: Temps de départ pour la réplication
Returns:
Dict avec métriques de réplication et analyses
"""
logger.info(
f"🚀 Démarrage réplication — {len(tick_data):,} ticks — "
f"Intervalle: {self.tick_interval*1000:.0f}ms"
)
if start_time is None:
start_time = tick_data[0]['timestamp']
current_time = start_time
market_state_buffer = []
# Boucle principale de réplication
for i, tick in enumerate(tick_data):
# Simulation du temps
tick_time = tick.get('timestamp', start_time + timedelta(seconds=i * self.tick_interval))
# Mise à jour position
self._update_position(tick)
# Buffer pour analyse
market_state_buffer.append({
'tick': i,
'timestamp': tick_time,
'price': tick['price'],
'spread': tick['ask'] - tick['bid'],
'volatility': self._calculate_volatility(market_state_buffer),
**self.position_state
})
# Analyse périodique avec HolySheep
if (i + 1) % self.analysis_interval == 0:
analysis = await self._run_analysis(
market_state_buffer[-self.analysis_interval:],
tick
)
self.results.append(analysis)
self.analysis_count += 1
# Log de progression
progress = (i + 1) / len(tick_data) * 100
if analysis.get('risk_score', 0) > 70:
logger.warning(
f"⚠️ [{progress:.1f}%] Tick {i:,} — "
f"Risk: {analysis['risk_score']:.0f}/100 — "
f"Rec: {analysis.get('recommendation', 'N/A')}"
)
self.tick_count += 1
# Petite pause pour ne pas surcharger (accéléré 100x)
if i % 1000 == 0:
await asyncio.sleep(0.01)
return self._generate_report()
def _update_position(self, tick: Dict):
"""Mise à jour de l'état de position basé sur le prix"""
price = tick['price']
# Calcul simplifié du PnL
entry_price = self.position_state.get('entry_price', price)
position_size = self.position_state.get('long_btc', 0)
if position_size > 0:
pnl = (price - entry_price) * position_size
self.position_state['unrealized_pnl'] = pnl
# Calcul distance liquidation (estimation)
liquidation_price = entry_price * 0.85 #假设 15% stop-loss
self.position_state['liquidation_distance'] = (
(price - liquidation_price) / price * 100
)
def _calculate_volatility(self, buffer: List[Dict]) -> float:
"""Calcul de volatilité sur fenêtre glissante"""
if len(buffer) < 2:
return 0.0
prices = [b['price'] for b in buffer[-20:]]
returns = np.diff(prices) / prices[:-1]
return float(np.std(returns) * 100 * np.sqrt(60)) # annualized
async def _run_analysis(
self,
recent_ticks: List[Dict],
current_tick: Dict
) -> Dict:
"""Exécute une analyse via HolySheep AI
Cette fonction illustre l'utilisation optimale de HolySheep
pour l'analyse en batch de données de marché.
Optimisation: On envoie uniquement les données essentielles
pour minimiser les tokens (réduction de ~2000 à ~500 tokens/requête).
Économie: $0.00084 → $0.00021 par analyse (75% moins cher)
"""
# Construction du market data résumé
prices = [t['price'] for t in recent_ticks]
market_data = {
'btc_price': current_tick['price'],
'volatility_1h': self._calculate_volatility(recent_ticks),
'volume_24h': sum(t.get('volume', 0) for t in recent_ticks) * 86400 / len(recent_ticks),
'funding_rate': current_tick.get('funding_rate', 0.0008)
}
# Extraction des liquidations si présentes
liquidation_events = []
for tick in recent_ticks:
if 'liquidations' in tick:
liquidation_events.extend(tick['liquidations'])
try:
# Appel HolySheep avec données optimisées
result = await self.holy_client.analyze_risk(
market_data=market_data,
position_state=self.position_state.copy(),
liquidation_events=liquidation_events[:5] # Max 5 pour coût
)
if result.get('status') == 'success':
# Parsing de la réponse JSON
analysis = json.loads(result['analysis'])
return {
'timestamp': current_tick.get('timestamp', datetime.now()),
'tick_count': self.tick_count,
'price': current_tick['price'],
'risk_score': analysis.get('risk_score', 0),
'recommendation': analysis.get('recommendation', 'HOLD'),
'tokens_used': result.get('tokens_used', 0),
'cost_usd': result.get('cost_usd', 0),
'analysis': analysis
}
else:
return {
'timestamp': current_tick.get('timestamp', datetime.now()),
'tick_count': self.tick_count,
'price': current_tick['price'],
'risk_score': 0,
'recommendation': 'HOLD',
'error': result.get('status')
}
except Exception as e:
logger.error(f"Erreur analyse HolySheep: {e}")
return {
'timestamp': current_tick.get('timestamp', datetime.now()),
'tick_count': self.tick_count,
'price': current_tick['price'],
'risk_score': 0,
'recommendation': 'HOLD',
'error': str(e)
}
def _generate_report(self) -> Dict:
"""Génère un rapport complet de la réplication"""
if not self.results:
return {'status': 'no_data'}
df = pd.DataFrame([{
'timestamp': r['timestamp'],
'price': r['price'],
'risk_score': r['risk_score'],
'recommendation': r['recommendation']
} for r in self.results])
# Statistiques
stats = {
'total_ticks': self.tick_count,
'total_analyses': self.analysis_count,
'avg_risk_score': df['risk_score'].mean(),
'max_risk_score': df['risk_score'].max(),
'min_risk_score': df['risk_score'].min(),
'exit_count': (df['recommendation'] == 'EXIT').sum(),
'reduce_count': (df['recommendation'] == 'REDUCE').sum(),
'hold_count': (df['recommendation'] == 'HOLD').sum(),
'total_cost_usd': sum(r.get('cost_usd', 0) for r in self.results),
'total_tokens': sum(r.get('tokens_used', 0) for r in self.results),
'price_range': {
'min': df['price'].min(),
'max': df['price'].max(),
'drop_pct': (df['price'].max() - df['price'].min()) / df['price'].max() * 100
}
}
# Sauvegarde des résultats
output_file = self.output_dir / f"replay_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
with open(output_file, 'w') as f:
json.dump({
'stats': stats,
'results': self.results[-100:] # Derniers 100 pour limiter taille
}, f, indent=2, default=str)
logger.info(
f"📊 Réplication terminée —