En 2026, le marché des cryptomonnaies atteint une capitalisation de 3,2 billions de dollars, et avec lui explosent les besoins en gestion des risques en temps réel. Les plateformes de trading traitent désormais des millions d'ordres par seconde, et les systèmes de risque doivent analyser ces flux avec une latence inférieure à 100 millisecondes pour éviter des pertes catastrophiques.
Dans ce tutoriel, je vais vous montrer comment intégrer l'API Kaiko pour les données de marché cryptocurrency avec un système de gestion des risques alimenté par l'intelligence artificielle. Et la bonne nouvelle ? Avec HolySheep AI, vous pouvez accéder aux meilleurs modèles IA à des tarifs défiant toute concurrence.
Comparatif des Coûts IA en 2026 : L'Économie Qui Change Tout
Avant d'entrer dans le vif du sujet, voici les tarifs actuels (2026) pour les principaux modèles de langage utilisés dans l'analyse financière :
| Modèle | Prix Output ($/MTok) | DeepSeek V3.2 | Gemini 2.5 Flash | GPT-4.1 | Claude Sonnet 4.5 |
|---|---|---|---|---|---|
| Prix par Million de Tokens | 0,42 $ | 2,50 $ | 8,00 $ | 15,00 $ | |
| Coût pour 10M tokens/mois | 4,20 $ | 25,00 $ | 80,00 $ | 150,00 $ | |
| Latence moyenne | <50ms | <80ms | <120ms | <150ms | |
| Économie vs Claude Sonnet 4.5 | 97% | 83% | 47% | Référence |
Pour un système de gestion des risques crypto traitant 10 millions de tokens par mois (scénario réaliste pour une plateforme de taille moyenne), DeepSeek V3.2 sur HolySheep vous coûte seulement 4,20 $ contre 150 $ sur l'API Anthropic standard. L'économie annuelle atteint 1 750 $, et ce n'est que pour UN modèle.
Pourquoi Intégrer Kaiko API avec un Système de Risque IA
Kaiko fournit des données de marché cryptocurrency de qualité institutionnelle : carnets d'ordres, trades, orderbook depth, données tick-by-tick sur plus de 30 000 paires de trading. Coupler ces flux avec un modèle IA permet de :
- Détecter les manipulations de marché en temps réel (wash trading, spoofing)
- Calculer les expositions dynamiques avec analyse sentimentale des news
- Prédire les liquidations en cascade grâce aux modèles de risque probabiliste
- Automatiser les limites de position basées sur le VaR (Value at Risk)
- Générer des rapports réglementaires automatiquement structurés
Architecture de la Solution
┌─────────────────────────────────────────────────────────────────────┐
│ ARCHITECTURE RISK MANAGEMENT │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────┐ │
│ │ Kaiko API │────▶│ Python │────▶│ Risk Engine │ │
│ │ (WebSocket) │ │ Collector │ │ (Calculations) │ │
│ └──────────────┘ └──────────────┘ └────────┬─────────┘ │
│ │ │
│ ┌──────────────┐ ┌────────▼─────────┐ │
│ │ HolySheep │◀─────────────────────────▶│ AI Analysis │ │
│ │ AI API │ DeepSeek V3.2 │ (Claude/GPT) │ │
│ │ <50ms │ 0,42 $/MTok └────────┬─────────┘ │
│ └──────────────┘ │ │
│ │ │
│ ┌─────────────────▼─────────┐ │
│ │ Dashboard / Alerting │ │
│ │ Telegram / Slack / Email │ │
│ └───────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────┘Installation et Configuration
# Installation des dépendances
pip install kaiko python-dotenv aiohttp websockets
Structure du projet
mkdir crypto-risk-system
cd crypto-risk-system
mkdir config data logs modelsConnexion à l'API Kaiko
# config/kaiko_config.py
import os
from dotenv import load_dotenv
load_dotenv()
class KaikoConfig:
API_KEY = os.getenv("KAIKO_API_KEY", "votre_cle_kaiko")
BASE_URL = "https://exchange-data-rest-api.kaiko.tech"
WS_URL = "wss://websocket.kaiko.tech/v1/data/直播"
# Endpoints principaux pour le risque
ENDPOINTS = {
"orderbook": "/api/v1/data/orderbook_l2",
"trades": "/api/v1/data/trades",
"asset_pairs": "/api/v1/data/asset_pairs",
"price_stats": "/api/v1/data/price_stats",
}
# Paires de trading surveillées
MONITORED_PAIRS = [
"btc-usdt", "eth-usdt", "sol-usdt",
"bnb-usdt", "xrp-usdt", "ada-usdt"
]
# Seuils de risque (ajustez selon vos besoins)
RISK_THRESHOLDS = {
"max_position_size_usd": 1_000_000,
"max_daily_loss_usd": 50_000,
"liquidity_ratio_min": 0.15, # 15% du volume quotidien
"volatility_alert_7d": 0.08, # 8% volatilité 7 jours
"spread_max_bps": 50, # 50 basis points
}
Exemple de configuration pour HolySheep AI
class HolySheepConfig:
API_KEY = os.getenv("HOLYSHEEP_API_KEY", "votre_cle_holysheep")
BASE_URL = "https://api.holysheep.ai/v1" # IMPORTANT: URL officielle
MODEL = "deepseek-chat" # Modèle économique pour le risque
MAX_TOKENS = 2000
TEMPERATURE = 0.3 # Réponses plus déterministes pour l'analyse
# Tarifs 2026 (inclus dans le coût total)
COST_PER_MILLION_TOKENS = 0.42 # DeepSeek V3.2
MONTHLY_BUDGET_USD = 100 # Budget maximumCollecte des Données de Marché en Temps Réel
# data/market_collector.py
import asyncio
import aiohttp
import json
import time
from typing import Dict, List, Optional
from datetime import datetime
from collections import deque
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class MarketDataCollector:
"""
Collecteur de données de marché via WebSocket Kaiko.
Gère les connexions, la reconnexion automatique et la mise en buffer.
"""
def __init__(self, config):
self.config = config
self.orderbooks = {} # {pair: deque of snapshots}
self.trades = {} # {pair: deque of recent trades}
self.price_history = deque(maxlen=1000)
self.connected = False
self.ws = None
async def connect_websocket(self):
"""Connexion au flux WebSocket Kaiko"""
headers = {
"X-API-Key": self.config.API_KEY,
"Content-Type": "application/json"
}
# Subscribe aux orderbooks et trades
subscribe_msg = {
"type": "subscribe",
"channels": [
{"name": "orderbook_l2", "assets": self.config.MONITORED_PAIRS},
{"name": "trades", "assets": self.config.MONITORED_PAIRS}
]
}
try:
async with aiohttp.ClientSession() as session:
self.ws = await session.ws_connect(
self.config.WS_URL,
headers=headers,
timeout=aiohttp.ClientTimeout(total=30)
)
await self.ws.send_json(subscribe_msg)
self.connected = True
logger.info("✅ Connecté au flux WebSocket Kaiko")
await self._process_messages()
except aiohttp.ClientError as e:
logger.error(f"❌ Erreur de connexion WebSocket: {e}")
await asyncio.sleep(5)
await self.connect_websocket() # Reconnexion automatique
async def _process_messages(self):
"""Traitement des messages en temps réel"""
while True:
msg = await self.ws.receive()
if msg.type == aiohttp.WSMsgType.TEXT:
data = json.loads(msg.data)
await self._handle_message(data)
elif msg.type == aiohttp.WSMsgType.CLOSED:
logger.warning("⚠️ WebSocket fermé, reconnexion...")
self.connected = False
await self.connect_websocket()
break
async def _handle_message(self, data: dict):
"""Route les données selon leur type"""
msg_type = data.get("type", "")
if msg_type == "orderbook_snapshot":
pair = data.get("asset")
self._update_orderbook(pair, data)
elif msg_type == "trade":
pair = data.get("asset")
self._add_trade(pair, data)
elif msg_type == "price_update":
self._update_price(data)
def _update_orderbook(self, pair: str, data: dict):
"""Met à jour le carnet d'ordres avec calcul du spread"""
if pair not in self.orderbooks:
self.orderbooks[pair] = deque(maxlen=100)
bids = data.get("bids", [])
asks = data.get("asks", [])
best_bid = float(bids[0]["price"]) if bids else 0
best_ask = float(asks[0]["price"]) if asks else 0
spread = (best_ask - best_bid) / best_bid * 10000 if best_bid > 0 else 0
mid_price = (best_bid + best_ask) / 2 if best_bid > 0 and best_ask > 0 else 0
snapshot = {
"timestamp": datetime.now().isoformat(),
"pair": pair,
"best_bid": best_bid,
"best_ask": best_ask,
"mid_price": mid_price,
"spread_bps": spread,
"bid_depth": sum(float(b["size"]) for b in bids[:10]),
"ask_depth": sum(float(a["size"]) for a in asks[:10])
}
self.orderbooks[pair].append(snapshot)
# Vérification du spread (alerte si trop large)
if spread > self.config.RISK_THRESHOLDS["spread_max_bps"]:
logger.warning(
f"🚨 {pair}: Spread anormal {spread:.1f} bps "
f"(seuil: {self.config.RISK_THRESHOLDS['spread_max_bps']} bps)"
)
def _add_trade(self, pair: str, data: dict):
"""Enregistre un trade et détecte les anomalies"""
if pair not in self.trades:
self.trades[pair] = deque(maxlen=500)
trade = {
"timestamp": data.get("timestamp", time.time()),
"pair": pair,
"price": float(data.get("price", 0)),
"size": float(data.get("size", 0)),
"side": data.get("side", "unknown"),
}
trade["value_usd"] = trade["price"] * trade["size"]
self.trades[pair].append(trade)
# Détection wash trading (mêmes tailles à intervalles réguliers)
recent_same_side = [
t for t in list(self.trades[pair])[-20:]
if t["side"] == trade["side"]
]
if len(recent_same_side) > 15:
sizes = [t["size"] for t in recent_same_side]
if len(set(sizes)) < 3: # Très peu de tailles différentes
logger.critical(
f"🚨🚨 {pair}: Wash trading suspect détecté! "
f"{len(recent_same_side)} trades similaires"
)
def _update_price(self, data: dict):
"""Historique des prix pour analyse de volatilité"""
self.price_history.append({
"timestamp": time.time(),
"pair": data.get("asset"),
"price": float(data.get("price", 0))
})
def get_liquidity_metrics(self, pair: str) -> dict:
"""Calcule les métriques de liquidité pour une paire"""
if pair not in self.orderbooks or len(self.orderbooks[pair]) < 2:
return {"error": "Données insuffisantes"}
recent = list(self.orderbooks[pair])[-1]
# Volume bid/ask sur 10 premiers niveaux
total_bid_liquidity = recent["bid_depth"] * recent["best_bid"]
total_ask_liquidity = recent["ask_depth"] * recent["best_ask"]
# Ratio de liquidité (devrait être > 0.15)
if total_bid_liquidity > 0:
liquidity_ratio = min(total_ask_liquidity, total_bid_liquidity) / max(total_ask_liquidity, total_bid_liquidity)
else:
liquidity_ratio = 0
return {
"pair": pair,
"mid_price": recent["mid_price"],
"total_bid_liquidity_usd": total_bid_liquidity,
"total_ask_liquidity_usd": total_ask_liquidity,
"liquidity_ratio": liquidity_ratio,
"risk_alert": liquidity_ratio < self.config.RISK_THRESHOLDS["liquidity_ratio_min"]
}
Test du collecteur
if __name__ == "__main__":
from config.kaiko_config import KaikoConfig
config = KaikoConfig()
collector = MarketDataCollector(config)
print("🧪 Test du collecteur de données Kaiko")
print(f"URL WebSocket: {config.WS_URL}")
print(f"Paires surveillées: {config.MONITORED_PAIRS}")
print("Démarrage dans 3 secondes...")
time.sleep(3)Analyse IA des Risques avec HolySheep
# models/risk_analyzer.py
import requests
import json
from typing import Dict, List, Optional
from datetime import datetime
import logging
logger = logging.getLogger(__name__)
class HolySheepRiskAnalyzer:
"""
Analyseur de risques utilisant l'IA via HolySheep API.
DeepSeek V3.2 pour l'analyse économique et GPT-4.1 pour les rapports détaillés.
"""
def __init__(self, config):
self.config = config
self.session_tokens = 0
self.session_cost = 0.0
def _make_request(self, model: str, messages: List[dict]) -> dict:
"""
Requête vers l'API HolySheep.
IMPORTANT: Utilise l'URL officielle https://api.holysheep.ai/v1
"""
url = f"{self.config.BASE_URL}/chat/completions"
headers = {
"Authorization": f"Bearer {self.config.API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"max_tokens": self.config.MAX_TOKENS,
"temperature": self.config.TEMPERATURE
}
try:
response = requests.post(url, headers=headers, json=payload, timeout=30)
response.raise_for_status()
result = response.json()
# Tracking des coûts (calculé par HolySheep)
usage = result.get("usage", {})
prompt_tokens = usage.get("prompt_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
total_tokens = usage.get("total_tokens", 0)
self.session_tokens += total_tokens
# DeepSeek: 0.42$/MTok, ratio approximatif output/prompt
cost = (total_tokens / 1_000_000) * self.config.COST_PER_MILLION_TOKENS
self.session_cost += cost
logger.info(
f"💬 {model}: {total_tokens} tokens "
f"(Coût session: ${self.session_cost:.4f})"
)
return result
except requests.exceptions.RequestException as e:
logger.error(f"❌ Erreur API HolySheep: {e}")
return {"error": str(e)}
def analyze_risk_realtime(self, market_data: dict, position: dict) -> dict:
"""
Analyse en temps réel des risques d'une position.
Utilise DeepSeek V3.2 pour la vitesse et le faible coût.
"""
prompt = f"""Analyse de risque en temps réel pour la position suivante:
DONNÉES DE MARCHÉ:
- Paire: {market_data.get('pair', 'N/A')}
- Prix actuel: ${market_data.get('mid_price', 0):,.2f}
- Spread: {market_data.get('spread_bps', 0):.2f} bps
- Liquidité bid: ${market_data.get('total_bid_liquidity_usd', 0):,.2f}
- Liquidité ask: ${market_data.get('total_ask_liquidity_usd', 0):,.2f}
- Ratio liquidité: {market_data.get('liquidity_ratio', 0):.2%}
POSITION ACTUELLE:
- Direction: {position.get('side', 'N/A')}
- Taille: {position.get('size', 0)} units
- Valeur USD: ${position.get('value_usd', 0):,.2f}
- Entry price: ${position.get('entry_price', 0):,.2f}
- PnL non réalisé: ${position.get('unrealized_pnl', 0):,.2f}
Seuils de risque configurés:
- Position max: ${self.config.RISK_THRESHOLDS['max_position_size_usd']:,.2f}
- Perte max quotidienne: ${self.config.RISK_THRESHOLDS['max_daily_loss_usd']:,.2f}
Réponds en JSON avec:
{{"risk_level": "LOW|MEDIUM|HIGH|CRITICAL", "score": 0-100,
"recommendations": ["action1", "action2"], "margin_call_risk": boolean}}"""
messages = [
{"role": "system", "content": "Tu es un analyste de risques financiers expert en cryptomonnaies. Réponds uniquement en JSON valide."},
{"role": "user", "content": prompt}
]
result = self._make_request("deepseek-chat", messages)
if "error" in result:
return {"risk_level": "UNKNOWN", "error": result["error"]}
try:
content = result["choices"][0]["message"]["content"]
# Nettoyage du JSON (au cas où il y a des backticks)
if content.startswith("```"):
content = content.split("```")[1]
if content.startswith("json"):
content = content[4:]
return json.loads(content)
except (json.JSONDecodeError, KeyError) as e:
logger.error(f"Erreur parsing réponse IA: {e}")
return {"risk_level": "PARSE_ERROR", "error": str(e)}
def generate_risk_report(self, all_positions: List[dict], portfolio_metrics: dict) -> str:
"""
Génère un rapport détaillé des risques du portfolio.
Utilise GPT-4.1 pour la qualité narrative supérieure.
Coût: 8$/MTok mais justifié pour les rapports réglementaires.
"""
positions_summary = "\n".join([
f"- {p['pair']}: {p['side']} {p['value_usd']:,.2f}$ "
f"(PnL: {p.get('unrealized_pnl', 0):,.2f}$)"
for p in all_positions
])
prompt = f"""Génère un rapport complet des risques de portfolio pour la direction.
MÉTRIQUES GLOBALES:
- Value at Risk (VaR) 95%: ${portfolio_metrics.get('var_95', 0):,.2f}
- Volatilité portfolio: {portfolio_metrics.get('volatility', 0):.2%}
- Sharpe Ratio: {portfolio_metrics.get('sharpe_ratio', 0):.2f}
- Exposition nette: ${portfolio_metrics.get('net_exposure', 0):,.2f}
POSITIONS:
{positions_summary}
Le rapport doit inclure:
1. Résumé exécutif (3-4 phrases)
2. Principaux facteurs de risque identifiés
3. Recommandations d'action prioritaire
4. Prévisions de liquidité sur 24h
5. Alertes réglementaires potentielles
Style: Professionnel, adapté à un comité de risque."""
messages = [
{"role": "system", "content": "Tu es un analyste financier senior spécialisé en gestion des risques. Rédige en français, ton professionnel."},
{"role": "user", "content": prompt}
]
result = self._make_request("gpt-4.1", messages)
if "error" in result:
return f"Erreur génération rapport: {result['error']}"
return result["choices"][0]["message"]["content"]
def detect_market_manipulation(self, trades: List[dict], pair: str) -> dict:
"""
Détecte les patterns de manipulation de marché via IA.
"""
# Filtrage des trades suspects
suspicious_trades = []
for i, trade in enumerate(trades):
# Vérification des patterns de spoofing
if i > 0 and trade["size"] > trades[i-1]["size"] * 5:
suspicious_trades.append({
"type": "SPOOFING_SUSPECTED",
"trade": trade,
"reason": "Taille anormalement grande vs précédent"
})
# Vérification wash trading
if i > 10:
recent_sizes = [t["size"] for t in trades[i-10:i+1]]
if len(set(recent_sizes)) < 3: # Beaucoup de tailles identiques
suspicious_trades.append({
"type": "WASH_TRADING_SUSPECTED",
"trade": trade,
"reason": f"20 derniers trades avec seulement {len(set(recent_sizes))} tailles différentes"
})
if suspicious_trades:
prompt = f"""Analyse les trades suivants pour {pair} et détermine le risque de manipulation:
{trades[:50]}
Patterns suspects identifiés:
{json.dumps(suspicious_trades[:10], indent=2)}
Réponds en JSON:
{{"manipulation_confirmed": boolean, "confidence": 0-100,
"pattern_type": "SPOOFING|WASH_TRADING|FRONTRUNNING|NONE",
"recommended_action": "string"}}"""
messages = [
{"role": "system", "content": "Tu es un expert en détection de fraude financière. Réponds en JSON uniquement."},
{"role": "user", "content": prompt}
]
result = self._make_request("deepseek-chat", messages)
if "error" not in result:
try:
return json.loads(result["choices"][0]["message"]["content"])
except json.JSONDecodeError:
pass
return {"manipulation_confirmed": False, "pattern_type": "NONE"}
def get_session_stats(self) -> dict:
"""Retourne les statistiques de la session en cours"""
return {
"total_tokens": self.session_tokens,
"estimated_cost_usd": self.session_cost,
"cost_vs_openai": self.session_cost / (self.session_tokens / 1_000_000 * 15) if self.session_tokens > 0 else 0,
"savings_percent": (1 - self.session_cost / (self.session_tokens / 1_000_000 * 15)) * 100 if self.session_tokens > 0 else 0
}
Test de l'analyseur
if __name__ == "__main__":
from config.kaiko_config import HolySheepConfig
config = HolySheepConfig()
analyzer = HolySheepRiskAnalyzer(config)
# Test avec données simulées
test_market_data = {
"pair": "btc-usdt",
"mid_price": 67500.00,
"spread_bps": 12.5,
"total_bid_liquidity_usd": 2_500_000,
"total_ask_liquidity_usd": 2_300_000,
"liquidity_ratio": 0.92
}
test_position = {
"side": "LONG",
"size": 1.5,
"value_usd": 101_250,
"entry_price": 65000,
"unrealized_pnl": 3750
}
print("🧪 Test de l'analyseur de risques HolySheep")
print(f"URL API: {config.BASE_URL}")
print(f"Coût DeepSeek V3.2: {config.COST_PER_MILLION_TOKENS}$/MTok")
print("\nAnalyse en cours...")
result = analyzer.analyze_risk_realtime(test_market_data, test_position)
print(f"\n📊 Résultat: {json.dumps(result, indent=2, ensure_ascii=False)}")
stats = analyzer.get_session_stats()
print(f"\n💰 Stats session: {stats}")Déclencheur Automatique de Réduction des Risques
# risk_engine/trigger_engine.py
import asyncio
import logging
from typing import Callable, List, Optional
from datetime import datetime
from enum import Enum
logger = logging.getLogger(__name__)
class RiskAction(Enum):
REDUCE_POSITION = "REDUCE_POSITION"
CLOSE_POSITION = "CLOSE_POSITION"
INCREASE_MARGIN = "INCREASE_MARGIN"
PAUSE_TRADING = "PAUSE_TRADING"
ALERT_ONLY = "ALERT_ONLY"
NO_ACTION = "NO_ACTION"
class RiskTriggerEngine:
"""
Moteur de déclenchement automatique basé sur les seuils de risque.
Réduit automatiquement les positions quand les limites sont atteinte.
"""
def __init__(self, config, notifier: Optional[Callable] = None):
self.config = config
self.notifier = notifier # Callback pour envoyer les alertes
self.active_positions = {}
self.triggered_count = 0
self.actions_log = []
def check_risk_triggers(self, position: dict, market_data: dict, ai_analysis: dict) -> List[dict]:
"""
Vérifie tous les déclencheurs de risque pour une position.
Retourne la liste des actions à prendre.
"""
actions = []
# 1. Vérification taille position
if position.get("value_usd", 0) > self.config.RISK_THRESHOLDS["max_position_size_usd"]:
actions.append({
"trigger": "MAX_POSITION_SIZE",
"action": RiskAction.REDUCE_POSITION,
"priority": "HIGH",
"reason": f"Position ${position.get('value_usd'):,.2f} dépasse limite ${self.config.RISK_THRESHOLDS['max_position_size_usd']:,.2f}",
"reduction_percent": min(
50,
(position.get("value_usd", 0) / self.config.RISK_THRESHOLDS["max_position_size_usd"] - 1) * 100
)
})
# 2. Vérification liquidité
liq_ratio = market_data.get("liquidity_ratio", 1)
if liq_ratio < self.config.RISK_THRESHOLDS["liquidity_ratio_min"]:
actions.append({
"trigger": "LOW_LIQUIDITY",
"action": RiskAction.REDUCE_POSITION,
"priority": "CRITICAL",
"reason": f"Ratio liquidité {liq_ratio:.2%} < {self.config.RISK_THRESHOLDS['liquidity_ratio_min']:.2%}",
"reduction_percent": 75
})
# 3. Vérification spread
spread_bps = market_data.get("spread_bps", 0)
if spread_bps > self.config.RISK_THRESHOLDS["spread_max_bps"]:
actions.append({
"trigger": "HIGH_SPREAD",
"action": RiskAction.ALERT_ONLY,
"priority": "MEDIUM",
"reason": f"Spread {spread_bps:.1f} bps dépasse seuil {self.config.RISK_THRESHOLDS['spread_max_bps']} bps"
})
# 4. Analyse IA (si disponible)
if ai_analysis and "risk_level" in ai_analysis:
risk_level = ai_analysis["risk_level"]
if risk_level == "CRITICAL":
actions.append({
"trigger": "AI_CRITICAL",
"action": RiskAction.CLOSE_POSITION,
"priority": "CRITICAL",
"reason": f"IA signale risque CRITIQUE: {ai_analysis.get('score', 'N/A')}/100",
"ai_recommendations": ai_analysis.get("recommendations", [])
})
self.triggered_count += 1
elif risk_level == "HIGH":
actions.append({
"trigger": "AI_HIGH",
"action": RiskAction.REDUCE_POSITION,
"priority": "HIGH",
"reason": f"IA signale risque HIGH: {ai_analysis.get('score', 'N/A')}/100",
"reduction_percent": 40
})
# 5. Vérification PnL quotidienne
daily_loss = position.get("daily_loss", 0)
if daily_loss > self.config.RISK_THRESHOLDS["max_daily_loss_usd"]:
actions.append({
"trigger": "DAILY_LOSS_LIMIT",
"action": RiskAction.PAUSE_TRADING,
"priority": "CRITICAL",
"reason": f"Perte quotidienne ${daily_loss:,.2f} > limite ${self.config.RISK_THRESHOLDS['max_daily_loss_usd']:,.2f}"
})
# Log et notification
if actions:
self._log_and_notify(position, actions)
return actions
def _log_and_notify(self, position: dict, actions: List[dict]):
"""Log les actions et envoie notification si configuré"""
timestamp = datetime.now().isoformat()
for action in actions:
log_entry = {
"timestamp": timestamp,
"pair": position.get("pair"),
"position_id": position.get("id"),
**action
}
self.actions_log.append(log_entry)
# Log console
priority_emoji = {"LOW": "📘", "MEDIUM": "📙", "HIGH": "📕", "CRITICAL": "🚨🚨"}
emoji = priority_emoji.get(action.get("priority", "LOW"), "📗")
logger.warning(
f"{emoji} {action['trigger']}: {action['action'].value} "
f"pour {position.get('pair')} - {action['reason']}"
)
# Notification externe
if self.notifier:
asyncio.create_task(self.notifier(position, actions))
def get_actions_summary(self) -> dict:
"""Résumé des actions de la session"""
priorities = {}
for log in self.actions_log:
p = log.get("priority", "UNKNOWN")
priorities[p] = priorities.get(p, 0) + 1
return {
"total_actions": len(self.actions_log),
"critical_triggers": self.triggered_count,
"by_priority": priorities,
"last_action": self.actions_log[-1] if self.actions_log else None
}
Test du moteur
if __name__ == "__main__":
from config.kaiko_config import KaikoConfig
from models.risk_analyzer import HolySheepRiskAnalyzer, HolySheepConfig
config = KaikoConfig()
risk_engine = RiskTriggerEngine(config)
test_position = {
"id": "POS-001",
"pair": "btc-usdt",
"side": "LONG",
"value_usd": 1_200_000, # Dépasse la limite de 1M
"daily_loss": 60_000 # Dépasse la limite de 50k
}
test_market_data = {
"pair": "btc-usdt",
"mid_price": 67500,
"spread_bps