""" ProfitTaker Worker - Gestion des fermetures de positions Responsabilité : Fermer les positions selon différentes stratégies """ import logging import time from datetime import datetime, timedelta from threading import Thread, Lock from typing import Dict, List, Optional import os class ProfitTaker: """Worker spécialisé dans la fermeture de positions""" def __init__(self, hyperliquid_client=None): self.hyperliquid_client = hyperliquid_client self.running = False # Storage des positions actives reconstruites self.active_positions = {} # {symbol_side: PositionData} self.lock = Lock() # Configuration self.check_interval = 10 # Vérifier toutes les 10 secondes # Logger self.setup_logger() self.logger.info("=== PROFIT TAKER WORKER STARTED ===") def setup_logger(self): """Configure le système de logging""" self.logger = logging.getLogger("ProfitTaker") self.logger.setLevel(logging.INFO) # Créer le dossier logs si nécessaire logs_dir = "logs" if not os.path.exists(logs_dir): os.makedirs(logs_dir) # Handler fichier file_handler = logging.FileHandler(os.path.join(logs_dir, "profit_taker.log")) file_handler.setLevel(logging.INFO) # Handler console console_handler = logging.StreamHandler() console_handler.setLevel(logging.INFO) # Format formatter = logging.Formatter('%(asctime)s - [%(name)s] - %(message)s') file_handler.setFormatter(formatter) console_handler.setFormatter(formatter) # Fix encoding for file handler import codecs file_handler.stream = codecs.open(os.path.join(logs_dir, "profit_taker.log"), 'a', encoding='utf-8') if not self.logger.handlers: self.logger.addHandler(file_handler) self.logger.addHandler(console_handler) def get_fills_from_hyperliquid(self, lookback_hours: int = 24) -> List[Dict]: """Récupère les fills depuis Hyperliquid""" if not self.hyperliquid_client: self.logger.warning("Client Hyperliquid non disponible") return [] try: # Calculer la timestamp de début (lookback_hours heures en arrière) start_time = int((datetime.now() - timedelta(hours=lookback_hours)).timestamp() * 1000) #self.logger.info(f"Récupération des fills depuis {lookback_hours}h...") # Récupérer les fills depuis Hyperliquid all_fills = self.hyperliquid_client.get_fill_info(limit=1000) if not all_fills: self.logger.info("Aucun fill récupéré") return [] # Filtrer par date filtered_fills = [] for fill in all_fills: fill_time = int(fill.get('time', 0)) if fill_time >= start_time: filtered_fills.append(fill) self.logger.info(f"Récupérés {len(filtered_fills)}/{len(all_fills)} fills depuis {lookback_hours}h") return filtered_fills except Exception as e: self.logger.error(f"Erreur récupération fills: {e}") return [] def get_current_positions_from_hyperliquid(self) -> List[Dict]: """Récupère les positions actuelles depuis Hyperliquid""" if not self.hyperliquid_client: return [] try: positions = self.hyperliquid_client.get_positions() if positions: self.logger.info(f"Récupérées {len(positions)} positions actives depuis Hyperliquid") return positions else: self.logger.info("Aucune position active sur Hyperliquid") return [] except Exception as e: self.logger.error(f"Erreur récupération positions: {e}") return [] def reconstruct_trades_from_fills(self, fills: List[Dict]) -> Dict[str, Dict]: """Reconstitue les trades à partir des fills""" trades = {} for fill in fills: try: symbol = fill.get('coin', '').replace('-PERP', '') side = 'LONG' if fill.get('dir') == 'Open Long' or fill.get('dir') == 'Close Short' else 'SHORT' # Déterminer si c'est une ouverture ou fermeture direction = fill.get('dir', '') is_opening = 'Open' in direction size = abs(float(fill.get('sz', 0))) price = float(fill.get('px', 0)) timestamp = int(fill.get('time', 0)) trade_key = f"{symbol}_{side}" if trade_key not in trades: trades[trade_key] = { 'symbol': symbol, 'side': side, 'opens': [], 'closes': [], 'total_opened': 0, 'total_closed': 0, 'avg_open_price': 0, 'current_size': 0 } if is_opening: trades[trade_key]['opens'].append({ 'size': size, 'price': price, 'time': timestamp }) trades[trade_key]['total_opened'] += size else: trades[trade_key]['closes'].append({ 'size': size, 'price': price, 'time': timestamp }) trades[trade_key]['total_closed'] += size except Exception as e: self.logger.error(f"Erreur traitement fill: {e}") continue # Calculer les prix moyens et tailles actuelles for trade_key, trade_data in trades.items(): opens = trade_data['opens'] if opens: total_value = sum(o['size'] * o['price'] for o in opens) total_size = sum(o['size'] for o in opens) trade_data['avg_open_price'] = total_value / total_size if total_size > 0 else 0 trade_data['current_size'] = trade_data['total_opened'] - trade_data['total_closed'] return trades def sync_with_current_positions(self, reconstructed_trades: Dict[str, Dict], current_positions: List[Dict]) -> Dict[str, Dict]: """Synchronise les trades reconstruits avec les positions actuelles de Hyperliquid""" active_positions = {} # Créer un set des positions actuelles pour référence current_position_keys = set() current_position_data = {} for pos in current_positions: symbol = pos.get('symbol', '').replace('-PERP', '') side = pos.get('side', '').upper() key = f"{symbol}_{side}" current_position_keys.add(key) current_position_data[key] = pos # Ne garder que les trades qui correspondent aux positions actuelles for trade_key, trade_data in reconstructed_trades.items(): if trade_key in current_position_keys and trade_data['current_size'] > 0: # Enrichir avec les données actuelles de Hyperliquid current_pos = current_position_data[trade_key] # Récupérer le prix de marché actuel current_market_price = 0 try: if self.hyperliquid_client: current_market_price = self.hyperliquid_client.get_current_market_price(trade_data['symbol']) if not current_market_price: current_market_price = float(current_pos.get('entry_price', 0)) except: current_market_price = float(current_pos.get('entry_price', 0)) position_data = { 'symbol': trade_data['symbol'], 'side': trade_data['side'], 'entry_price': float(current_pos.get('entry_price', 0)), # Utiliser le vrai entry price d'Hyperliquid 'current_size': abs(float(current_pos.get('size', 0))), 'current_price': current_market_price, 'unrealized_pnl': float(current_pos.get('unrealized_pnl', 0)), 'leverage': int(current_pos.get('leverage', {}).get('value', 1)), 'opens_history': trade_data['opens'], 'closes_history': trade_data['closes'], 'liquidation_price': current_pos.get('liquidation_px'), 'last_updated': datetime.now().isoformat(), 'raw_hyperliquid_data': current_pos, 'peak_pnl_percent': float('-inf') # Initialiser à -∞ pour capturer toute valeur } # Calculer le PnL en pourcentage basé sur les prix if position_data['entry_price'] > 0 and position_data['current_price'] > 0: entry_price = position_data['entry_price'] current_price = position_data['current_price'] side = position_data['side'] # Calculer le PnL basé sur la différence de prix if side == 'LONG': price_diff_pct = ((current_price - entry_price) / entry_price) * 100 else: # SHORT price_diff_pct = ((entry_price - current_price) / entry_price) * 100 # Appliquer le levier au PnL position_data['pnl_percent'] = price_diff_pct * position_data['leverage'] else: position_data['pnl_percent'] = 0 # Calculer et mettre à jour le peakPNL (ne jamais l'écraser si inférieur) current_pnl = position_data['pnl_percent'] # Récupérer l'ancien peakPNL si la position existe déjà existing_position = self.active_positions.get(trade_key) previous_peak = existing_position.get('peak_pnl_percent', float('-inf')) if existing_position else float('-inf') # Mettre à jour le peakPNL seulement si le PnL actuel est supérieur (même négatif) position_data['peak_pnl_percent'] = max(previous_peak, current_pnl) active_positions[trade_key] = position_data return active_positions def update_active_positions(self): """Met à jour le repo des positions actives""" try: #self.logger.info("Mise à jour des positions actives...") # 1. Récupérer les fills récents fills = self.get_fills_from_hyperliquid(lookback_hours=72) # 3 jours de lookback # 2. Récupérer les positions actuelles current_positions = self.get_current_positions_from_hyperliquid() # 3. Reconstituer les trades reconstructed_trades = self.reconstruct_trades_from_fills(fills) self.logger.info(f"Reconstruits {len(reconstructed_trades)} trades depuis les fills") # 4. Synchroniser avec les positions actuelles active_positions = self.sync_with_current_positions(reconstructed_trades, current_positions) # 5. Mettre à jour le storage thread-safe with self.lock: self.active_positions = active_positions self.logger.info(f"Positions actives mises à jour: {len(active_positions)}") # Log clair et détaillé des positions if active_positions: self.logger.info("=" * 80) self.logger.info("POSITIONS ACTIVES:") self.logger.info("=" * 80) for key, pos in active_positions.items(): symbol = pos['symbol'] side = pos['side'] entry = pos['entry_price'] current = pos['current_price'] size = pos['current_size'] pnl_pct = pos['pnl_percent'] peak_pnl = pos['peak_pnl_percent'] leverage = pos['leverage'] # Formatage propre selon la taille du nombre if entry >= 1000: entry_str = f"${entry:,.2f}" current_str = f"${current:,.2f}" else: entry_str = f"${entry:.5f}" current_str = f"${current:.5f}" pnl_sign = "+" if pnl_pct >= 0 else "" pnl_color = "UP" if pnl_pct >= 0 else "DOWN" # Gestion de l'affichage du peak (éviter -inf) if peak_pnl == float('-inf'): peak_display = " N/A " else: peak_sign = "+" if peak_pnl >= 0 else "" peak_display = f"{peak_sign}{peak_pnl:>6.2f}%" self.logger.info(f"{symbol:<6} {side:<5} | Entry: {entry_str:<12} -> Current: {current_str:<12} | " f"Size: {size:<8.3f} | Leverage: {leverage:>2}x | " f"PnL: {pnl_sign}{pnl_pct:>6.2f}% | Peak: {peak_display} {pnl_color}") self.logger.info("=" * 80) else: self.logger.info("Aucune position active détectée") except Exception as e: self.logger.error(f"Erreur mise à jour positions: {e}") def close_position_by_symbol(self, symbol: str, reason: str = "STOP_LOSS") -> Dict: """Ferme une position par symbole via le client Hyperliquid""" try: if not self.hyperliquid_client: return {"success": False, "error": "Client Hyperliquid non disponible"} # Récupérer les positions actuelles hl_positions = self.hyperliquid_client.get_positions() target_position = None # Chercher la position par symbole for pos in hl_positions: pos_symbol = pos.get('symbol', '').replace('-PERP', '') if pos_symbol == symbol and float(pos.get('size', 0)) != 0: target_position = pos break if not target_position: return {"success": False, "error": f"Position {symbol} non trouvée"} # Déterminer le côté de fermeture (opposé à la position) position_side = target_position.get('side', '').upper() is_buy = (position_side == 'SHORT') # Pour fermer SHORT on achète, pour fermer LONG on vend size = abs(float(target_position.get('size', 0))) self.logger.info(f"Fermeture position {symbol} {position_side}: Size={size}, Reason={reason}") # Placer l'ordre de fermeture result = self.hyperliquid_client.place_market_order( symbol=symbol, is_buy=is_buy, size=size, reduce_only=True, order_tag=f"CLOSE_{reason}_{symbol}" ) if result.get('success'): self.logger.info(f"Position {symbol} fermée avec succès: {reason}") return {"success": True, "message": f"Position {symbol} fermée"} else: self.logger.error(f"Échec fermeture {symbol}: {result.get('error')}") return {"success": False, "error": result.get('error', 'Erreur inconnue')} except Exception as e: error_msg = f"Erreur fermeture position {symbol}: {e}" self.logger.error(error_msg) return {"success": False, "error": error_msg} def check_stop_loss_conditions(self): """Vérifie les conditions de stop loss pour toutes les positions actives""" stop_loss_threshold = -12.0 # -12% PnL with self.lock: positions_to_close = [] for key, position in self.active_positions.items(): current_pnl = position['pnl_percent'] symbol = position['symbol'] # Vérifier condition de stop loss if current_pnl <= stop_loss_threshold: self.logger.warning(f"STOP LOSS TRIGGERED: {symbol} PnL={current_pnl:.2f}% <= {stop_loss_threshold}%") positions_to_close.append(symbol) # Fermer les positions en dehors du lock for symbol in positions_to_close: try: result = self.close_position_by_symbol(symbol, "STOP_LOSS_12PCT") if result.get('success'): self.logger.critical(f"STOP LOSS EXECUTED: {symbol} fermée avec succès") else: self.logger.error(f"STOP LOSS FAILED: {symbol} - {result.get('error')}") except Exception as e: self.logger.error(f"Erreur stop loss {symbol}: {e}") def check_trailing_stop_conditions(self): """Vérifie les conditions de trailing stop pour les positions en profit""" min_profit_threshold = 3.0 # Minimum 3% de profit pour activer le trailing trailing_drawdown = 3.0 # Fermer si on recule de 3% depuis le peak with self.lock: positions_to_close = [] for key, position in self.active_positions.items(): current_pnl = position['pnl_percent'] peak_pnl = position['peak_pnl_percent'] symbol = position['symbol'] # Conditions pour le trailing stop: # 1. Position en profit d'au moins 3% # 2. On a reculé de 3% depuis le peak if (current_pnl >= min_profit_threshold and peak_pnl != float('-inf') and peak_pnl >= min_profit_threshold and current_pnl <= (peak_pnl - trailing_drawdown)): drawdown = peak_pnl - current_pnl self.logger.warning(f"TRAILING STOP TRIGGERED: {symbol} PnL={current_pnl:.2f}% | Peak={peak_pnl:.2f}% | Drawdown={drawdown:.2f}%") positions_to_close.append(symbol) # Fermer les positions en dehors du lock for symbol in positions_to_close: try: result = self.close_position_by_symbol(symbol, "TRAILING_STOP_3PCT") if result.get('success'): self.logger.critical(f"TRAILING STOP EXECUTED: {symbol} fermée avec succès") else: self.logger.error(f"TRAILING STOP FAILED: {symbol} - {result.get('error')}") except Exception as e: self.logger.error(f"Erreur trailing stop {symbol}: {e}") def profit_taker_loop(self): """Boucle principale du ProfitTaker""" self.logger.info("Démarrage de la boucle ProfitTaker...") while self.running: try: # Mettre à jour les positions actives self.update_active_positions() # Vérifier les conditions de stop loss self.check_stop_loss_conditions() # Vérifier les conditions de trailing stop self.check_trailing_stop_conditions() # Pause avant la prochaine itération time.sleep(self.check_interval) except Exception as e: self.logger.error(f"Erreur dans profit_taker_loop: {e}") time.sleep(5) def start(self): """Démarre le ProfitTaker""" if not self.running: self.running = True self.worker_thread = Thread(target=self.profit_taker_loop) self.worker_thread.daemon = True self.worker_thread.start() self.logger.info("ProfitTaker démarré") def stop(self): """Arrête le ProfitTaker""" self.running = False if hasattr(self, 'worker_thread'): self.worker_thread.join(timeout=5) self.logger.info("ProfitTaker arrêté") def get_positions_summary(self) -> Dict: """Retourne un résumé des positions actives""" with self.lock: return { 'total_positions': len(self.active_positions), 'positions': {key: { 'symbol': pos['symbol'], 'side': pos['side'], 'pnl_percent': pos['pnl_percent'], 'current_size': pos['current_size'] } for key, pos in self.active_positions.items()} }