""" Stratégie INVERSE - Out of the box thinking Si les signaux classiques perdent dans un range, l'inverse devrait gagner """ import pandas as pd import numpy as np from typing import Optional, Tuple import logging import sys sys.path.append('..') import config from signals.short_term_signals import Signal, ShortTermSignalGenerator class MarketRegimeDetector: """Détecte si on est en RANGE ou TREND""" def __init__(self): self.logger = logging.getLogger(self.__class__.__name__) def detect_regime(self, df: pd.DataFrame, lookback: int = 100) -> str: """ Détecte le régime: RANGE ou TREND Logique simple: - Range: prix oscille entre support/resistance sans direction claire - Trend: prix fait des HH/HL (uptrend) ou LH/LL (downtrend) """ if len(df) < lookback: return 'UNKNOWN' window = df.iloc[-lookback:] closes = window['close'].values # Calculer high/low de la période period_high = closes.max() period_low = closes.min() period_range = period_high - period_low # Prix actuel current = closes[-1] # Distance du high/low en % distance_from_high = (period_high - current) / period_high distance_from_low = (current - period_low) / period_low # 1. Détection RANGE: prix reste dans 60% milieu de la range middle_zone = 0.6 # 60% du milieu if distance_from_high < middle_zone and distance_from_low < middle_zone: # Vérifier volatilité faible returns = np.diff(closes) / closes[:-1] volatility = np.std(returns) # Range = faible volatilité + prix au milieu if volatility < 0.003: # < 0.3% return 'RANGE' # 2. Détection TREND: prix fait des higher highs ou lower lows # Diviser en 4 quartiles q1 = closes[:lookback//4] q2 = closes[lookback//4:lookback//2] q3 = closes[lookback//2:3*lookback//4] q4 = closes[3*lookback//4:] # Moyennes par quartile avg_q1 = q1.mean() avg_q2 = q2.mean() avg_q3 = q3.mean() avg_q4 = q4.mean() # Uptrend: chaque quartile plus haut que le précédent if avg_q2 > avg_q1 and avg_q3 > avg_q2 and avg_q4 > avg_q3: return 'UPTREND' # Downtrend: chaque quartile plus bas if avg_q2 < avg_q1 and avg_q3 < avg_q2 and avg_q4 < avg_q3: return 'DOWNTREND' # Par défaut: range (choppy) return 'RANGE' class InverseStrategy: """Stratégie qui INVERSE les signaux en fonction du régime""" def __init__(self): self.logger = logging.getLogger(self.__class__.__name__) self.signal_generator = ShortTermSignalGenerator() self.regime_detector = MarketRegimeDetector() self.logger.info("🔄 InverseStrategy initialisée") def generate_inverse_signal(self, df: pd.DataFrame, symbol: str) -> Optional[Signal]: """ Génère un signal en inversant selon le régime Returns: Signal inversé si range, normal si trend, None si aucun """ # Détecter régime regime = self.regime_detector.detect_regime(df) self.logger.debug(f"{symbol}: Régime détecté = {regime}") # Générer signaux classiques signals = self.signal_generator.generate_signals(df, symbol) if not signals: return None # Prendre le meilleur signal best_signal = max(signals, key=lambda x: x.score) # LOGIQUE INVERSE if regime == 'RANGE': # En RANGE: INVERSER le signal inverse_side = 'SHORT' if best_signal.side == 'LONG' else 'LONG' # Inverser aussi SL et TP if inverse_side == 'LONG': # Signal original était SHORT, on inverse new_sl = best_signal.take_profit # L'ancien TP devient SL new_tp = best_signal.stop_loss # L'ancien SL devient TP else: # Signal original était LONG, on inverse new_sl = best_signal.take_profit new_tp = best_signal.stop_loss self.logger.info( f"🔄 INVERSE: {symbol} {best_signal.side} → {inverse_side} " f"(régime: {regime}, score: {best_signal.score:.2f})" ) return Signal( symbol=symbol, side=inverse_side, score=best_signal.score, strategy=f'INVERSE_{best_signal.strategy}', entry_price=best_signal.entry_price, stop_loss=new_sl, take_profit=new_tp, confluences=best_signal.confluences + [f'INVERSE_{regime}'], timeframe=best_signal.timeframe, timestamp=best_signal.timestamp ) elif regime in ['UPTREND', 'DOWNTREND']: # En TREND: garder le signal normal self.logger.debug( f"➡️ NORMAL: {symbol} {best_signal.side} (régime: {regime})" ) return best_signal else: # Régime inconnu: pas de trade return None class AdaptiveInverseBacktester: """Backtester qui teste la stratégie inverse""" def __init__(self, initial_capital: float = 1000, leverage: int = 100): self.logger = logging.getLogger(self.__class__.__name__) self.initial_capital = initial_capital self.leverage = leverage self.inverse_strategy = InverseStrategy() self.trading_fee = config.TRADING_FEE_TAKER self.slippage = config.SLIPPAGE_PCT def backtest(self, df: pd.DataFrame, symbol: str): """Backtest avec stratégie inverse""" from backtesting.backtester import DegenBacktester, Trade self.logger.info(f"\n{'='*60}") self.logger.info(f"BACKTEST INVERSE: {symbol} | {len(df)} chandeliers") self.logger.info(f"{'='*60}") capital = self.initial_capital trades = [] i = 50 # Warmup while i < len(df) - 100: window = df.iloc[:i+1].copy() # Générer signal inverse signal = self.inverse_strategy.generate_inverse_signal(window, symbol) if not signal: i += 1 continue # Simuler trade (logique simplifiée) entry_time = df.iloc[i]['timestamp'] entry_price = signal.entry_price sl = signal.stop_loss tp = signal.take_profit # Position size - AGRESSIF pour leverage élevé # Plus le leverage est haut, plus on peut risquer car SL plus serré if self.leverage >= 200: risk_pct = 0.02 # 2% pour 200x+ elif self.leverage >= 150: risk_pct = 0.015 # 1.5% pour 150x elif self.leverage >= 100: risk_pct = 0.012 # 1.2% pour 100x else: risk_pct = 0.01 # 1% pour <100x position_size_usd = self.initial_capital * risk_pct * self.leverage quantity = position_size_usd / entry_price # Chercher sortie for j in range(i + 1, min(i + 100, len(df))): current = df.iloc[j] high = current['high'] low = current['low'] exit_price = None exit_reason = None if signal.side == 'LONG': if low <= sl: exit_price = sl exit_reason = 'SL' elif high >= tp: exit_price = tp exit_reason = 'TP' else: # SHORT if high >= sl: exit_price = sl exit_reason = 'SL' elif low <= tp: exit_price = tp exit_reason = 'TP' if exit_price: # Calculer PnL if signal.side == 'LONG': pnl_pct = (exit_price - entry_price) / entry_price else: pnl_pct = (entry_price - exit_price) / entry_price pnl_usd = position_size_usd * pnl_pct pnl_usd -= position_size_usd * self.trading_fee * 2 # Entry + exit capital += pnl_usd trades.append({ 'entry_time': entry_time, 'exit_time': current['timestamp'], 'side': signal.side, 'strategy': signal.strategy, 'pnl_pct': pnl_pct * 100, 'pnl_usd': pnl_usd, 'exit_reason': exit_reason }) i = j + 1 break else: i += 1 # Résultats if not trades: self.logger.warning("Aucun trade généré") return None winning = [t for t in trades if t['pnl_usd'] > 0] losing = [t for t in trades if t['pnl_usd'] <= 0] total_return = (capital - self.initial_capital) / self.initial_capital * 100 win_rate = len(winning) / len(trades) if trades else 0 self.logger.info(f"\n{'='*60}") self.logger.info(f"📊 RÉSULTATS INVERSE") self.logger.info(f"{'='*60}") self.logger.info(f"Capital initial: ${self.initial_capital:,.2f}") self.logger.info(f"Capital final: ${capital:,.2f}") self.logger.info(f"Return: {total_return:+.2f}%") self.logger.info(f"Total trades: {len(trades)}") self.logger.info(f"Win rate: {win_rate*100:.1f}%") self.logger.info(f"Gagnants: {len(winning)}") self.logger.info(f"Perdants: {len(losing)}") if winning: avg_win = np.mean([t['pnl_pct'] for t in winning]) self.logger.info(f"Gain moyen: {avg_win:+.2f}%") if losing: avg_loss = np.mean([t['pnl_pct'] for t in losing]) self.logger.info(f"Perte moyenne: {avg_loss:+.2f}%") self.logger.info(f"{'='*60}\n") return { 'total_return': total_return, 'win_rate': win_rate, 'trades': trades, 'final_capital': capital } if __name__ == '__main__': logging.basicConfig(level=logging.INFO) from apollo_client import ApolloDegenClient logger = logging.getLogger(__name__) logger.info("🔄 Test Stratégie INVERSE\n") # Récupérer données client = ApolloDegenClient() df = client.get_klines('BTCUSDT', '1m', limit=1500) logger.info(f"Données: {len(df)} chandeliers BTC 1m\n") # Tester stratégie inverse backtester = AdaptiveInverseBacktester(initial_capital=1000, leverage=100) result = backtester.backtest(df, 'BTCUSDT') if result: if result['total_return'] > 0: logger.info(f"✅ STRATÉGIE INVERSE PROFITABLE: {result['total_return']:+.2f}%") else: logger.info(f"❌ Stratégie inverse aussi négative: {result['total_return']:+.2f}%")