""" ═══════════════════════════════════════════════════════════════════════════════ 🔄 SMART ROTATION - Système de Rotation Intelligente du Portefeuille ═══════════════════════════════════════════════════════════════════════════════ Ce module analyse les positions actuelles et les opportunités d'achat pour effectuer des rotations intelligentes: - Vendre une position dont le cycle de gain est terminé - Acheter immédiatement une valeur avec un meilleur potentiel Stratégie: 1. Calculer le "score de fin de cycle" pour chaque position 2. Calculer le "score d'opportunité" pour chaque candidat d'achat 3. Si une rotation offre un gain potentiel significatif → exécuter Auteur: Trading IA Bot Date: Décembre 2024 ═══════════════════════════════════════════════════════════════════════════════ """ import json import logging import os from datetime import datetime, timedelta from typing import Dict, List, Tuple, Optional import numpy as np logger = logging.getLogger(__name__) # Configuration de la rotation - RE-OPTIMISÉE le 28/12/2025 # Analyse: 54% des trades étaient des rotations avec -2.09 USDT de perte! # Solution: Paramètres beaucoup plus conservateurs pour éviter le sur-trading # Importer les paramètres depuis config.py si disponibles try: import config ROTATION_CONFIG = { 'min_cycle_end_score': getattr(config, 'ROTATION_MIN_CYCLE_END_SCORE', 20), 'min_opportunity_score': getattr(config, 'ROTATION_MIN_OPPORTUNITY_SCORE', 50), 'min_score_advantage': getattr(config, 'ROTATION_MIN_SCORE_ADVANTAGE', 5), 'min_profit_for_rotation': getattr(config, 'ROTATION_MIN_PROFIT', -0.5), 'min_hold_time_minutes': getattr(config, 'ROTATION_MIN_HOLD_TIME', 10), 'rotation_cooldown_minutes': getattr(config, 'ROTATION_COOLDOWN', 30), 'max_rotations_per_hour': getattr(config, 'ROTATION_MAX_PER_HOUR', 4), 'enabled': getattr(config, 'ENABLE_SMART_ROTATION', True) } logger.info(f"🔄 Config rotation chargée depuis config.py") except ImportError: ROTATION_CONFIG = { # Score minimum pour considérer un cycle comme terminé # RÉDUIT à 20 - Permettre rotations des positions en légère perte 'min_cycle_end_score': 20, # Score minimum pour qu'une opportunité soit valide # RÉDUIT à 55 - Accepter les bonnes opportunités 'min_opportunity_score': 55, # Différence minimum de score pour justifier une rotation # RÉDUIT à 15 - Plus flexible 'min_score_advantage': 15, # Profit minimum à sécuriser avant rotation (%) # ÉLARGI à -0.5% - Permettre de sortir des positions en légère perte 'min_profit_for_rotation': -0.5, # Temps minimum en position avant rotation (minutes) # RÉDUIT à 10 - Plus réactif 'min_hold_time_minutes': 10, # Cooldown entre rotations pour le même symbole (minutes) # RÉDUIT à 30 - Plus réactif 'rotation_cooldown_minutes': 30, # Maximum de rotations par heure # AUGMENTÉ à 4 - Permettre plus de rotations 'max_rotations_per_hour': 4, # Activer le mode rotation 'enabled': True } class SmartRotation: """ Gestionnaire de rotation intelligente du portefeuille. """ def __init__(self, config: Dict = None): self.config = config or ROTATION_CONFIG.copy() self.rotation_history = [] self.last_rotations = {} # {symbol: timestamp} self.rotations_this_hour = 0 self.hour_start = datetime.now() self._prices_cache = {} self._prices_cache_time = None logger.info("🔄 Smart Rotation initialisé") def get_current_prices(self, symbols: List[str] = None) -> Dict[str, float]: """ Récupère les prix actuels depuis l'API Binance (avec cache 30s). """ import time # Cache de 30 secondes if self._prices_cache_time and (time.time() - self._prices_cache_time) < 30: return self._prices_cache try: import requests # Utiliser l'API production Binance (prix publics, pas besoin de clé API) # Le testnet n'a pas toujours tous les prix base_url = "https://api.binance.com" response = requests.get(f"{base_url}/api/v3/ticker/price", timeout=10) if response.status_code == 200: prices = {item['symbol']: float(item['price']) for item in response.json()} self._prices_cache = prices self._prices_cache_time = time.time() logger.info(f"🔄 Prix récupérés: {len(prices)} symboles") return prices else: logger.warning(f"Erreur API prix: status={response.status_code}") except Exception as e: logger.warning(f"Erreur récupération prix: {e}") return self._prices_cache or {} def _generate_alternative_candidates(self, watchlist: List[Dict], positions: Dict[str, Dict], current_prices: Dict[str, float]) -> List[Dict]: """ Génère des candidats alternatifs à partir des top cryptos Binance. Utilisé quand la watchlist ne contient pas assez de candidats hors positions. """ try: import requests # Récupérer les top performers des dernières 24h response = requests.get("https://api.binance.com/api/v3/ticker/24hr", timeout=10) if response.status_code != 200: return watchlist tickers = response.json() # Filtrer: USDT pairs, volume significatif, variation positive candidates = [] position_symbols = set(positions.keys()) watchlist_symbols = {w.get('symbol', '') for w in watchlist} for ticker in tickers: symbol = ticker.get('symbol', '') # Uniquement paires USDT if not symbol.endswith('USDT'): continue # Ignorer si déjà en position ou dans watchlist if symbol in position_symbols or symbol in watchlist_symbols: continue # ═══════════════════════════════════════════════════════════════════ # IGNORER LES STABLECOINS - Ils ne peuvent jamais atteindre le TP! # ═══════════════════════════════════════════════════════════════════ STABLECOINS = [ 'USDCUSDT', 'BUSDUSDT', 'TUSDUSDT', 'EURUSDT', 'FDUSDUSDT', 'USDEUSDT', 'USDPUSDT', 'DAIUSDT', 'PAXUSDT', 'GUSDUSDT', 'USTUSDT', 'USTCUSDT', 'FRAXUSDT', 'LUSDUSDT', 'CUSDUSDT', 'USDTUSDT', 'SUSDUSDT', 'RAIUSDT', 'MMUSDT', 'BFUSDUSDT' ] if symbol in STABLECOINS: continue # Ignorer aussi tout ce qui ressemble à un stablecoin (USD dans le nom) base_asset = symbol.replace('USDT', '') if 'USD' in base_asset or 'EUR' in base_asset: continue try: price_change_pct = float(ticker.get('priceChangePercent', 0)) quote_volume = float(ticker.get('quoteVolume', 0)) current_price = float(ticker.get('lastPrice', 0)) high_24h = float(ticker.get('highPrice', 0)) low_24h = float(ticker.get('lowPrice', 0)) # Critères de sélection: volume > 5M USDT, variation entre -2% et +5% if quote_volume < 5_000_000: continue # Éviter les crashs et les pumps excessifs if price_change_pct < -2 or price_change_pct > 5: continue # Calculer la position dans la range 24h price_range = high_24h - low_24h if price_range > 0: range_position = (current_price - low_24h) / price_range else: range_position = 0.5 # Calculer un score d'opportunité simple # Favoriser les cryptos proches du bas de la range avec momentum positif if range_position < 0.3 and price_change_pct > 0: score = 60 # Rebond potentiel elif range_position < 0.5 and price_change_pct > 0.5: score = 55 # Momentum positif, pas surachetée elif price_change_pct > 1 and range_position < 0.7: score = 50 # Bonne performance sans être en haut else: score = 45 # Par défaut candidates.append({ 'symbol': symbol, 'price': current_price, 'price_change_pct': price_change_pct, 'quote_volume': quote_volume, 'range_position': range_position, 'features': { 'price_current': current_price, 'ema_diff': 0.05 if price_change_pct > 0 else -0.02, 'ema_slope': price_change_pct / 20, 'ema_slope_long': price_change_pct / 30, 'momentum_3': price_change_pct / 3, 'momentum_5': price_change_pct / 5, 'rsi': 50 + (price_change_pct * 3), 'bb_position': range_position, 'bb_bandwidth': 2.0 }, 'smart_signal': 'POSSIBLE', 'score': score, 'generated': True # Marqueur pour indiquer que c'est généré }) except (ValueError, TypeError): continue # Trier par score et prendre les 10 meilleurs candidates.sort(key=lambda x: x.get('score', 0), reverse=True) top_candidates = candidates[:10] logger.info(f"🔄 {len(top_candidates)} candidats alternatifs générés") for c in top_candidates[:5]: logger.debug(f" {c['symbol']}: score={c['score']}, change={c['price_change_pct']:.2f}%, vol={c['quote_volume']/1e6:.1f}M") # Ajouter à la watchlist return watchlist + top_candidates except Exception as e: logger.warning(f"Erreur génération candidats: {e}") return watchlist def calculate_cycle_end_score(self, position: Dict, features: Dict) -> Dict: """ Calcule un score indiquant si le cycle de gain est terminé pour une position. Score élevé = cycle terminé → candidat à la vente/rotation Args: position: Données de la position (entry_price, quantity, etc.) features: Indicateurs techniques actuels Returns: Dict avec score et raisons """ score = 0 reasons = [] entry_price = position.get('entry_price', 0) current_price = features.get('price_current', 0) if entry_price <= 0 or current_price <= 0: return {'score': 0, 'reasons': ['Données de prix invalides']} profit_pct = ((current_price / entry_price) - 1) * 100 # Récupérer les indicateurs ema_diff = features.get('ema_diff', 0) ema_slope = features.get('ema_slope', 0) ema_slope_long = features.get('ema_slope_long', 0) momentum_3 = features.get('momentum_3', 0) momentum_5 = features.get('momentum_5', 0) rsi = features.get('rsi', 50) bb_position = features.get('bb_position', 0.5) bb_bandwidth = features.get('bb_bandwidth', 1.5) # ══════════════════════════════════════════════════════════════════ # OPTIMISATION: Réduire les scores pour éviter les rotations prématurées # Les anciennes valeurs provoquaient 62% de rotations perdantes! # ══════════════════════════════════════════════════════════════════ # ══════════════════════════════════════════════════════════════════ # 1. MOMENTUM ÉPUISÉ (25 pts max - réduit de 35) # ══════════════════════════════════════════════════════════════════ if momentum_3 < -0.8: # Plus strict: -0.8 au lieu de -0.5 score += 25 reasons.append(f"🔻 Momentum effondré ({momentum_3:.2f}%)") elif momentum_3 < -0.4 and momentum_5 < -0.5: # Plus strict score += 18 reasons.append(f"📉 Momentum en baisse (m3={momentum_3:.2f}%)") elif momentum_3 < -0.2 and momentum_5 < -0.2: # Plus strict score += 10 reasons.append("↘️ Momentum négatif") # Supprimé: le "momentum plat" déclenchait trop de faux positifs # ══════════════════════════════════════════════════════════════════ # 2. TENDANCE EMA INVERSÉE (25 pts max - réduit de 30) # ══════════════════════════════════════════════════════════════════ # Si acheté au creux (EMA9 < EMA21), le cycle se termine quand EMA9 > EMA21 # et commence à redescendre - PLUS STRICT MAINTENANT if ema_diff > 0.8 and ema_slope < -0.15: # Plus strict: 0.8 et -0.15 # Était en tendance haussière, maintenant EMA9 baisse fortement score += 25 reasons.append(f"🔄 Pic atteint (EMA_diff={ema_diff:.2f}%, slope↓)") elif ema_diff > 0.5 and ema_slope < -0.2: # Plus strict score += 18 reasons.append("📊 EMA en retournement") elif ema_diff > 0.3 and ema_slope_long < -0.2: # Plus strict score += 12 reasons.append("⚠️ Tendance moyen terme baissière") # Supprimé: ema_slope < -0.15 seul déclenchait trop de faux positifs # ══════════════════════════════════════════════════════════════════ # 3. RSI EN SURACHAT OU EN BAISSE (15 pts max - réduit de 20) # ══════════════════════════════════════════════════════════════════ if rsi > 80: # Plus strict: 80 au lieu de 75 score += 15 reasons.append(f"🔴 RSI surachat ({rsi:.0f})") elif rsi > 70 and momentum_3 < -0.3: # Plus strict score += 10 reasons.append(f"⚠️ RSI élevé + baisse ({rsi:.0f})") # Supprimé: RSI > 55 était trop sensible # ══════════════════════════════════════════════════════════════════ # 4. POSITION BOLLINGER (12 pts max - réduit de 15) # ══════════════════════════════════════════════════════════════════ # Si proche de BB haute et momentum baisse fortement = sommet atteint if bb_position > 0.90 and momentum_3 < -0.3: # Plus strict score += 12 reasons.append("📈 Sommet BB (prêt à redescendre)") elif bb_position > 0.85 and ema_slope < -0.2: # Plus strict score += 8 reasons.append("BB haute + tendance baissière") # Supprimé: bb_position > 0.65 était trop sensible # ══════════════════════════════════════════════════════════════════ # 5. STAGNATION / CONSOLIDATION (10 pts max - réduit de 15) # ══════════════════════════════════════════════════════════════════ if bb_bandwidth < 0.3: # Plus strict: 0.3 au lieu de 0.5 score += 10 reasons.append(f"😴 Squeeze BB (volatilité morte: {bb_bandwidth:.2f}%)") # Supprimé: bb_bandwidth < 0.8 était trop sensible # Si momentum plat depuis longtemps - conditions plus strictes if abs(momentum_3) < 0.05 and abs(momentum_5) < 0.05: # Plus strict score += 8 reasons.append("⏸️ Stagnation totale") # ══════════════════════════════════════════════════════════════════ # 6. BONUS PROFIT - Seulement pour les profits significatifs # ══════════════════════════════════════════════════════════════════ if profit_pct >= 2.0: score += 20 # Profit élevé = vendre et passer à autre chose reasons.append(f"💰 Profit à sécuriser (+{profit_pct:.2f}%)") elif profit_pct >= 1.0: # TP atteint = cycle terminé score += 15 reasons.append(f"🎯 TP atteint (+{profit_pct:.2f}%)") elif profit_pct >= 0.5: # Plus strict: 0.5 au lieu de 0.2 score += 8 reasons.append(f"📈 Petit profit (+{profit_pct:.2f}%)") # ══════════════════════════════════════════════════════════════════ # 6b. POSITIONS EN PERTE - Rotation vers meilleure opportunité # ══════════════════════════════════════════════════════════════════ # Si en perte ET momentum négatif = position qui empire if profit_pct <= -0.3 and momentum_3 < 0: score += 20 reasons.append(f"📉 En perte avec momentum négatif ({profit_pct:.2f}%)") elif profit_pct <= -0.2 and ema_slope < -0.1: score += 15 reasons.append(f"⚠️ En perte avec tendance baissière ({profit_pct:.2f}%)") elif profit_pct <= -0.1 and momentum_3 < -0.1 and momentum_5 < -0.1: score += 12 reasons.append(f"↘️ Légère perte en aggravation ({profit_pct:.2f}%)") # ══════════════════════════════════════════════════════════════════ # 7. POSITION STAGNANTE - Bonus de temps si la position ne bouge pas # ══════════════════════════════════════════════════════════════════ # Plus une position stagne longtemps sans atteindre TP, plus elle est candidate entry_time = position.get('timestamp') or position.get('entry_time') if entry_time: try: if isinstance(entry_time, str): entry_dt = datetime.fromisoformat(entry_time) else: entry_dt = entry_time hold_minutes = (datetime.now() - entry_dt).total_seconds() / 60 # Bonus progressif pour stagnation dans le temps # Position qui stagne > 15 min = +5 pts, > 30 min = +10 pts, > 60 min = +15 pts if hold_minutes >= 60 and abs(profit_pct) < 0.3: score += 15 reasons.append(f"⏰ Stagne depuis {hold_minutes:.0f}min sans TP") elif hold_minutes >= 30 and abs(profit_pct) < 0.3: score += 10 reasons.append(f"⏰ Stagne depuis {hold_minutes:.0f}min") elif hold_minutes >= 15 and abs(profit_pct) < 0.2: score += 5 reasons.append(f"⏳ En attente depuis {hold_minutes:.0f}min") except: pass # Score maximum: ~95 maintenant, normaliser à 100 normalized_score = min(100, int(score * 100 / 95)) return { 'score': normalized_score, 'raw_score': score, 'profit_pct': profit_pct, 'reasons': reasons, 'cycle_ended': normalized_score >= self.config['min_cycle_end_score'] } def calculate_opportunity_score(self, candidate: Dict, features: Dict) -> Dict: """ Calcule un score d'opportunité pour un candidat d'achat potentiel. Score élevé = excellente opportunité d'achat Args: candidate: Données du candidat (symbol, smart_signal, etc.) features: Indicateurs techniques Returns: Dict avec score et raisons """ score = 0 reasons = [] # Récupérer les indicateurs ema_diff = features.get('ema_diff', 0) ema_slope = features.get('ema_slope', 0) momentum_3 = features.get('momentum_3', 0) momentum_5 = features.get('momentum_5', 0) rsi = features.get('rsi', 50) bb_position = features.get('bb_position', 0.5) bb_bandwidth = features.get('bb_bandwidth', 1.5) # Smart signal et compatibilité smart_signal = candidate.get('smart_signal', 'HOLD') smart_score = candidate.get('smart_score', 0) # COMPATIBILITÉ IA DÉSACTIVÉE: Remplacée par Volatility Scorer # compatibility = candidate.get('ai_compatibility', {}) # compat_grade = compatibility.get('grade', 'C') # compat_score = compatibility.get('score', 50) # ══════════════════════════════════════════════════════════════════ # 1. SIGNAL IA (25 pts max) # ══════════════════════════════════════════════════════════════════ if smart_signal == 'ACHAT': score += 25 reasons.append("✅ Signal ACHAT IA") elif smart_signal == 'POSSIBLE': score += 15 reasons.append("🟡 Signal POSSIBLE") elif smart_signal == 'HOLD': score += 5 # Pas de points pour VENTE/ABANDONNEE # ══════════════════════════════════════════════════════════════════ # 2. SCORE IA (20 pts max) - Remplace l'ancien système de grades # Utilise le score IA réel qui inclut déjà volatility_scorer + performance_analyzer # ══════════════════════════════════════════════════════════════════ ai_score_bonus = int(smart_score / 100 * 20) # 0-100 → 0-20 pts score += ai_score_bonus if smart_score >= 80: reasons.append(f"🌟 Score IA élevé ({smart_score})") elif smart_score >= 60: reasons.append(f"✅ Score IA correct ({smart_score})") # ══════════════════════════════════════════════════════════════════ # 3. CREUX EMA - Zone d'achat idéale (25 pts max) # ══════════════════════════════════════════════════════════════════ # EMA9 < EMA21 = creux = meilleur moment pour acheter if ema_diff < -0.3: score += 25 reasons.append(f"📉 Creux profond EMA ({ema_diff:.2f}%)") elif ema_diff < -0.15: score += 20 reasons.append(f"📊 Bon creux EMA ({ema_diff:.2f}%)") elif ema_diff < -0.05: score += 12 reasons.append(f"↘️ Léger creux EMA") elif ema_diff > 0 and ema_diff < 0.3: score += 8 reasons.append("Tendance neutre/haussière") # ══════════════════════════════════════════════════════════════════ # 4. MOMENTUM DE REBOND (20 pts max) # ══════════════════════════════════════════════════════════════════ # Idéal: au creux avec début de rebond if ema_diff < 0 and momentum_3 > 0.1: score += 20 reasons.append(f"🚀 Rebond détecté (mom3=+{momentum_3:.2f}%)") elif momentum_3 > 0.05 and ema_slope >= 0: score += 15 reasons.append("📈 Momentum positif") elif momentum_3 >= -0.1: score += 8 reasons.append("➡️ Momentum stabilisé") # Momentum très négatif = pénalité elif momentum_3 < -0.5: score -= 15 reasons.append(f"⚠️ Chute en cours ({momentum_3:.2f}%)") # ══════════════════════════════════════════════════════════════════ # 5. RSI EN SURVENTE (15 pts max) # ══════════════════════════════════════════════════════════════════ if rsi < 25: score += 15 reasons.append(f"🔥 RSI survente forte ({rsi:.0f})") elif rsi < 35: score += 12 reasons.append(f"📊 RSI survente ({rsi:.0f})") elif rsi < 45: score += 6 reasons.append(f"RSI favorable ({rsi:.0f})") elif rsi > 70: score -= 10 reasons.append(f"⚠️ RSI trop élevé ({rsi:.0f})") # ══════════════════════════════════════════════════════════════════ # 6. POSITION BOLLINGER (15 pts max) # ══════════════════════════════════════════════════════════════════ if bb_position < 0.15: score += 15 reasons.append("💎 BB basse (zone d'achat)") elif bb_position < 0.25: score += 12 reasons.append("📉 Proche BB basse") elif bb_position < 0.4: score += 6 reasons.append("BB zone basse") elif bb_position > 0.85: score -= 10 reasons.append("⚠️ BB trop haute") # ══════════════════════════════════════════════════════════════════ # 7. VOLATILITÉ (10 pts max) # ══════════════════════════════════════════════════════════════════ if 1.0 <= bb_bandwidth <= 3.0: score += 10 reasons.append(f"✅ Bonne volatilité ({bb_bandwidth:.2f}%)") elif 0.5 <= bb_bandwidth <= 4.0: score += 5 else: reasons.append(f"⚠️ Volatilité anormale ({bb_bandwidth:.2f}%)") # Score maximum théorique: ~130, normaliser à 100 normalized_score = max(0, min(100, int(score * 100 / 130))) return { 'score': normalized_score, 'raw_score': score, 'symbol': candidate.get('symbol', ''), 'reasons': reasons, 'is_opportunity': normalized_score >= self.config['min_opportunity_score'] } def find_rotation_opportunities(self, positions: Dict[str, Dict], watchlist: List[Dict], features_map: Dict[str, Dict]) -> List[Dict]: """ Identifie les meilleures opportunités de rotation. Args: positions: Positions actuelles {symbol: position_data} watchlist: Liste des candidats d'achat avec features features_map: Dictionnaire {symbol: features} Returns: Liste des rotations recommandées triées par avantage """ if not self.config.get('enabled', True): return [] # Vérifier le cooldown horaire self._check_hourly_reset() if self.rotations_this_hour >= self.config['max_rotations_per_hour']: logger.info(f"⏳ Max rotations/heure atteint ({self.rotations_this_hour})") return [] rotations = [] # Récupérer les prix actuels pour calculer le P&L réel current_prices = self.get_current_prices() # Si la watchlist ne contient pas assez de candidats alternatifs, # générer des candidats potentiels à partir des top cryptos watchlist_symbols = {w.get('symbol', '') for w in watchlist} candidates_count = len(watchlist_symbols - set(positions.keys())) if candidates_count < 5: logger.info(f"🔄 Watchlist limitée ({candidates_count} candidats), génération de candidats alternatifs...") watchlist = list(watchlist) # Copie watchlist = self._generate_alternative_candidates(watchlist, positions, current_prices) # 1. Analyser toutes les positions actuelles positions_scores = [] logger.debug(f"🔄 Analyse rotation: {len(positions)} positions, {len(watchlist)} candidats") for symbol, pos in positions.items(): features = features_map.get(symbol, {}).copy() # Enrichir les features avec le prix actuel réel if symbol in current_prices: real_price = current_prices[symbol] features['price_current'] = real_price entry_price = pos.get('entry_price', 0) if entry_price > 0: profit_pct = ((real_price / entry_price) - 1) * 100 # Ajuster momentum en fonction du P&L réel features['momentum_3'] = features.get('momentum_3', profit_pct / 5) features['momentum_5'] = features.get('momentum_5', profit_pct / 8) logger.debug(f" {symbol}: Prix actuel={real_price:.6f}, P&L={profit_pct:.2f}%") if not features: logger.debug(f" {symbol}: Pas de features disponibles") continue # Vérifier le temps minimum en position (utiliser 'timestamp' OU 'entry_time') entry_time = pos.get('timestamp') or pos.get('entry_time') hold_minutes = 999 # Par défaut, considérer comme assez ancien if entry_time: try: if isinstance(entry_time, str): entry_dt = datetime.fromisoformat(entry_time) else: entry_dt = entry_time hold_minutes = (datetime.now() - entry_dt).total_seconds() / 60 if hold_minutes < self.config['min_hold_time_minutes']: logger.debug(f" {symbol}: Trop récent ({hold_minutes:.0f}min < {self.config['min_hold_time_minutes']}min)") continue except Exception as e: logger.debug(f" {symbol}: Erreur parsing date: {e}") # Vérifier cooldown de rotation if symbol in self.last_rotations: cooldown_end = self.last_rotations[symbol] + timedelta( minutes=self.config['rotation_cooldown_minutes'] ) if datetime.now() < cooldown_end: logger.debug(f" {symbol}: En cooldown rotation") continue cycle_analysis = self.calculate_cycle_end_score(pos, features) logger.debug(f" {symbol}: Score cycle={cycle_analysis['score']}, P&L={cycle_analysis['profit_pct']:.2f}%, ended={cycle_analysis['cycle_ended']}") # Les positions en perte sont PRIORITAIRES pour la rotation # min_profit_for_rotation est le seuil MINIMUM de perte autorisé (ex: -0.5%) # On VEUT vendre les positions en perte, pas les bloquer! # Bloquer seulement les positions avec trop petit profit positif if cycle_analysis['profit_pct'] > 0 and cycle_analysis['profit_pct'] < 0.1: # Position avec petit profit - pas encore prête pour rotation if cycle_analysis['score'] < 50: logger.debug(f" {symbol}: Petit profit ({cycle_analysis['profit_pct']:.2f}%) et score trop bas ({cycle_analysis['score']})") continue positions_scores.append({ 'symbol': symbol, 'position': pos, 'features': features, 'cycle_analysis': cycle_analysis, 'hold_minutes': hold_minutes }) # 2. Analyser les opportunités d'achat opportunities = [] logger.debug(f"🔍 Analyse de {len(watchlist)} candidats potentiels") for candidate in watchlist: symbol = candidate.get('symbol', '') # Ignorer si déjà en position if symbol in positions: logger.debug(f" {symbol}: SKIP (déjà en position)") continue features = candidate.get('features', {}) if not features: logger.debug(f" {symbol}: SKIP (pas de features)") continue # Ignorer si signal de vente ou abandonnée smart_signal = candidate.get('smart_signal', 'HOLD') if smart_signal in ['VENTE', 'ABANDONNEE', 'EN_POSITION']: logger.debug(f" {symbol}: SKIP (signal={smart_signal})") continue # CRITIQUE: Vérifier le score IA RÉEL après tous les blocages # Le score IA peut être réduit par performance_blacklist, volatility_poor, etc. ai_score = candidate.get('smart_score', 0) MIN_AI_SCORE_FOR_ROTATION = 60 # Seuil minimum strict if ai_score < MIN_AI_SCORE_FOR_ROTATION: logger.debug(f" {symbol}: SKIP (score IA={ai_score} < {MIN_AI_SCORE_FOR_ROTATION})") continue opp_analysis = self.calculate_opportunity_score(candidate, features) logger.debug(f" {symbol}: score={opp_analysis['score']}, ai_score={ai_score}, is_opp={opp_analysis['is_opportunity']}") if opp_analysis['is_opportunity']: opportunities.append({ 'symbol': symbol, 'candidate': candidate, 'features': features, 'opportunity_analysis': opp_analysis, 'ai_score': ai_score # Stocker le score IA réel pour trading_bot.py }) # 3. Trouver les meilleures paires rotation (vente → achat) logger.debug(f"📊 Positions fin de cycle: {sum(1 for p in positions_scores if p['cycle_analysis']['cycle_ended'])}") logger.debug(f"📊 Opportunités: {len(opportunities)}") for pos_item in positions_scores: if not pos_item['cycle_analysis']['cycle_ended']: continue for opp_item in opportunities: # Calculer l'avantage de rotation # sell_score = score de fin de cycle (20+ signifie cycle terminé) # buy_score = score d'opportunité (50+ est une bonne opportunité) sell_score = pos_item['cycle_analysis']['score'] buy_score = opp_item['opportunity_analysis']['score'] # Avantage = score d'opportunité - seuil minimum # Plus l'opportunité est bonne, plus on veut faire la rotation # L'avantage dépend principalement de la qualité de l'opportunité d'achat advantage = buy_score - self.config['min_opportunity_score'] # Bonus si la position est vraiment en difficulté (perte) if pos_item['cycle_analysis'].get('profit_pct', 0) < -0.3: advantage += 5 # +5 pts si position en perte > 0.3% logger.debug(f" Rotation {pos_item['symbol']} -> {opp_item['symbol']}: sell={sell_score}, buy={buy_score}, advantage={advantage}") if advantage >= self.config['min_score_advantage']: rotations.append({ 'sell_symbol': pos_item['symbol'], 'sell_score': sell_score, 'sell_reasons': pos_item['cycle_analysis']['reasons'], 'sell_profit_pct': pos_item['cycle_analysis']['profit_pct'], 'buy_symbol': opp_item['symbol'], 'buy_score': buy_score, 'buy_ai_score': opp_item.get('ai_score', 0), # Score IA réel (après blocages) 'buy_reasons': opp_item['opportunity_analysis']['reasons'], 'advantage': advantage, 'rotation_score': sell_score + buy_score, 'timestamp': datetime.now().isoformat() }) # Trier par avantage décroissant rotations.sort(key=lambda x: x['advantage'], reverse=True) return rotations def should_rotate(self, rotation: Dict) -> Tuple[bool, str]: """ Décide si une rotation doit être exécutée. Args: rotation: Données de la rotation proposée Returns: (should_execute, reason) """ if not self.config.get('enabled', True): return False, "Rotation désactivée" # ══════════════════════════════════════════════════════════════════════ # BLOCAGE CRASH MARCHÉ - Analyse 29/12: rotations en crash = double perte! # Vérifier si le mode crash global est actif (défini dans ai_predictor.py) # ══════════════════════════════════════════════════════════════════════ try: from ai_predictor import AIPredictor if AIPredictor._market_crash_mode: return False, f"Mode CRASH MARCHÉ actif - pas de rotation!" except Exception: pass # Continuer si import échoue # Vérifier le cooldown horaire self._check_hourly_reset() if self.rotations_this_hour >= self.config['max_rotations_per_hour']: return False, f"Max rotations/heure ({self.config['max_rotations_per_hour']})" # Vérifier l'avantage minimum if rotation['advantage'] < self.config['min_score_advantage']: return False, f"Avantage insuffisant ({rotation['advantage']} < {self.config['min_score_advantage']})" # Vérifier les scores individuels if rotation['sell_score'] < self.config['min_cycle_end_score']: return False, f"Score vente trop bas ({rotation['sell_score']})" if rotation['buy_score'] < self.config['min_opportunity_score']: return False, f"Score achat trop bas ({rotation['buy_score']})" return True, "Rotation validée" def execute_rotation(self, rotation: Dict, execute_callback=None) -> Dict: """ Exécute une rotation (vente puis achat). Args: rotation: Données de la rotation execute_callback: Fonction pour exécuter les ordres Returns: Résultat de l'exécution """ should_exec, reason = self.should_rotate(rotation) if not should_exec: return { 'success': False, 'reason': reason, 'rotation': rotation } sell_symbol = rotation['sell_symbol'] buy_symbol = rotation['buy_symbol'] logger.info(f"\n🔄 ROTATION EN COURS") logger.info(f" 📤 VENTE: {sell_symbol} (Score fin cycle: {rotation['sell_score']})") logger.info(f" 📥 ACHAT: {buy_symbol} (Score opportunité: {rotation['buy_score']})") logger.info(f" 📊 Avantage: +{rotation['advantage']} pts") result = { 'success': True, 'sell_symbol': sell_symbol, 'buy_symbol': buy_symbol, 'sell_executed': False, 'buy_executed': False, 'rotation': rotation } if execute_callback: try: # Exécuter la vente sell_result = execute_callback(sell_symbol, 'SELL', f"🔄 ROTATION: Cycle terminé (score={rotation['sell_score']})") result['sell_executed'] = sell_result if sell_result: # Exécuter l'achat buy_result = execute_callback(buy_symbol, 'BUY', f"🔄 ROTATION: Opportunité détectée (score={rotation['buy_score']})") result['buy_executed'] = buy_result except Exception as e: logger.error(f"Erreur exécution rotation: {e}") result['success'] = False result['error'] = str(e) # Enregistrer la rotation if result.get('sell_executed') or result.get('buy_executed'): self._record_rotation(rotation) return result def _record_rotation(self, rotation: Dict): """Enregistre une rotation effectuée.""" self.rotations_this_hour += 1 self.last_rotations[rotation['sell_symbol']] = datetime.now() self.last_rotations[rotation['buy_symbol']] = datetime.now() # Historique rotation['executed_at'] = datetime.now().isoformat() self.rotation_history.append(rotation) # Garder seulement les 50 dernières if len(self.rotation_history) > 50: self.rotation_history = self.rotation_history[-50:] # Sauvegarder self._save_history() def _check_hourly_reset(self): """Reset le compteur horaire si nécessaire.""" now = datetime.now() if (now - self.hour_start).total_seconds() >= 3600: self.rotations_this_hour = 0 self.hour_start = now def _save_history(self): """Sauvegarde l'historique des rotations.""" try: filepath = os.path.join(os.path.dirname(__file__), 'rotation_history.json') with open(filepath, 'w', encoding='utf-8') as f: json.dump(self.rotation_history, f, indent=2, ensure_ascii=False) except Exception as e: logger.debug(f"Erreur sauvegarde historique rotation: {e}") def get_status(self) -> Dict: """Retourne le statut du système de rotation.""" self._check_hourly_reset() return { 'enabled': self.config.get('enabled', True), 'rotations_this_hour': self.rotations_this_hour, 'max_per_hour': self.config['max_rotations_per_hour'], 'remaining_this_hour': max(0, self.config['max_rotations_per_hour'] - self.rotations_this_hour), 'hour_reset_in': 3600 - (datetime.now() - self.hour_start).total_seconds(), 'total_rotations': len(self.rotation_history), 'last_rotation': self.rotation_history[-1] if self.rotation_history else None, 'config': self.config } # Instance globale _smart_rotation = None def get_smart_rotation() -> SmartRotation: """Retourne l'instance globale du gestionnaire de rotation.""" global _smart_rotation if _smart_rotation is None: _smart_rotation = SmartRotation() return _smart_rotation def analyze_rotation_opportunity(positions: Dict, watchlist: List[Dict], features_map: Dict) -> List[Dict]: """ Fonction utilitaire pour analyser les opportunités de rotation. Args: positions: Positions actuelles watchlist: Candidats d'achat features_map: Features par symbole Returns: Liste des rotations recommandées """ rotation_mgr = get_smart_rotation() return rotation_mgr.find_rotation_opportunities(positions, watchlist, features_map) if __name__ == '__main__': # Test du module logging.basicConfig(level=logging.INFO) # Données de test test_position = { 'entry_price': 100.0, 'quantity': 1.0, 'entry_time': datetime.now() - timedelta(hours=1) } test_features_sell = { 'price_current': 102.5, # +2.5% profit 'ema_diff': 0.8, # EMA9 > EMA21 (cycle haussier terminé) 'ema_slope': -0.15, # EMA en baisse 'momentum_3': -0.3, # Momentum négatif 'momentum_5': -0.5, 'rsi': 72, # RSI élevé 'bb_position': 0.85, # Proche BB haute 'bb_bandwidth': 1.2 } test_features_buy = { 'price_current': 50.0, 'ema_diff': -0.25, # EMA9 < EMA21 (creux) 'ema_slope': 0.05, # Début de rebond 'momentum_3': 0.15, # Momentum positif 'momentum_5': 0.08, 'rsi': 32, # RSI survente 'bb_position': 0.2, # Proche BB basse 'bb_bandwidth': 1.5 } test_candidate = { 'symbol': 'ETHUSDT', 'smart_signal': 'ACHAT', 'smart_score': 75, 'features': test_features_buy, 'ai_compatibility': {'grade': 'A', 'score': 85} } rotation = SmartRotation() # Test analyse cycle cycle_result = rotation.calculate_cycle_end_score(test_position, test_features_sell) print(f"\n{'='*60}") print("🔴 Analyse Fin de Cycle (BTCUSDT)") print(f"{'='*60}") print(f"Score: {cycle_result['score']}/100") print(f"Profit: {cycle_result['profit_pct']:+.2f}%") print(f"Cycle terminé: {'✅ OUI' if cycle_result['cycle_ended'] else '❌ NON'}") print("Raisons:") for r in cycle_result['reasons']: print(f" - {r}") # Test analyse opportunité opp_result = rotation.calculate_opportunity_score(test_candidate, test_features_buy) print(f"\n{'='*60}") print("🟢 Analyse Opportunité (ETHUSDT)") print(f"{'='*60}") print(f"Score: {opp_result['score']}/100") print(f"Opportunité valide: {'✅ OUI' if opp_result['is_opportunity'] else '❌ NON'}") print("Raisons:") for r in opp_result['reasons']: print(f" - {r}") # Simuler une rotation print(f"\n{'='*60}") print("🔄 Simulation Rotation") print(f"{'='*60}") advantage = opp_result['score'] - (100 - cycle_result['score']) print(f"Vente BTCUSDT: score {cycle_result['score']}") print(f"Achat ETHUSDT: score {opp_result['score']}") print(f"Avantage rotation: +{advantage} pts") print(f"Rotation recommandée: {'✅ OUI' if advantage >= 25 else '❌ NON'}")