En 2026, la sécurité des plateformes d'échange de cryptomonnaie représente un défi critique. Chaque jour, des millions de transactions transitent par les API des exchanges, générant des volumes massifs de logs. La détection automatique des comportements suspects devient non seulement une nécessité réglementaire, mais aussi un avantage compétitif majeur. Dans ce tutoriel complet, je vous explique comment implémenter un système de reconnaissance de patterns anormaux basé sur l'intelligence artificielle, en optimisant vos coûts d'infrastructure grâce aux API IA les plus performantes du marché.
Comparatif des Coûts API IA pour l'Analyse de Logs (2026)
Avant de plongez dans l'implémentation technique, analysons la rentabilité des différentes options disponibles sur le marché. Les tarifs ci-dessous incluent les prix output par million de tokens, qui représentent le coût réel de vos requêtes d'analyse.
| Modèle IA | Prix / 1M Tokens Output | Coût pour 10M Tokens/mois | Latence Moyenne | Score Performance |
|---|---|---|---|---|
| DeepSeek V3.2 | 0,42 $ | 4,20 $ | 45 ms | ★★★★☆ |
| Gemini 2.5 Flash | 2,50 $ | 25,00 $ | 35 ms | ★★★★★ |
| GPT-4.1 | 8,00 $ | 80,00 $ | 55 ms | ★★★★☆ |
| Claude Sonnet 4.5 | 15,00 $ | 150,00 $ | 60 ms | ★★★★★ |
Pour un système d'analyse de logs 处理 10 millions de tokens par mois, DeepSeek V3.2 offre l'économie la plus significative avec seulement 4,20 $ mensuels. Cependant, pour les environnements de production exigeant une précision maximale sur la détection de fraude, Gemini 2.5 Flash représente le meilleur équilibre qualité-prix avec 25 $ mensuels et une latence de seulement 35 ms. HolySheep AI (S'inscrire ici) agrège toutes ces options avec un taux de change avantageux de ¥1=$1, soit une économie de 85% par rapport aux fournisseurs occidentaux.
Architecture du Système de Détection d'Anomalies
Mon expérience pratique de trois années dans l'intégration de systèmes de trading algorithmique m'a démontré que la détection efficace des patterns suspects repose sur trois piliers fondamentaux : l'ingestion temps réel des logs, l'analyse sémantique par IA, et la conservation auditée immutable. Le schéma suivant illustre l'architecture que nous allons implémenter.
Composants Principaux
- Collecteur de Logs : Capture les événements API en temps réel via webhooks et polling
- Normaliseur de Données : Standardise les formats de logs hétérogènes (Binance, Coinbase, Kraken)
- Moteur d'Analyse IA : Classification des transactions selon le risque
- Base de Données d'Audit : Stockage immutable avec horodatage certifié
- Tableau de Bord de Monitoring : Visualisation des alertes et statistiques
Implémentation Complète en Python
1. Configuration de l'Environnement et Client HolySheep
# Installation des dépendances
pip install requests pandas python-json-logger elasticsearch python-dateutil
configuration.py
import os
from dataclasses import dataclass
from typing import Optional
@dataclass
class HolySheepConfig:
"""Configuration du client HolySheep AI pour analyse de logs"""
api_key: str = "YOUR_HOLYSHEEP_API_KEY"
base_url: str = "https://api.holysheep.ai/v1"
model: str = "deepseek-v3.2" # Modèle économique pour volume élevé
# Limites de traitement
max_tokens_per_request: int = 8000
max_batch_size: int = 50 # Logs par lot
retry_attempts: int = 3
timeout_seconds: int = 30
class LogAnalyzerConfig:
"""Configuration du système d'analyse de logs"""
# Paramètres d'échange (configurez selon vos besoins)
EXCHANGE_API_KEY: Optional[str] = os.getenv("EXCHANGE_API_KEY")
EXCHANGE_SECRET: Optional[str] = os.getenv("EXCHANGE_SECRET")
EXCHANGE_NAME: str = "binance" # binance, coinbase, kraken
# Seuils de détection d'anomalies
RISK_THRESHOLD_HIGH: float = 0.85
RISK_THRESHOLD_MEDIUM: float = 0.60
RISK_THRESHOLD_LOW: float = 0.30
# Configuration d'audit
AUDIT_RETENTION_DAYS: int = 2555 # 7 ans pour conformité MiCA
AUDIT_BACKUP_ENABLED: bool = True
Initialisation du client
config = HolySheepConfig()
analyzer_config = LogAnalyzerConfig()
print(f"Client HolySheep configuré : {config.base_url}")
print(f"Modèle utilisé : {config.model}")2. Module d'Ingestion et Normalisation des Logs
# log_collector.py
import json
import time
import hashlib
from datetime import datetime, timezone
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, asdict
import requests
@dataclass
class NormalizedLog:
"""Structure normalisée pour tous les types de logs d'échange"""
log_id: str
timestamp: str
exchange: str
event_type: str
user_id: str
wallet_address: Optional[str]
transaction_amount: Optional[float]
transaction_currency: Optional[str]
action: str
status: str
ip_address: Optional[str]
user_agent: Optional[str]
request_path: str
response_code: int
latency_ms: float
raw_data: Dict[str, Any]
checksum: str
class LogCollector:
"""Collecteur et normaliseur de logs multi-échanges"""
def __init__(self, exchange_name: str, api_key: str, secret: str):
self.exchange = exchange_name
self.api_key = api_key
self.secret = secret
self.exchange_handlers = {
"binance": self._normalize_binance_log,
"coinbase": self._normalize_coinbase_log,
"kraken": self._normalize_kraken_log,
}
def _generate_log_id(self, data: Dict) -> str:
"""Génère un ID unique et déterministe basé sur le contenu"""
content = f"{data.get('timestamp', '')}{data.get('orderId', '')}{data.get('clientOrderId', '')}"
return hashlib.sha256(content.encode()).hexdigest()[:16]
def _compute_checksum(self, log: NormalizedLog) -> str:
"""Calcule un checksum pour intégrité des données d'audit"""
content = f"{log.log_id}{log.timestamp}{log.action}{log.status}{log.transaction_amount}"
return hashlib.sha256(content.encode()).hexdigest()
def _normalize_binance_log(self, raw_log: Dict) -> NormalizedLog:
"""Normalise les logs Binance au format standard"""
return NormalizedLog(
log_id=self._generate_log_id(raw_log),
timestamp=raw_log.get("eventTime", datetime.now(timezone.utc).isoformat()),
exchange="binance",
event_type=raw_log.get("eventType", "unknown"),
user_id=str(raw_log.get("userId", "")),
wallet_address=raw_log.get("walletAddress"),
transaction_amount=float(raw_log.get("amount", 0)),
transaction_currency=raw_log.get("symbol", ""),
action=raw_log.get("side", "UNKNOWN"),
status=raw_log.get("status", "UNKNOWN"),
ip_address=raw_log.get("ipAddress"),
user_agent=raw_log.get("userAgent"),
request_path=raw_log.get("endpoint", ""),
response_code=raw_log.get("responseCode", 0),
latency_ms=float(raw_log.get("latencyMs", 0)),
raw_data=raw_log,
checksum=""
)
def _normalize_coinbase_log(self, raw_log: Dict) -> NormalizedLog:
"""Normalise les logs Coinbase Advanced Trade"""
return NormalizedLog(
log_id=self._generate_log_id(raw_log),
timestamp=raw_log.get("time", datetime.now(timezone.utc).isoformat()),
exchange="coinbase",
event_type=raw_log.get("type", "unknown"),
user_id=str(raw_log.get("user_id", "")),
wallet_address=raw_log.get("profile_id"),
transaction_amount=float(raw_log.get("size", raw_log.get("funds", 0))),
transaction_currency=raw_log.get("product_id", ""),
action=raw_log.get("side", "UNKNOWN").upper(),
status=raw_log.get("status", "UNKNOWN"),
ip_address=raw_log.get("ip_address"),
user_agent=raw_log.get("user_agent"),
request_path=raw_log.get("api_url", ""),
response_code=raw_log.get("status_code", 0),
latency_ms=float(raw_log.get("response_time_ms", 0)),
raw_data=raw_log,
checksum=""
)
def _normalize_kraken_log(self, raw_log: Dict) -> NormalizedLog:
"""Normalise les logs Kraken"""
return NormalizedLog(
log_id=self._generate_log_id(raw_log),
timestamp=raw_log.get("timestamp", datetime.now(timezone.utc).isoformat()),
exchange="kraken",
event_type=raw_log.get("event", "unknown"),
user_id=str(raw_log.get("user_id", raw_log.get("docalias", ""))),
wallet_address=raw_log.get("descr", {}).get("pair"),
transaction_amount=float(raw_log.get("vol", raw_log.get("cost", 0))),
transaction_currency=raw_log.get("pair", ""),
action=raw_log.get("descr", {}).get("type", "UNKNOWN"),
status=raw_log.get("status", "UNKNOWN"),
ip_address=raw_log.get("ip", ""),
user_agent=None,
request_path=raw_log.get("path", ""),
response_code=raw_log.get("http_status", 200),
latency_ms=float(raw_log.get("execution_time_ms", 0)),
raw_data=raw_log,
checksum=""
)
def normalize(self, raw_logs: List[Dict]) -> List[NormalizedLog]:
"""Normalise un lot de logs selon l'échange source"""
handler = self.exchange_handlers.get(self.exchange, self._normalize_binance_log)
normalized = []
for raw_log in raw_logs:
try:
log = handler(raw_log)
log.checksum = self._compute_checksum(log)
normalized.append(log)
except Exception as e:
print(f"Erreur de normalisation: {e}")
continue
return normalized
Test du collecteur
collector = LogCollector(
exchange_name="binance",
api_key="test_key",
secret="test_secret"
)
test_logs = [
{
"eventTime": "2026-01-15T10:30:00Z",
"userId": 12345678,
"orderId": "ORD123456",
"amount": 1.5,
"symbol": "BTCUSDT",
"side": "BUY",
"status": "FILLED",
"ipAddress": "192.168.1.100",
"latencyMs": 45.2,
"responseCode": 200
}
]
normalized = collector.normalize(test_logs)
print(f"Logs normalisés : {len(normalized)} entrées")
print(f"Checksum : {normalized[0].checksum[:16]}...")3. Module d'Analyse IA avec HolySheep
# ai_analyzer.py
import json
import time
import requests
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass
from enum import Enum
class RiskLevel(Enum):
"""Niveaux de risque détectés"""
CRITICAL = "CRITICAL"
HIGH = "HIGH"
MEDIUM = "MEDIUM"
LOW = "LOW"
NORMAL = "NORMAL"
@dataclass
class AnalysisResult:
"""Résultat de l'analyse IA d'un log"""
log_id: str
risk_level: RiskLevel
risk_score: float
patterns_detected: List[str]
explanation: str
recommended_action: str
confidence: float
processing_time_ms: float
class AIAnalyzer:
"""Analyseur de patterns anormaux via API HolySheep AI"""
def __init__(self, api_key: str, base_url: str, model: str = "deepseek-v3.2"):
self.api_key = api_key
self.base_url = base_url
self.model = model
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
def _build_analysis_prompt(self, logs: List[Dict]) -> str:
"""Construit le prompt pour analyse de sécurité crypto"""
logs_json = json.dumps(logs[:10], indent=2) # Limite à 10 logs par requête
prompt = f"""Analyse ces logs de transactions de cryptomonnaie pour détecter des patterns suspects.
LOGS À ANALYSER (format JSON) :
{logs_json}
Pour chaque transaction, identifie les patterns d'anomalie suivants :
1. LAVAGE D'ARGENT : Petites transactions multiples vers différentes adresses
2. FRONTRUNNING : Ordres placés juste avant de grands ordres pour profiter du slippage
3. PUMP AND DUMP : Achats massifs followed by ventes rapides
4. WASH TRADING : Achats et ventes simultanés pour créer un faux volume
5. STRUCTURE FRAGMENTÉE : Échanges divisés pour éviter les seuils de déclaration
6. VULNÉRABILITÉ API : Requêtes anormales ou tentatives d'exploitation
7. COMPORTEMENT BOT : Patterns répétitifs caractéristiques d'automatisation malveillante
Réponds en JSON structuré avec ce format exact :
{{
"analysis": [
{{
"log_id": "ID du log",
"risk_level": "CRITICAL|HIGH|MEDIUM|LOW|NORMAL",
"risk_score": 0.0-1.0,
"patterns_detected": ["liste des patterns"],
"explanation": "explication détaillée",
"recommended_action": "action recommandée",
"confidence": 0.0-1.0
}}
]
}}
Analyse uniquement les données fournies. Sois précis et conservateur dans tes évaluations."""
return prompt
def analyze_batch(self, logs: List[Dict]) -> List[AnalysisResult]:
"""Analyse un lot de logs via l'API HolySheep"""
prompt = self._build_analysis_prompt(logs)
payload = {
"model": self.model,
"messages": [
{
"role": "system",
"content": "Tu es un expert en cybersécurité et détection de fraude sur les plateformes d'échange de cryptomonnaie."
},
{
"role": "user",
"content": prompt
}
],
"temperature": 0.1, # Température basse pour cohérence
"max_tokens": 4000,
"response_format": {"type": "json_object"}
}
start_time = time.time()
try:
response = self.session.post(
f"{self.base_url}/chat/completions",
json=payload,
timeout=30
)
response.raise_for_status()
result = response.json()
processing_time = (time.time() - start_time) * 1000
content = result["choices"][0]["message"]["content"]
analysis_data = json.loads(content)
results = []
for item in analysis_data.get("analysis", []):
results.append(AnalysisResult(
log_id=item.get("log_id", ""),
risk_level=RiskLevel(item.get("risk_level", "NORMAL")),
risk_score=float(item.get("risk_score", 0.0)),
patterns_detected=item.get("patterns_detected", []),
explanation=item.get("explanation", ""),
recommended_action=item.get("recommended_action", ""),
confidence=float(item.get("confidence", 0.0)),
processing_time_ms=processing_time
))
return results
except requests.exceptions.Timeout:
print(f"Timeout lors de l'analyse HolySheep (délai > 30s)")
return []
except requests.exceptions.RequestException as e:
print(f"Erreur API HolySheep : {e}")
return []
except json.JSONDecodeError as e:
print(f"Erreur parsing réponse JSON : {e}")
return []
def analyze_realtime(self, log: Dict) -> AnalysisResult:
"""Analyse un log unique en temps réel"""
results = self.analyze_batch([log])
return results[0] if results else None
Démonstration avec l'API HolySheep
analyzer = AIAnalyzer(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1",
model="deepseek-v3.2"
)
Logs de test avec patterns suspects
test_logs = [
{
"log_id": "a1b2c3d4e5f6",
"timestamp": "2026-01-15T14:23:11Z",
"user_id": "U88421",
"action": "SELL",
"amount": 0.0012,
"currency": "BTC",
"ip_address": "185.220.101.45",
"frequency_per_minute": 47,
"status": "FILLED"
},
{
"log_id": "b2c3d4e5f6g7",
"timestamp": "2026-01-15T14:23:12Z",
"user_id": "U88421",
"action": "SELL",
"amount": 0.0018,
"currency": "BTC",
"ip_address": "185.220.101.45",
"frequency_per_minute": 47,
"status": "FILLED"
},
{
"log_id": "c3d4e5f6g7h8",
"timestamp": "2026-01-15T14:23:15Z",
"user_id": "U88421",
"action": "SELL",
"amount": 0.0021,
"currency": "BTC",
"ip_address": "91.92.93.94",
"frequency_per_minute": 47,
"status": "FILLED"
}
]
Analyse (décommenter pour tester avec une vraie clé API)
results = analyzer.analyze_batch(test_logs)
for result in results:
print(f"[{result.risk_level.value}] {result.log_id}: {result.risk_score:.2%}")
print(f"Patterns: {', '.join(result.patterns_detected)}")
print()
print("Module AI Analyzer initialisé avec HolySheep API")4. Module de Conservation et Audit Immuable
# audit_logger.py
import json
import sqlite3
import hashlib
from datetime import datetime, timezone, timedelta
from typing import List, Dict, Any, Optional
from pathlib import Path
from dataclasses import asdict
import cryptography
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.backends import default_backend
class AuditLogger:
"""Système de conservation auditée immutable pour conformité réglementaire"""
def __init__(self, db_path: str = "audit_logs.db"):
self.db_path = db_path
self.conn = None
self._init_database()
self._generate_signing_keys()
def _init_database(self):
"""Initialise la base de données SQLite avec schéma audité"""
self.conn = sqlite3.connect(self.db_path, check_same_thread=False)
self.conn.row_factory = sqlite3.Row
cursor = self.conn.cursor()
# Table principale des audits
cursor.execute("""
CREATE TABLE IF NOT EXISTS audit_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
log_id TEXT UNIQUE NOT NULL,
timestamp TEXT NOT NULL,
exchange TEXT NOT NULL,
event_data TEXT NOT NULL,
analysis_result TEXT,
risk_level TEXT,
risk_score REAL,
checksum_chain TEXT,
previous_hash TEXT,
current_hash TEXT,
signature TEXT,
created_at TEXT DEFAULT CURRENT_TIMESTAMP
)
""")
# Index pour performances de recherche
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_timestamp
ON audit_logs(timestamp)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_risk_level
ON audit_logs(risk_level)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_exchange
ON audit_logs(exchange)
""")
# Table de métadonnées d'intégrité
cursor.execute("""
CREATE TABLE IF NOT EXISTS integrity_metadata (
id INTEGER PRIMARY KEY,
last_hash TEXT,
total_records INTEGER DEFAULT 0,
last_update TEXT
)
""")
self.conn.commit()
def _generate_signing_keys(self):
"""Génère une paire de clés ECDSA pour signature des audits"""
self.private_key = ec.generate_private_key(ec.SECP256R1(), default_backend())
self.public_key = self.private_key.public_key()
def _compute_hash(self, data: Dict) -> str:
"""Calcule un hash SHA-256 des données"""
json_str = json.dumps(data, sort_keys=True, default=str)
return hashlib.sha256(json_str.encode()).hexdigest()
def _get_previous_hash(self) -> Optional[str]:
"""Récupère le hash du dernier enregistrement"""
cursor = self.conn.cursor()
cursor.execute("SELECT current_hash FROM audit_logs ORDER BY id DESC LIMIT 1")
result = cursor.fetchone()
return result["current_hash"] if result else None
def _sign_data(self, data: Dict) -> str:
"""Signe les données avec la clé privée ECDSA"""
data_hash = self._compute_hash(data)
signature = self.private_key.sign(
data_hash.encode(),
ec.ECDSA(hashes.SHA256())
)
return signature.hex()
def log_event(self, log_entry: Dict, analysis_result: Optional[Dict] = None) -> str:
"""Enregistre un événement dans la chaîne d'audit"""
timestamp = datetime.now(timezone.utc).isoformat()
log_id = log_entry.get("log_id", self._compute_hash(log_entry)[:16])
previous_hash = self._get_previous_hash()
# Données à hasher pour la chaîne
chain_data = {
"log_id": log_id,
"timestamp": timestamp,
"event_data": log_entry,
"previous_hash": previous_hash
}
current_hash = self._compute_hash(chain_data)
signature = self._sign_data(chain_data)
checksum_chain = f"{previous_hash or 'GENESIS'}->{current_hash[:16]}"
cursor = self.conn.cursor()
cursor.execute("""
INSERT OR REPLACE INTO audit_logs
(log_id, timestamp, exchange, event_data, analysis_result,
risk_level, risk_score, checksum_chain, previous_hash, current_hash, signature)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
log_id,
timestamp,
log_entry.get("exchange", "unknown"),
json.dumps(log_entry),
json.dumps(analysis_result) if analysis_result else None,
analysis_result.get("risk_level") if analysis_result else None,
analysis_result.get("risk_score") if analysis_result else None,
checksum_chain,
previous_hash,
current_hash,
signature
))
# Mise à jour des métadonnées
cursor.execute("""
INSERT OR REPLACE INTO integrity_metadata
(id, last_hash, total_records, last_update)
VALUES (1, ?, (SELECT COUNT(*) FROM audit_logs), ?)
""", (current_hash, timestamp))
self.conn.commit()
return log_id
def verify_chain_integrity(self) -> Dict[str, Any]:
"""Vérifie l'intégrité de la chaîne d'audit"""
cursor = self.conn.cursor()
cursor.execute("SELECT * FROM audit_logs ORDER BY id")
logs = cursor.fetchall()
valid = True
errors = []
for i, log in enumerate(logs):
expected_previous = logs[i-1]["current_hash"] if i > 0 else None
if log["previous_hash"] != expected_previous:
valid = False
errors.append(f"Chaîne brisée à l'index {i}: hash précédent invalide")
# Vérification de signature
chain_data = {
"log_id": log["log_id"],
"timestamp": log["timestamp"],
"event_data": json.loads(log["event_data"]),
"previous_hash": log["previous_hash"]
}
try:
signature_bytes = bytes.fromhex(log["signature"])
data_hash = self._compute_hash(chain_data).encode()
self.public_key.verify(signature_bytes, data_hash, ec.ECDSA(hashes.SHA256()))
except Exception as e:
valid = False
errors.append(f"Signature invalide pour {log['log_id']}: {e}")
return {
"valid": valid,
"total_records": len(logs),
"errors": errors
}
def query_logs(self,
start_date: Optional[str] = None,
end_date: Optional[str] = None,
risk_level: Optional[str] = None,
exchange: Optional[str] = None,
limit: int = 100) -> List[Dict]:
"""Interroge les logs d'audit avec filtres"""
query = "SELECT * FROM audit_logs WHERE 1=1"
params = []
if start_date:
query += " AND timestamp >= ?"
params.append(start_date)
if end_date:
query += " AND timestamp <= ?"
params.append(end_date)
if risk_level:
query += " AND risk_level = ?"
params.append(risk_level)
if exchange:
query += " AND exchange = ?"
params.append(exchange)
query += " ORDER BY timestamp DESC LIMIT ?"
params.append(limit)
cursor = self.conn.cursor()
cursor.execute(query, params)
return [dict(row) for row in cursor.fetchall()]
def export_for_compliance(self, output_path: str, days: int = 30):
"""Exporte les logs pour conformité réglementaire (MiCA, AML)"""
cutoff = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
logs = self.query_logs(start_date=cutoff)
export_data = {
"export_date": datetime.now(timezone.utc).isoformat(),
"period_start": cutoff,
"period_end": datetime.now(timezone.utc).isoformat(),
"total_records": len(logs),
"integrity_verified": self.verify_chain_integrity()["valid"],
"logs": logs
}
with open(output_path, "w") as f:
json.dump(export_data, f, indent=2, default=str)
return output_data
Démonstration
audit_logger = AuditLogger("crypto_audit.db")
Enregistrement d'un événement
test_event = {
"log_id": "test123",
"timestamp": datetime.now(timezone.utc).isoformat(),
"exchange": "binance",
"user_id": "U12345",
"action": "SELL",
"amount": 0.5,
"currency": "ETH"
}
log_id = audit_logger.log_event(test_event, {
"risk_level": "HIGH",
"risk_score": 0.82,
"patterns_detected": ["RAPID_SUCCESSION", "MULTI_ADDRESS"]
})
print(f"Événement enregistré : {log_id}")
Vérification d'intégrité
integrity = audit_logger.verify_chain_integrity()
print(f"Intégrité de la chaîne : {'VALIDE' if integrity['valid'] else 'INVALIDE'}")
Export pour conformité
export_path = audit_logger.export_for_compliance("compliance_export.json", days=30)
print(f"Export généré : {export_path}")Pipeline Complet d'Analyse Temps Réel
# main.py - Pipeline complet d'analyse de logs
import asyncio
import json
import time
from datetime import datetime, timezone
from typing import List, Dict
from log_collector import LogCollector, NormalizedLog
from ai_analyzer import AIAnalyzer, AnalysisResult, RiskLevel
from audit_logger import AuditLogger
class CryptoSecurityPipeline:
"""Pipeline complet de détection d'anomalies et audit"""
def __init__(self,
holy_sheep_api_key: str,
exchange_config: Dict):
# Initialisation des composants
self.collector = LogCollector(
exchange_name=exchange_config["name"],
api_key=exchange_config["api_key"],
secret=exchange_config["secret"]
)
self.analyzer = AIAnalyzer(
api_key=holy_sheep_api_key,
base_url="https://api.holysheep.ai/v1",
model="deepseek-v3.2"
)
self.audit_logger = AuditLogger()
# Statistiques
self.stats = {
"total_processed": 0,
"critical_alerts": 0,
"high_alerts": 0,
"medium_alerts": 0,
"low_alerts": 0,
"normal": 0,
"processing_time_avg_ms": 0
}
def process_logs(self, raw_logs: List[Dict]) -> Dict:
"""Traitement complet d'un lot de logs"""
start_time = time.time()
# Étape 1 : Normalisation
normalized_logs = self.collector.normalize(raw_logs)
# Étape 2 : Analyse IA
log_dicts = [asdict(log) for log in normalized_logs]
analysis_results = self.analyzer.analyze_batch(log_dicts)
# Étape 3 : Audit et alerte
for i, log in enumerate(normalized_logs):
analysis = analysis_results[i] if i < len(analysis_results) else None
# Enregistrement dans l'audit
self.audit_logger.log_event(
log_entry=asdict(log),
analysis_result=asdict(analysis) if analysis else None
)
# Mise à jour des statistiques
self._update_stats(analysis)
# Génération d'alertes pour risques élevés
if analysis and analysis.risk_level in [RiskLevel.CRITICAL, RiskLevel.HIGH]:
self._generate_alert(log, analysis)
processing_time = (time.time() - start_time) * 1000
return {
"logs_processed": len(normalized_logs),
"analysis_results": len(analysis_results),
"processing_time_ms": processing_time,
"stats": self.stats
}
def _update_stats(self, analysis: AnalysisResult):
"""Met à jour les statistiques de traitement"""
if not analysis:
return
self.stats["total_processed"] += 1
risk_counts = {
RiskLevel.CRITICAL: "critical_alerts",
RiskLevel.HIGH: "high_alerts",
RiskLevel.MEDIUM: "medium_alerts",
RiskLevel.LOW: "low_alerts",
RiskLevel.NORMAL: "normal"
}
stat_key = risk_counts.get(analysis.risk_level, "normal")
self.stats[stat_key] += 1
def _generate_alert(self, log: NormalizedLog, analysis: AnalysisResult):
"""Génère une alerte pour les transactions à risque"""
alert = {
"alert_id": f"ALERT-{int(time.time())}-{log.log_id[:8]}",
"timestamp": datetime.now(timezone.utc).isoformat(),
"severity": analysis.risk_level.value,
"log_id": log.log_id,
"user_id": log.user_id,
"exchange": log.exchange,
"transaction": {
"amount": log.transaction_amount,
"currency": log.transaction_currency,
"action": log.action
},
"risk_score": analysis.risk_score,
"patterns": analysis.patterns_detected,
"explanation": analysis.explanation,
"recommendation": analysis.recommended_action,
"confidence": analysis.confidence
}
# Log de l'alerte (remplacer par webhook/SMS/email en production)
print(f"🚨 ALERTE {alert['severity']}: {alert['alert_id']}")
print(f" Utilisateur: {alert['user_id']}")
print(f" Transaction: {alert['transaction']}")
print(f" Patterns: {', '.join(alert['patterns'])}")
print(f" Recommandation: {alert['recommendation']}")
print()
return alert
def get_dashboard_data(self) -> Dict:
"""Génère les données pour le tableau de bord"""
recent_alerts = self.audit_logger.query_logs(
risk_level="HIGH",
limit=50
)
return {
"timestamp": datetime.now(timezone.utc).iso