#!/usr/bin/env python3 """ Volume Profile Analyzer - Analyse avancée des volumes Détecte accumulation, distribution, breakouts et zones de liquidité """ import numpy as np from typing import Dict, List, Optional, Tuple import logging logger = logging.getLogger("VolumeProfileAnalyzer") class VolumeProfileAnalyzer: """Analyseur avancé de profil de volume""" def __init__(self): self.volume_history = {} # Cache des moyennes de volume par symbole logger.info("✅ Volume Profile Analyzer initialisé") def analyze_volume_profile(self, symbol: str, prices: List[float], volumes: List[float]) -> Dict: """ Analyse complète du profil de volume Returns: Dict avec détection accumulation/distribution, breakouts, anomalies """ if not volumes or len(volumes) < 20: return {'success': False, 'reason': 'insufficient_data'} try: volumes_array = np.array(volumes[-50:]) # Dernières 50 bougies prices_array = np.array(prices[-50:]) # Calculer moyennes avg_volume_20 = np.mean(volumes_array[-20:]) avg_volume_50 = np.mean(volumes_array) current_volume = volumes_array[-1] # Stocker historique self.volume_history[symbol] = { 'avg_20': avg_volume_20, 'avg_50': avg_volume_50, 'timestamp': np.datetime64('now') } result = { 'success': True, 'current_volume': current_volume, 'avg_volume_20': avg_volume_20, 'avg_volume_50': avg_volume_50, 'volume_ratio': current_volume / avg_volume_20 if avg_volume_20 > 0 else 1, 'score_bonus': 0, 'signals': [] } # 1. DÉTECTION VOLUME ANORMAL (Breakout potentiel) if current_volume > avg_volume_20 * 3: result['signals'].append('BREAKOUT_VOLUME') result['score_bonus'] += 25 logger.debug(f"🔥 {symbol}: BREAKOUT VOLUME détecté (×{current_volume/avg_volume_20:.1f})") elif current_volume > avg_volume_20 * 2: result['signals'].append('HIGH_VOLUME') result['score_bonus'] += 15 # 2. DÉTECTION ACCUMULATION (volume élevé + prix stable) accumulation_score = self._detect_accumulation(prices_array, volumes_array) if accumulation_score > 0: result['accumulation_score'] = accumulation_score result['signals'].append('ACCUMULATION') result['score_bonus'] += accumulation_score logger.debug(f"💎 {symbol}: ACCUMULATION détectée (score={accumulation_score})") # 3. DÉTECTION DISTRIBUTION (volume élevé + prix en baisse) distribution_score = self._detect_distribution(prices_array, volumes_array) if distribution_score > 0: result['distribution_score'] = distribution_score result['signals'].append('DISTRIBUTION') result['score_bonus'] -= distribution_score # Négatif car baissier logger.debug(f"⚠️ {symbol}: DISTRIBUTION détectée (score={distribution_score})") # 4. VOLUME CLUSTERS (Support/Résistance par volume) volume_clusters = self._find_volume_clusters(prices_array, volumes_array) result['volume_clusters'] = volume_clusters # 5. VOLUME TREND (Volume en hausse/baisse) volume_trend = self._analyze_volume_trend(volumes_array) result['volume_trend'] = volume_trend if volume_trend == 'INCREASING': result['score_bonus'] += 5 elif volume_trend == 'DECREASING': result['score_bonus'] -= 5 # 6. ON-BALANCE VOLUME (OBV) obv = self._calculate_obv(prices_array, volumes_array) result['obv'] = obv result['obv_trend'] = 'BULLISH' if obv[-1] > obv[-10] else 'BEARISH' if result['obv_trend'] == 'BULLISH': result['score_bonus'] += 10 return result except Exception as e: logger.error(f"Erreur analyse volume profile {symbol}: {e}") return {'success': False, 'reason': str(e)} def _detect_accumulation(self, prices: np.ndarray, volumes: np.ndarray) -> float: """ Détecte une phase d'accumulation (volume élevé + prix stable) Returns: Score 0-20 selon intensité de l'accumulation """ if len(prices) < 20 or len(volumes) < 20: return 0 # Volume moyen des 10 dernières bougies vs 20 précédentes recent_volume = np.mean(volumes[-10:]) older_volume = np.mean(volumes[-20:-10]) # Variation de prix (stabilité) price_volatility = np.std(prices[-10:]) / np.mean(prices[-10:]) # Accumulation = volume en hausse + prix stable volume_increase = recent_volume / older_volume if older_volume > 0 else 1 if volume_increase > 1.3 and price_volatility < 0.015: # Volume +30%, volatilité <1.5% return min(20, int((volume_increase - 1) * 30)) return 0 def _detect_distribution(self, prices: np.ndarray, volumes: np.ndarray) -> float: """ Détecte une phase de distribution (volume élevé + prix en baisse) Returns: Score 0-20 selon intensité de la distribution """ if len(prices) < 10 or len(volumes) < 10: return 0 recent_volume = np.mean(volumes[-10:]) avg_volume = np.mean(volumes[-20:]) # Tendance prix (baissière?) price_change = (prices[-1] - prices[-10]) / prices[-10] # Distribution = volume élevé + prix en baisse if recent_volume > avg_volume * 1.3 and price_change < -0.02: # -2% return min(20, int(abs(price_change) * 500)) return 0 def _find_volume_clusters(self, prices: np.ndarray, volumes: np.ndarray) -> List[Dict]: """ Identifie les zones de prix avec volume important (support/résistance) Returns: Liste de clusters {price_level, volume_total, type} """ if len(prices) < 20: return [] # Créer bins de prix price_min = np.min(prices) price_max = np.max(prices) n_bins = 10 bins = np.linspace(price_min, price_max, n_bins) clusters = [] for i in range(len(bins) - 1): # Trouver volumes dans ce bin mask = (prices >= bins[i]) & (prices < bins[i+1]) volume_in_bin = np.sum(volumes[mask]) if volume_in_bin > 0: clusters.append({ 'price_level': (bins[i] + bins[i+1]) / 2, 'volume_total': volume_in_bin, 'price_min': bins[i], 'price_max': bins[i+1] }) # Trier par volume décroissant clusters.sort(key=lambda x: x['volume_total'], reverse=True) return clusters[:5] # Top 5 clusters def _analyze_volume_trend(self, volumes: np.ndarray) -> str: """ Analyse la tendance du volume (hausse, baisse, stable) Returns: 'INCREASING', 'DECREASING', 'STABLE' """ if len(volumes) < 20: return 'STABLE' recent_avg = np.mean(volumes[-10:]) older_avg = np.mean(volumes[-20:-10]) change = (recent_avg - older_avg) / older_avg if older_avg > 0 else 0 if change > 0.15: # +15% return 'INCREASING' elif change < -0.15: # -15% return 'DECREASING' else: return 'STABLE' def _calculate_obv(self, prices: np.ndarray, volumes: np.ndarray) -> np.ndarray: """ Calcule l'On-Balance Volume (OBV) Indicateur cumulatif : +volume si prix monte, -volume si prix baisse """ obv = np.zeros(len(prices)) obv[0] = volumes[0] for i in range(1, len(prices)): if prices[i] > prices[i-1]: obv[i] = obv[i-1] + volumes[i] elif prices[i] < prices[i-1]: obv[i] = obv[i-1] - volumes[i] else: obv[i] = obv[i-1] return obv def get_volume_quality_score(self, volume_result: Dict) -> Tuple[float, str]: """ Évalue la qualité du signal basé sur le volume Returns: (score 0-100, explication) """ if not volume_result.get('success'): return 50, "Données volume insuffisantes" score = 50 # Neutre par défaut reasons = [] # Bonus pour volume élevé volume_ratio = volume_result.get('volume_ratio', 1) if volume_ratio > 2: score += 20 reasons.append(f"Volume élevé (×{volume_ratio:.1f})") elif volume_ratio > 1.5: score += 10 # Accumulation = très positif if 'ACCUMULATION' in volume_result.get('signals', []): score += 15 reasons.append("Phase d'accumulation") # Distribution = négatif if 'DISTRIBUTION' in volume_result.get('signals', []): score -= 20 reasons.append("Phase de distribution (⚠️)") # OBV bullish if volume_result.get('obv_trend') == 'BULLISH': score += 10 reasons.append("OBV haussier") score = max(0, min(100, score)) explanation = " | ".join(reasons) if reasons else "Volume normal" return score, explanation # Instance globale _volume_analyzer = None def get_volume_analyzer() -> VolumeProfileAnalyzer: """Retourne l'instance globale de l'analyseur de volume""" global _volume_analyzer if _volume_analyzer is None: _volume_analyzer = VolumeProfileAnalyzer() return _volume_analyzer