""" Stratégies INVERSE Ultra-Agressives V7: Ultra Leverage sur signaux parfaits V8: Pyramiding (ajout de positions) V9: Optimisation par stratégie """ import pandas as pd import numpy as np from typing import Optional, Dict, List import logging import sys sys.path.append('..') import config from signals.short_term_signals import Signal, ShortTermSignalGenerator from strategies.inverse_strategy_optimized import MarketRegimeDetectorAdvanced class InverseStrategyV7_UltraLeverage: """V7: Leverage et risk dynamiques selon qualité du signal""" def __init__(self): self.logger = logging.getLogger(self.__class__.__name__) self.signal_generator = ShortTermSignalGenerator() self.regime_detector = MarketRegimeDetectorAdvanced() self.version = "V7_ULTRA_LEVERAGE" def generate_inverse_signal(self, df: pd.DataFrame, symbol: str) -> Optional[Signal]: regime = self.regime_detector.detect_regime(df) signals = self.signal_generator.generate_signals(df, symbol) if not signals or regime != 'RANGE': return None best_signal = max(signals, key=lambda x: x.score) inverse_side = 'SHORT' if best_signal.side == 'LONG' else 'LONG' return Signal( symbol=symbol, side=inverse_side, score=best_signal.score, strategy=f'INVERSE_V7_{best_signal.strategy}', entry_price=best_signal.entry_price, stop_loss=best_signal.take_profit, take_profit=best_signal.stop_loss, confluences=best_signal.confluences + ['ULTRA_LEV'], timeframe=best_signal.timeframe, timestamp=best_signal.timestamp ) def get_leverage_and_risk(self, score: float, num_confluences: int) -> tuple: """Leverage et risk dynamiques""" # Signal parfait: score 1.0 + 4 confluences if score >= 1.0 and num_confluences >= 4: return 250, 0.03 # 250x leverage, 3% risk # Excellent: score 1.0 ou 4+ confluences elif score >= 1.0 or num_confluences >= 4: return 225, 0.025 # 225x leverage, 2.5% risk # Très bon: score 0.95+ elif score >= 0.95: return 200, 0.022 # 200x leverage, 2.2% risk # Bon: score 0.90+ elif score >= 0.90: return 200, 0.02 # 200x leverage, 2% risk # Standard else: return 150, 0.015 # 150x leverage, 1.5% risk class InverseStrategyV8_Pyramiding: """V8: Pyramiding - Ajoute à la position en profit""" def __init__(self): self.logger = logging.getLogger(self.__class__.__name__) self.signal_generator = ShortTermSignalGenerator() self.regime_detector = MarketRegimeDetectorAdvanced() self.version = "V8_PYRAMIDING" def generate_inverse_signal(self, df: pd.DataFrame, symbol: str) -> Optional[Signal]: regime = self.regime_detector.detect_regime(df) signals = self.signal_generator.generate_signals(df, symbol) if not signals or regime != 'RANGE': return None best_signal = max(signals, key=lambda x: x.score) inverse_side = 'SHORT' if best_signal.side == 'LONG' else 'LONG' return Signal( symbol=symbol, side=inverse_side, score=best_signal.score, strategy=f'INVERSE_V8_{best_signal.strategy}', entry_price=best_signal.entry_price, stop_loss=best_signal.take_profit, take_profit=best_signal.stop_loss, confluences=best_signal.confluences + ['PYRAMID'], timeframe=best_signal.timeframe, timestamp=best_signal.timestamp ) def should_add_to_position(self, side: str, entry_price: float, current_price: float, pyramids_added: int) -> tuple: """Détermine si on doit pyramider""" if side == 'LONG': pnl_pct = (current_price - entry_price) / entry_price else: pnl_pct = (entry_price - current_price) / entry_price # Premier ajout à +0.5% if pyramids_added == 0 and pnl_pct >= 0.005: return True, 0.5 # Ajouter 50% de la position initiale # Deuxième ajout à +1.0% elif pyramids_added == 1 and pnl_pct >= 0.010: return True, 0.25 # Ajouter 25% de la position initiale return False, 0 class InverseStrategyV9_StrategyOptimizer: """V9: Optimise selon quelle stratégie performe le mieux""" def __init__(self): self.logger = logging.getLogger(self.__class__.__name__) self.signal_generator = ShortTermSignalGenerator() self.regime_detector = MarketRegimeDetectorAdvanced() self.version = "V9_STRATEGY_OPT" # Tracking de performance par stratégie self.strategy_performance = { 'SCALPING_RSI_VOLUME': {'wins': 0, 'losses': 0, 'total_pnl': 0}, 'MACD_MOMENTUM': {'wins': 0, 'losses': 0, 'total_pnl': 0}, 'BOLLINGER_BOUNCE': {'wins': 0, 'losses': 0, 'total_pnl': 0} } def generate_inverse_signal(self, df: pd.DataFrame, symbol: str) -> Optional[Signal]: regime = self.regime_detector.detect_regime(df) signals = self.signal_generator.generate_signals(df, symbol) if not signals or regime != 'RANGE': return None # Filtrer selon performance historique viable_signals = [] for sig in signals: base_strategy = sig.strategy perf = self.strategy_performance[base_strategy] # Skip stratégies avec plus de 2 losses de suite if perf['losses'] >= 2 and perf['wins'] == 0: continue # Boost score si stratégie performe bien win_rate = perf['wins'] / (perf['wins'] + perf['losses']) if (perf['wins'] + perf['losses']) > 0 else 0.5 if win_rate >= 0.8: sig.score *= 1.1 # +10% score boost viable_signals.append(sig) if not viable_signals: # Si toutes filtrées, prendre la meilleure quand même viable_signals = signals best_signal = max(viable_signals, key=lambda x: x.score) inverse_side = 'SHORT' if best_signal.side == 'LONG' else 'LONG' return Signal( symbol=symbol, side=inverse_side, score=best_signal.score, strategy=f'INVERSE_V9_{best_signal.strategy}', entry_price=best_signal.entry_price, stop_loss=best_signal.take_profit, take_profit=best_signal.stop_loss, confluences=best_signal.confluences + ['STRAT_OPT'], timeframe=best_signal.timeframe, timestamp=best_signal.timestamp ) def update_performance(self, strategy_name: str, pnl_usd: float): """Met à jour les stats de performance""" # Extraire le nom de base de la stratégie for base_strategy in self.strategy_performance.keys(): if base_strategy in strategy_name: if pnl_usd > 0: self.strategy_performance[base_strategy]['wins'] += 1 else: self.strategy_performance[base_strategy]['losses'] += 1 self.strategy_performance[base_strategy]['total_pnl'] += pnl_usd break def get_position_multiplier(self, strategy_name: str) -> float: """Multiplicateur de position selon performance""" for base_strategy, perf in self.strategy_performance.items(): if base_strategy in strategy_name: total_trades = perf['wins'] + perf['losses'] if total_trades >= 3: win_rate = perf['wins'] / total_trades if win_rate >= 0.9: return 1.5 # +50% position elif win_rate >= 0.8: return 1.25 # +25% position elif win_rate <= 0.3: return 0.5 # -50% position return 1.0 # Standard class InverseStrategyV10_Combined_Aggressive: """V10: Combinaison de toutes les optimisations agressives""" def __init__(self): self.logger = logging.getLogger(self.__class__.__name__) self.signal_generator = ShortTermSignalGenerator() self.regime_detector = MarketRegimeDetectorAdvanced() self.version = "V10_COMBINED_AGGRESSIVE" # Tracking performance self.strategy_performance = { 'SCALPING_RSI_VOLUME': {'wins': 0, 'losses': 0, 'total_pnl': 0}, 'MACD_MOMENTUM': {'wins': 0, 'losses': 0, 'total_pnl': 0}, 'BOLLINGER_BOUNCE': {'wins': 0, 'losses': 0, 'total_pnl': 0} } def generate_inverse_signal(self, df: pd.DataFrame, symbol: str) -> Optional[Signal]: regime = self.regime_detector.detect_regime(df) signals = self.signal_generator.generate_signals(df, symbol) if not signals or regime != 'RANGE': return None # Filtrage et scoring best_score = 0 best_signal = None for sig in signals: base_strategy = sig.strategy perf = self.strategy_performance[base_strategy] # Skip mauvaises stratégies if perf['losses'] >= 2 and perf['wins'] == 0: continue # Boost score win_rate = perf['wins'] / (perf['wins'] + perf['losses']) if (perf['wins'] + perf['losses']) > 0 else 0.5 adjusted_score = sig.score * (1.1 if win_rate >= 0.8 else 1.0) if adjusted_score > best_score: best_score = adjusted_score best_signal = sig if not best_signal: best_signal = max(signals, key=lambda x: x.score) inverse_side = 'SHORT' if best_signal.side == 'LONG' else 'LONG' return Signal( symbol=symbol, side=inverse_side, score=best_signal.score, strategy=f'INVERSE_V10_{best_signal.strategy}', entry_price=best_signal.entry_price, stop_loss=best_signal.take_profit, take_profit=best_signal.stop_loss, confluences=best_signal.confluences + ['COMBINED_AGG'], timeframe=best_signal.timeframe, timestamp=best_signal.timestamp ) def get_leverage_and_risk(self, score: float, num_confluences: int, strategy_name: str) -> tuple: """Leverage/risk ultra dynamique""" # Base selon score if score >= 1.0 and num_confluences >= 4: base_lev, base_risk = 250, 0.03 elif score >= 1.0 or num_confluences >= 4: base_lev, base_risk = 225, 0.025 elif score >= 0.95: base_lev, base_risk = 200, 0.022 else: base_lev, base_risk = 200, 0.02 # Ajustement selon performance stratégie for base_strategy, perf in self.strategy_performance.items(): if base_strategy in strategy_name: total = perf['wins'] + perf['losses'] if total >= 3: wr = perf['wins'] / total if wr >= 0.9: base_risk *= 1.3 # +30% risk elif wr <= 0.3: base_risk *= 0.7 # -30% risk break return base_lev, base_risk def should_add_to_position(self, side: str, entry_price: float, current_price: float, pyramids_added: int) -> tuple: """Pyramiding agressif""" if side == 'LONG': pnl_pct = (current_price - entry_price) / entry_price else: pnl_pct = (entry_price - current_price) / entry_price if pyramids_added == 0 and pnl_pct >= 0.005: return True, 0.5 elif pyramids_added == 1 and pnl_pct >= 0.010: return True, 0.25 return False, 0 def update_performance(self, strategy_name: str, pnl_usd: float): """Update stats""" for base_strategy in self.strategy_performance.keys(): if base_strategy in strategy_name: if pnl_usd > 0: self.strategy_performance[base_strategy]['wins'] += 1 else: self.strategy_performance[base_strategy]['losses'] += 1 self.strategy_performance[base_strategy]['total_pnl'] += pnl_usd break