Dans l'écosystème crypto de 2026, la construction d'un modèle de gestion des risques quantitatifs fiable repose sur l'accès à des données de liquidation précises et à un historique granulaire des marchés. Cet article détaille comment intégrer les données de liquidation Binance Futures avec l'API historique Tardis pour créer un système de risk management robuste, tout en accélérant drastiquement votre développement avec l'IA générative de HolySheep AI.
Tableau comparatif : Sources de données et solutions d'intégration
| Critère | HolySheep AI | API officielle Binance | Tardis Historical | Autres relais (Kaiko, CoinMetrics) |
|---|---|---|---|---|
| Latence moyenne | <50ms | Variable (200-500ms) | 100-300ms | 500ms-2s |
| Prix par million de tokens | DeepSeek V3.2 : $0.42 | N/A (données brutes) | $50-500/mois selon plan | $200-2000/mois |
| Économie vs concurrence | 85%+ (taux ¥1=$1) | Gratuit (limité) | Standard | Élevé |
| Modes de paiement | WeChat Pay, Alipay, USDT | Carte, transfer | Carte, wire | Carte uniquement |
| Crédits gratuits | Oui, dès l'inscription | Non | Essai 7 jours | Essai limité |
| Analyse IA des données | Intégrée nativement | Non | Non | Addon coûteux |
Introduction : Pourquoi combiner Binance Liquidation Data et Tardis API ?
En tant qu'ingénieur quantitatif ayant travaillé sur des desks de trading haute fréquence pendant 8 ans, j'ai constaté que la qualité du risque management repose à 80% sur la qualité des données sous-jacentes. Les données de liquidation Binance Futures offrent une vision unique des changements brutaux de sentiment du marché, tandis que l'API Tardis fournit l'historique minute-by-minute nécessaire pour reconstruire le contexte temporel.
Dans ce tutoriel complet, nous allons construire :
- Un pipeline d'ingestion des données de liquidation en temps réel
- Un système de reconstruction historique avec Tardis
- Un modèle de risk scoring alimenté par IA générative via HolySheep AI
- Un système d'alertes automatisé pour les positions à risque
Prérequis et architecture du système
Architecture globale
┌─────────────────────────────────────────────────────────────────┐
│ ARCHITECTURE RISK MODEL │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────┐ │
│ │ Binance │ │ Tardis │ │ HolySheep AI │ │
│ │ WebSocket │───▶│ Historical │───▶│ (Risk Analysis) │ │
│ │ Liquidation │ │ API │ │ <50ms latency │ │
│ └──────────────┘ └──────────────┘ └──────────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ PostgreSQL + TimescaleDB │ │
│ │ (Historical Storage) │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Risk Scoring Engine │ │
│ │ (Python + TensorFlow) │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Alert System (Telegram/Slack) │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘Installation des dépendances
# Installation des packages Python requis
pip install asyncio-websocket==0.10.0
pip install tardis-client==2.0.0
pip install pandas numpy
pip install sqlalchemy timescaledb
pip install python-dotenv
pip install httpx aiohttp
Variables d'environnement (.env)
cat > .env << 'EOF'
BINANCE_WS_ENDPOINT=wss://fstream.binance.com/ws
TARDIS_API_KEY=your_tardis_api_key
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
POSTGRES_CONNECTION=postgresql://user:pass@localhost:5432/riskdb
EOFPartie 1 : Ingestion des données de liquidation Binance en temps réel
#!/usr/bin/env python3
"""
Binance Futures Liquidation Stream Collector
Auteur: HolySheep AI Technical Team
Version: 2026.04
"""
import asyncio
import json
import logging
from datetime import datetime
from dataclasses import dataclass, asdict
from typing import Dict, List, Optional
import aiohttp
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, BigInteger, Float, String, DateTime, Index
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
Base = declarative_base()
@dataclass
class LiquidationRecord:
"""Modèle de données pour une liquidation"""
event_time: int
symbol: str
side: str # BUY ou SELL
price: float
quantity: float
trade_time: int
is_auto_liquidate: bool
created_at: datetime = None
class LiquidationTable(Base):
__tablename__ = 'liquidations'
id = Column(BigInteger, primary_key=True, autoincrement=True)
event_time = Column(BigInteger, nullable=False)
symbol = Column(String(20), nullable=False)
side = Column(String(10), nullable=False)
price = Column(Float, nullable=False)
quantity = Column(Float, nullable=False)
trade_time = Column(BigInteger, nullable=False)
is_auto_liquidate = Column(String(5), nullable=False)
created_at = Column(DateTime, default=datetime.utcnow)
__table_args__ = (
Index('idx_symbol_time', 'symbol', 'event_time'),
Index('idx_trade_time', 'trade_time'),
)
class BinanceLiquidationCollector:
"""Collecteur WebSocket pour les liquidations Binance Futures"""
def __init__(self, symbols: List[str], db_url: str):
self.base_url = "wss://fstream.binance.com/ws"
self.symbols = [f"{s.lower()}_perpetual" for s in symbols]
self.db_engine = create_async_engine(db_url, echo=False)
self.running = False
self.buffer: List[LiquidationRecord] = []
self.buffer_size = 100
async def initialize_database(self):
"""Initialise les tables de la base de données"""
async with self.db_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
logger.info("✅ Base de données initialisée avec succès")
def create_subscribe_message(self) -> str:
"""Génère le message de subscription WebSocket"""
streams = [f"{s}@liquidation" for s in self.symbols]
return json.dumps({
"method": "SUBSCRIBE",
"params": streams,
"id": 1
})
async def process_liquidation(self, data: dict) -> Optional[LiquidationRecord]:
"""Traite un événement de liquidation"""
try:
record = LiquidationRecord(
event_time=data['E'],
symbol=data['s'],
side=data['o']['S'],
price=float(data['o']['p']),
quantity=float(data['o']['q']),
trade_time=data['o']['T'],
is_auto_liquidate=str(data['o']['X'])
)
self.buffer.append(record)
# Flush quand le buffer est plein
if len(self.buffer) >= self.buffer_size:
await self.flush_buffer()
return record
except KeyError as e:
logger.error(f"❌ Erreur de parsing: {e}")
return None
async def flush_buffer(self):
"""Écrit le buffer en base de données"""
if not self.buffer:
return
async with AsyncSession(self.db_engine) as session:
records = [
LiquidationTable(**asdict(r))
for r in self.buffer
]
session.add_all(records)
await session.commit()
logger.info(f"📊 Flush: {len(self.buffer)} liquidations écrites")
self.buffer.clear()
async def connect(self):
"""Connexion principale au WebSocket Binance"""
await self.initialize_database()
self.running = True
while self.running:
try:
async with aiohttp.ClientSession() as session:
async with session.ws_connect(
self.base_url,
timeout=aiohttp.ClientTimeout(total=30)
) as ws:
# Subscribe
await ws.send_str(self.create_subscribe_message())
logger.info(f"📡 Connecté aux streams: {self.symbols}")
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
data = json.loads(msg.data)
# Skip subscription confirmation
if 'result' in data:
continue
# Traiter la liquidation
record = await self.process_liquidation(data)
if record:
logger.info(
f"💥 Liquidation: {record.symbol} | "
f"{record.side} | {record.quantity} @ {record.price}"
)
elif msg.type == aiohttp.WSMsgType.ERROR:
logger.error(f"❌ Erreur WebSocket: {msg.data}")
break
except aiohttp.ClientError as e:
logger.warning(f"⚠️ Reconnexion dans 5s: {e}")
await asyncio.sleep(5)
async def stop(self):
"""Arrête le collecteur proprement"""
self.running = False
await self.flush_buffer()
await self.db_engine.dispose()
logger.info("🛑 Collecteur arrêté")
Point d'entrée
if __name__ == "__main__":
from dotenv import load_dotenv
load_dotenv()
symbols = ["BTCUSDT", "ETHUSDT", "BNBUSDT", "SOLUSDT"]
collector = BinanceLiquidationCollector(
symbols=symbols,
db_url=os.getenv("POSTGRES_CONNECTION")
)
try:
asyncio.run(collector.connect())
except KeyboardInterrupt:
asyncio.run(collector.stop())Partie 2 : Intégration de l'API Tardis pour l'historique
#!/usr/bin/env python3
"""
Tardis Historical API Integration
Récupération et reconstruction de l'historique des marchés
"""
import asyncio
import httpx
from datetime import datetime, timedelta
from typing import List, Dict, Generator
from dataclasses import dataclass
import pandas as pd
from dotenv import load_dotenv
import os
load_dotenv()
@dataclass
class OHLCV:
"""Données OHLCV pour une période"""
timestamp: datetime
open: float
high: float
low: float
close: float
volume: float
trades: int
@dataclass
class FundingRate:
"""Taux de funding historique"""
timestamp: datetime
funding_rate: float
next_funding_time: datetime
class TardisAPIClient:
"""
Client pour l'API Tardis Historical
Documentation: https://docs.tardis.dev/
"""
BASE_URL = "https://api.tardis.dev/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
self.client = httpx.AsyncClient(
timeout=httpx.Timeout(30.0),
headers=self.headers
)
async def get_exchanges(self) -> List[Dict]:
"""Liste les exchanges disponibles"""
response = await self.client.get(f"{self.BASE_URL}/exchanges")
response.raise_for_status()
return response.json()
async def get_symbols(self, exchange: str) -> List[Dict]:
"""Liste les symbols disponibles pour un exchange"""
response = await self.client.get(
f"{self.BASE_URL}/exchanges/{exchange}/symbols"
)
response.raise_for_status()
return response.json()
async def fetch_liquidation_history(
self,
symbol: str,
start_date: datetime,
end_date: datetime,
chunksize: timedelta = timedelta(hours=1)
) -> Generator[List[Dict], None, None]:
"""
Fetch l'historique des liquidations par chunks
Args:
symbol: Paire de trading (ex: "binance-futures:BTC-USDT-PERPETUAL")
start_date: Date de début
end_date: Date de fin
chunksize: Taille du chunk temporel
Yields:
Listes de records de liquidation
"""
current = start_date
while current < end_date:
chunk_end = min(current + chunksize, end_date)
params = {
"symbol": symbol,
"startDate": current.isoformat(),
"endDate": chunk_end.isoformat(),
"types": "liquidation"
}
try:
response = await self.client.get(
f"{self.BASE_URL}/historical/liquidation",
params=params
)
response.raise_for_status()
data = response.json()
if data.get('data'):
yield data['data']
logger.info(
f"📥 Chunk {current} → {chunk_end}: "
f"{len(data.get('data', []))} liquidations"
)
except httpx.HTTPStatusError as e:
logger.error(f"❌ Erreur API Tardis: {e.response.status_code}")
if e.response.status_code == 429:
await asyncio.sleep(60) # Rate limit
else:
break
current = chunk_end
async def fetch_ohlcv(
self,
symbol: str,
start_date: datetime,
end_date: datetime,
timeframe: str = "1m"
) -> List[OHLCV]:
"""
Récupère les données OHLCV historiques
Args:
symbol: Format "exchange:symbol"
timeframe: 1m, 5m, 15m, 1h, 4h, 1d
"""
params = {
"symbol": symbol,
"startDate": start_date.isoformat(),
"endDate": end_date.isoformat(),
"timeframe": timeframe
}
response = await self.client.get(
f"{self.BASE_URL}/historical/ohlcv",
params=params
)
response.raise_for_status()
data = response.json()
return [
OHLCV(
timestamp=datetime.fromisoformat(item['timestamp']),
open=float(item['open']),
high=float(item['high']),
low=float(item['low']),
close=float(item['close']),
volume=float(item['volume']),
trades=item.get('trades', 0)
)
for item in data.get('data', [])
]
async def fetch_funding_rates(
self,
symbol: str,
start_date: datetime,
end_date: datetime
) -> List[FundingRate]:
"""Récupère l'historique des taux de funding"""
params = {
"symbol": symbol,
"startDate": start_date.isoformat(),
"endDate": end_date.isoformat(),
"types": "funding"
}
response = await self.client.get(
f"{self.BASE_URL}/historical/compact",
params=params
)
response.raise_for_status()
data = response.json()
return [
FundingRate(
timestamp=datetime.fromisoformat(item['timestamp']),
funding_rate=float(item['fundingRate']),
next_funding_time=datetime.fromisoformat(item.get('nextFundingTime', ''))
)
for item in data.get('data', [])
]
async def close(self):
await self.client.aclose()
async def main():
"""Exemple d'utilisation"""
tardis = TardisAPIClient(api_key=os.getenv("TARDIS_API_KEY"))
# Définir la période d'analyse
end_date = datetime.utcnow()
start_date = end_date - timedelta(days=30)
# Récupérer les liquidations BTC
symbol = "binance-futures:BTC-USDT-PERPETUAL"
print(f"📊 Récupération des liquidations pour {symbol}")
print(f" Période: {start_date} → {end_date}")
all_liquidations = []
async for chunk in tardis.fetch_liquidation_history(
symbol=symbol,
start_date=start_date,
end_date=end_date,
chunksize=timedelta(hours=6)
):
all_liquidations.extend(chunk)
print(f"✅ Total liquidations: {len(all_liquidations)}")
# Analyser la distribution
if all_liquidations:
df = pd.DataFrame(all_liquidations)
print(f"\n📈 Statistiques:")
print(f" Volume total: ${df['amount'].sum():,.2f}")
print(f" Volume moyen: ${df['amount'].mean():,.2f}")
print(f" Max liquidation: ${df['amount'].max():,.2f}")
await tardis.close()
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
asyncio.run(main())Partie 3 : Modèle de risk scoring avec HolySheep AI
Ici intervient HolySheep AI pour accélérer drastiquement le développement de votre modèle de risk scoring. Avec une latence inférieure à 50ms et des prix à partir de $0.42/M tokens pour DeepSeek V3.2, HolySheep offre le meilleur rapport performance/coût du marché en 2026.
#!/usr/bin/env python3
"""
Risk Scoring Model avec HolySheep AI
Auteur: HolySheep AI Integration
"""
import os
import json
import asyncio
import httpx
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import List, Dict, Optional
from dotenv import load_dotenv
import pandas as pd
load_dotenv()
⚠️ CONFIGURATION HOLYSHEEP AI - Ne remplacez JAMAIS par api.openai.com
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
@dataclass
class RiskAnalysis:
"""Résultat de l'analyse de risque par l'IA"""
symbol: str
risk_score: float # 0-100
risk_level: str # LOW, MEDIUM, HIGH, CRITICAL
factors: List[str]
recommendations: List[str]
confidence: float
class HolySheepRiskAnalyzer:
"""
Analyseur de risque utilisant l'IA générative HolySheep
Avantages HolySheep:
- Latence <50ms (vs 200-500ms sur OpenAI)
- Prix: DeepSeek V3.2 à $0.42/M tokens (vs $8/M sur GPT-4.1)
- Support natif WeChat/Alipay
- Crédits gratuits dès l'inscription
"""
def __init__(self, api_key: str = None):
self.api_key = api_key or HOLYSHEEP_API_KEY
self.base_url = HOLYSHEEP_BASE_URL
self.headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
self.client = httpx.AsyncClient(
timeout=httpx.Timeout(60.0),
headers=self.headers
)
async def analyze_liquidation_pattern(
self,
liquidation_data: pd.DataFrame,
ohlcv_data: pd.DataFrame,
symbol: str
) -> RiskAnalysis:
"""
Analyse les patterns de liquidation pour générer un score de risque
Utilise HolySheep AI pour interpréter les données et produire
des recommandations actionnables.
"""
# Préparer le prompt avec les données
recent_liquidations = liquidation_data.tail(100)
recent_ohlcv = ohlcv_data.tail(100)
prompt = f"""Analyse de risque quantitatif pour {symbol}
DONNÉES DE LIQUIDATION RÉCENTES:
{recent_liquidations.to_string()}
DONNÉES OHLCV RÉCENTES:
{recent_ohlcv.to_string()}
INDICATEURS CALCULÉS:
- Volume total liquidations 24h: {recent_liquidations['amount'].sum():,.2f} USDT
- Nombre de liquidations: {len(recent_liquidations)}
- Prix actuel: {recent_ohlcv['close'].iloc[-1]:,.2f}
- Volatilité 24h: {recent_ohlcv['close'].pct_change().std() * 100:.2f}%
- Volume trading 24h: {recent_ohlcv['volume'].sum():,.2f}
TÂCHE: Analyse le risque actuel et fournis:
1. Un score de risque (0-100)
2. Le niveau (LOW/MEDIUM/HIGH/CRITICAL)
3. Les facteurs clés de risque
4. Des recommandations de position (augmenter/réduire/maintenir)
5. Un niveau de confiance (0-1)
Réponds en JSON avec ce format exact:
{{"risk_score": 0-100, "risk_level": "", "factors": [], "recommendations": [], "confidence": 0.0}}
"""
try:
response = await self.client.post(
f"{self.base_url}/chat/completions",
json={
"model": "deepseek-chat-v3.2", # $0.42/M tokens 💰
"messages": [
{"role": "system", "content": "Tu es un analyste de risque quantitatif expert en crypto. Réponds uniquement en JSON valide."},
{"role": "user", "content": prompt}
],
"temperature": 0.3, # Réponses déterministes pour le trading
"max_tokens": 500
}
)
response.raise_for_status()
result = response.json()
# Parser la réponse JSON
content = result['choices'][0]['message']['content']
analysis_data = json.loads(content)
return RiskAnalysis(
symbol=symbol,
risk_score=analysis_data['risk_score'],
risk_level=analysis_data['risk_level'],
factors=analysis_data['factors'],
recommendations=analysis_data['recommendations'],
confidence=analysis_data['confidence']
)
except httpx.HTTPStatusError as e:
print(f"❌ Erreur HolySheep API: {e.response.status_code}")
return None
except json.JSONDecodeError as e:
print(f"❌ Erreur parsing JSON: {e}")
return None
async def batch_analyze(
self,
symbols_data: Dict[str, Dict]
) -> List[RiskAnalysis]:
"""
Analyse par lot plusieurs symbols simultanément
HolySheep supporte les requêtes parallèles avec latence <50ms
"""
tasks = [
self.analyze_liquidation_pattern(
data['liquidations'],
data['ohlcv'],
symbol
)
for symbol, data in symbols_data.items()
]
results = await asyncio.gather(*tasks)
return [r for r in results if r is not None]
async def generate_portfolio_report(
self,
positions: List[Dict],
market_data: Dict
) -> str:
"""
Génère un rapport de risk management complet du portfolio
Utilise le modèle GPT-4.1 via HolySheep pour une analyse détaillée
Coût: $8/M tokens via HolySheep (vs prix standard)
"""
prompt = f"""Génère un rapport de risk management pour le portfolio suivant:
POSITIONS ACTUELLES:
{json.dumps(positions, indent=2)}
DONNÉES DE MARCHÉ:
{json.dumps(market_data, indent=2)}
Le rapport doit inclure:
1. Résumé exécutif du risque global
2. Top 3 positions à risque
3. Recommandations de hedging
4. Alerts prioritaires
5. Allocation optimale suggérée
Style: Professionnel, concise, orientée action."""
response = await self.client.post(
f"{self.base_url}/chat/completions",
json={
"model": "gpt-4.1", # $8/M tokens - modèle premium
"messages": [
{"role": "system", "content": "Tu es un risk manager expert en finance quantitative."},
{"role": "user", "content": prompt}
],
"temperature": 0.2,
"max_tokens": 1500
}
)
result = response.json()
return result['choices'][0]['message']['content']
async def close(self):
await self.client.aclose()
async def example_usage():
"""Exemple d'utilisation complète"""
# Initialiser l'analyseur HolySheep
analyzer = HolySheepRiskAnalyzer()
# Simuler des données
sample_liquidations = pd.DataFrame({
'timestamp': pd.date_range('2026-04-27', periods=50, freq='1H'),
'symbol': ['BTCUSDT'] * 50,
'amount': [10000 + i * 100 for i in range(50)],
'side': ['SELL' if i % 2 == 0 else 'BUY' for i in range(50)]
})
sample_ohlcv = pd.DataFrame({
'timestamp': pd.date_range('2026-04-27', periods=100, freq='15T'),
'open': [65000 + i * 10 for i in range(100)],
'high': [65100 + i * 10 for i in range(100)],
'low': [64900 + i * 10 for i in range(100)],
'close': [65000 + i * 10 for i in range(100)],
'volume': [1000 + i * 50 for i in range(100)]
})
# Analyser le risque
analysis = await analyzer.analyze_liquidation_pattern(
liquidation_data=sample_liquidations,
ohlcv_data=sample_ohlcv,
symbol="BTCUSDT"
)
if analysis:
print(f"📊 Analyse de risque pour {analysis.symbol}")
print(f" Score: {analysis.risk_score}/100 ({analysis.risk_level})")
print(f" Confiance: {analysis.confidence:.1%}")
print(f" Facteurs: {', '.join(analysis.factors[:3])}")
await analyzer.close()
if __name__ == "__main__":
asyncio.run(example_usage())Partie 4 : Système d'alertes automatisé
#!/usr/bin/env python3
"""
Alert System - Intégration Telegram/Slack
"""
import os
import asyncio
import httpx
from datetime import datetime
from dataclasses import dataclass
from enum import Enum
from typing import List, Optional
from dotenv import load_dotenv
load_dotenv()
class AlertLevel(Enum):
INFO = "ℹ️ INFO"
WARNING = "⚠️ WARNING"
HIGH = "🚨 HIGH"
CRITICAL = "🔴 CRITICAL"
@dataclass
class RiskAlert:
level: AlertLevel
symbol: str
title: str
message: str
risk_score: float
timestamp: datetime
recommendations: List[str]
class AlertDispatcher:
"""Système de dispatch des alertes vers multiples canaux"""
def __init__(self):
self.telegram_token = os.getenv("TELEGRAM_BOT_TOKEN")
self.telegram_chat_id = os.getenv("TELEGRAM_CHAT_ID")
self.slack_webhook = os.getenv("SLACK_WEBHOOK_URL")
async def send_telegram(self, alert: RiskAlert) -> bool:
"""Envoie une alerte via Telegram Bot"""
if not self.telegram_token or not self.telegram_chat_id:
return False
emoji = alert.level.value
message = f"""
{emoji} *ALERTE RISQUE - {alert.symbol}*
📌 *Titre:* {alert.title}
📊 *Score:* {alert.risk_score}/100
🕐 *Heure:* {alert.timestamp.strftime('%Y-%m-%d %H:%M:%S UTC')}
📝 *Message:*
{alert.message}
💡 *Recommandations:*
{chr(10).join([f"• {r}" for r in alert.recommendations])}
_Generé automatiquement par HolySheep Risk Model_
"""
url = f"https://api.telegram.org/bot{self.telegram_token}/sendMessage"
try:
async with httpx.AsyncClient() as client:
response = await client.post(url, json={
"chat_id": self.telegram_chat_id,
"text": message,
"parse_mode": "Markdown"
})
return response.status_code == 200
except Exception as e:
print(f"❌ Erreur Telegram: {e}")
return False
async def send_slack(self, alert: RiskAlert) -> bool:
"""Envoie une alerte via Slack Webhook"""
if not self.slack_webhook:
return False
color_map = {
AlertLevel.INFO: "#36a64f",
AlertLevel.WARNING: "#ff9900",
AlertLevel.HIGH: "#ff6600",
AlertLevel.CRITICAL: "#ff0000"
}
payload = {
"attachments": [{
"color": color_map[alert.level],
"blocks": [
{
"type": "header",
"text": {
"type": "plain_text",
"text": f"{alert.level.value} {alert.symbol}"
}
},
{
"type": "section",
"fields": [
{"type": "mrkdwn", "text": f"*Score:*\n{alert.risk_score}/100"},
{"type": "mrkdwn", "text": f"*Heure:*\n{alert.timestamp.isoformat()}"}
]
},
{
"type": "section",
"text": {"type": "mrkdwn", "text": alert.message}
}
]
}]
}
try:
async with httpx.AsyncClient() as client:
response = await client.post(
self.slack_webhook,
json=payload
)
return response.status_code == 200
except Exception as e:
print(f"❌ Erreur Slack: {e}")
return False
async def dispatch(self, alert: RiskAlert):
"""Dispatch l'alerte vers tous les canaux configurés"""
tasks = [
self.send_telegram(alert),
self.send_slack(alert)
]
results = await asyncio.gather(*tasks, return_exceptions=True)
success = sum(1 for r in results if r is True)
print(f"📨 Alerte dispatchée: {success}/{len(results)} canaux")Pour qui / pour qui ce n'est pas fait
| ✅ Ce tutoriel est fait pour | ❌ Ce tutoriel n'est pas fait pour |
|---|---|
|
|