#!/usr/bin/env python3 """ Correlation Analyzer - Analyse des corrélations inter-cryptos Détecte les rotations sectorielles et dépendances BTC """ import numpy as np from typing import Dict, List, Optional, Tuple import logging from collections import defaultdict logger = logging.getLogger("CorrelationAnalyzer") class CorrelationAnalyzer: """Analyse les corrélations entre cryptos et avec BTC""" def __init__(self): # Cache des prix récents pour calcul corrélations self.price_history = defaultdict(list) # {symbol: [prices]} self.max_history = 100 # Garder 100 dernières valeurs # Secteurs crypto (classification simplifiée) self.sectors = { 'BTC': ['BTC'], 'L1': ['ETH', 'SOL', 'ADA', 'AVAX', 'DOT', 'ATOM', 'NEAR', 'ALGO'], 'L2': ['MATIC', 'ARB', 'OP', 'IMX'], 'DEFI': ['UNI', 'AAVE', 'SNX', 'CRV', 'COMP', 'MKR', 'SUSHI'], 'EXCHANGE': ['BNB', 'CRO', 'FTT', 'OKB'], 'GAMING': ['AXS', 'SAND', 'MANA', 'ENJ', 'GALA'], 'ORACLE': ['LINK', 'BAND', 'TRB'], 'STORAGE': ['FIL', 'AR', 'STORJ'], 'PRIVACY': ['XMR', 'ZEC', 'DASH'], 'MEME': ['DOGE', 'SHIB', 'PEPE', 'FLOKI'] } # Mapping inverse: symbol -> sector self.symbol_to_sector = {} for sector, symbols in self.sectors.items(): for symbol in symbols: self.symbol_to_sector[symbol] = sector # Corrélations pré-calculées (à mettre à jour périodiquement) self.correlations = {} # {(symbol1, symbol2): correlation} self.btc_correlations = {} # {symbol: correlation_with_btc} # Secteurs en rotation (détection dynamique) self.active_sectors = [] logger.info("✅ Correlation Analyzer initialisé") def update_price(self, symbol: str, price: float): """ Met à jour l'historique de prix pour un symbole Args: symbol: Symbole (ex: 'BTC', 'ETH') price: Prix actuel """ # Nettoyer le symbole (enlever /USDT) clean_symbol = symbol.replace('/USDT', '').replace('USDT', '') # Ajouter au cache self.price_history[clean_symbol].append(price) # Limiter la taille if len(self.price_history[clean_symbol]) > self.max_history: self.price_history[clean_symbol] = self.price_history[clean_symbol][-self.max_history:] def calculate_correlation(self, symbol1: str, symbol2: str, period: int = 30) -> float: """ Calcule la corrélation entre 2 symboles Args: symbol1: Premier symbole symbol2: Deuxième symbole period: Nombre de points pour le calcul Returns: Corrélation de Pearson (-1 à +1) """ clean_symbol1 = symbol1.replace('/USDT', '').replace('USDT', '') clean_symbol2 = symbol2.replace('/USDT', '').replace('USDT', '') # Vérifier qu'on a assez de données if clean_symbol1 not in self.price_history or clean_symbol2 not in self.price_history: return 0.0 prices1 = self.price_history[clean_symbol1] prices2 = self.price_history[clean_symbol2] if len(prices1) < period or len(prices2) < period: return 0.0 # Prendre les dernières valeurs p1 = np.array(prices1[-period:]) p2 = np.array(prices2[-period:]) # Calculer les rendements returns1 = np.diff(p1) / p1[:-1] returns2 = np.diff(p2) / p2[:-1] # Corrélation de Pearson if len(returns1) < 2 or len(returns2) < 2: return 0.0 correlation = np.corrcoef(returns1, returns2)[0, 1] # Gérer les NaN if np.isnan(correlation): return 0.0 return correlation def get_btc_correlation(self, symbol: str, period: int = 30) -> float: """ Retourne la corrélation avec BTC Args: symbol: Symbole à analyser period: Période de calcul Returns: Corrélation avec BTC (-1 à +1) """ return self.calculate_correlation(symbol, 'BTC', period) def analyze_correlation_strength(self, symbol: str) -> Dict: """ Analyse la force de corrélation d'un symbole avec BTC et son secteur Returns: Dict avec analyse de corrélation """ clean_symbol = symbol.replace('/USDT', '').replace('USDT', '') analysis = { 'symbol': clean_symbol, 'btc_correlation': 0.0, 'btc_correlation_strength': 'UNKNOWN', 'sector': 'UNKNOWN', 'sector_correlation': 0.0, 'is_sector_leader': False, 'correlation_bonus': 0.0, 'recommendations': [] } # Corrélation avec BTC btc_corr = self.get_btc_correlation(clean_symbol, 30) analysis['btc_correlation'] = round(btc_corr, 3) # Classification de la corrélation if btc_corr > 0.7: analysis['btc_correlation_strength'] = 'TRÈS FORTE' analysis['recommendations'].append("⚠️ Très corrélé à BTC - suivre BTC de près") elif btc_corr > 0.5: analysis['btc_correlation_strength'] = 'FORTE' analysis['recommendations'].append("📊 Forte corrélation BTC") elif btc_corr > 0.3: analysis['btc_correlation_strength'] = 'MOYENNE' analysis['correlation_bonus'] += 5 # Bonus pour décorrélation partielle analysis['recommendations'].append("✅ Corrélation modérée - diversification OK") elif btc_corr > 0: analysis['btc_correlation_strength'] = 'FAIBLE' analysis['correlation_bonus'] += 10 # Bon pour diversification analysis['recommendations'].append("✅ Faible corrélation - excellente diversification") else: analysis['btc_correlation_strength'] = 'NÉGATIVE' analysis['correlation_bonus'] += 15 # Très bon pour hedging analysis['recommendations'].append("🔥 Corrélation négative - potentiel de hedging") # Identifier le secteur sector = self.symbol_to_sector.get(clean_symbol, 'UNKNOWN') analysis['sector'] = sector # Analyser la performance du secteur if sector != 'UNKNOWN' and sector in self.sectors: sector_symbols = self.sectors[sector] sector_perfs = [] for sec_symbol in sector_symbols: if sec_symbol in self.price_history and len(self.price_history[sec_symbol]) >= 10: prices = self.price_history[sec_symbol] perf = (prices[-1] - prices[-10]) / prices[-10] * 100 sector_perfs.append(perf) if sector_perfs: avg_sector_perf = np.mean(sector_perfs) # Ce symbole performe-t-il mieux que son secteur ? if clean_symbol in self.price_history and len(self.price_history[clean_symbol]) >= 10: prices = self.price_history[clean_symbol] symbol_perf = (prices[-1] - prices[-10]) / prices[-10] * 100 if symbol_perf > avg_sector_perf * 1.2: analysis['is_sector_leader'] = True analysis['correlation_bonus'] += 10 analysis['recommendations'].append(f"🚀 Leader du secteur {sector}") analysis['symbol_vs_sector'] = round(symbol_perf - avg_sector_perf, 2) return analysis def detect_sector_rotation(self) -> List[str]: """ Détecte les secteurs en rotation (performance récente forte) Returns: Liste des secteurs actifs """ sector_performance = {} for sector, symbols in self.sectors.items(): if sector == 'BTC': continue perfs = [] for symbol in symbols: if symbol in self.price_history and len(self.price_history[symbol]) >= 20: prices = self.price_history[symbol] # Performance sur 20 dernières valeurs perf = (prices[-1] - prices[-20]) / prices[-20] * 100 perfs.append(perf) if perfs: avg_perf = np.mean(perfs) sector_performance[sector] = avg_perf # Trier par performance sorted_sectors = sorted(sector_performance.items(), key=lambda x: x[1], reverse=True) # Secteurs en rotation = top 3 avec perf > 0 active_sectors = [sector for sector, perf in sorted_sectors[:3] if perf > 0] self.active_sectors = active_sectors if active_sectors: logger.info(f"🔄 Secteurs en rotation: {', '.join(active_sectors)}") return active_sectors def is_in_active_sector(self, symbol: str) -> Tuple[bool, float]: """ Vérifie si le symbole est dans un secteur actif Returns: (is_active, bonus_score) """ clean_symbol = symbol.replace('/USDT', '').replace('USDT', '') sector = self.symbol_to_sector.get(clean_symbol, 'UNKNOWN') if sector in self.active_sectors: # Bonus selon la position dans le secteur if self.active_sectors.index(sector) == 0: return True, 15 # Secteur #1 elif self.active_sectors.index(sector) == 1: return True, 10 # Secteur #2 else: return True, 5 # Secteur #3 return False, 0 def get_correlation_matrix(self, symbols: List[str], period: int = 30) -> Dict: """ Calcule une matrice de corrélation pour plusieurs symboles Args: symbols: Liste de symboles period: Période de calcul Returns: Dict avec matrice de corrélation """ matrix = {} for sym1 in symbols: matrix[sym1] = {} for sym2 in symbols: if sym1 == sym2: matrix[sym1][sym2] = 1.0 else: corr = self.calculate_correlation(sym1, sym2, period) matrix[sym1][sym2] = round(corr, 3) return matrix def get_diversification_score(self, portfolio_symbols: List[str]) -> Dict: """ Évalue la diversification d'un portfolio Args: portfolio_symbols: Liste des symboles du portfolio Returns: Dict avec score de diversification """ if len(portfolio_symbols) < 2: return {'diversification_score': 100, 'quality': 'N/A'} # Calculer corrélations moyennes correlations = [] for i, sym1 in enumerate(portfolio_symbols): for sym2 in portfolio_symbols[i+1:]: corr = self.calculate_correlation(sym1, sym2, 30) correlations.append(abs(corr)) if not correlations: return {'diversification_score': 100, 'quality': 'N/A'} avg_correlation = np.mean(correlations) # Score de diversification (inversement proportionnel à la corrélation) # 0 corrélation = 100 score, 1 corrélation = 0 score diversification_score = (1 - avg_correlation) * 100 # Qualité if diversification_score > 70: quality = 'EXCELLENTE' elif diversification_score > 50: quality = 'BONNE' elif diversification_score > 30: quality = 'MOYENNE' else: quality = 'FAIBLE' return { 'diversification_score': round(diversification_score, 1), 'avg_correlation': round(avg_correlation, 3), 'quality': quality, 'num_positions': len(portfolio_symbols) } def recommend_for_diversification(self, current_symbols: List[str], candidates: List[str], top_n: int = 5) -> List[Tuple[str, float]]: """ Recommande des symboles pour améliorer la diversification Args: current_symbols: Symboles actuels du portfolio candidates: Symboles candidats top_n: Nombre de recommandations Returns: Liste de (symbol, diversification_score) """ recommendations = [] for candidate in candidates: if candidate in current_symbols: continue # Calculer corrélation moyenne avec le portfolio actuel correlations = [] for current in current_symbols: corr = self.calculate_correlation(candidate, current, 30) correlations.append(abs(corr)) if correlations: avg_corr = np.mean(correlations) # Score de diversification (plus bas = meilleur) div_score = 1 - avg_corr recommendations.append((candidate, round(div_score, 3))) # Trier par score décroissant recommendations.sort(key=lambda x: x[1], reverse=True) return recommendations[:top_n] # Instance globale _correlation_analyzer = None def get_correlation_analyzer() -> CorrelationAnalyzer: """Retourne l'instance globale du correlation analyzer""" global _correlation_analyzer if _correlation_analyzer is None: _correlation_analyzer = CorrelationAnalyzer() return _correlation_analyzer