#!/usr/bin/env python3 """ Position Guardian - Système de surveillance indépendant pour les positions Hyperliquid Sécurité failsafe pour fermer automatiquement les positions selon les critères de stop loss """ import asyncio import time from datetime import datetime import logging from typing import Dict, Optional import sys import os # Add project path for imports sys.path.append(os.path.dirname(os.path.abspath(__file__))) from hyperliquid_client import hyperliquid_client # from real_time_trading_engine import RealTimeTradingEngine # Obsolète class PositionGuardian: def __init__(self, hyperliquid_client): self.client = hyperliquid_client # Initialize trading engine to use its closure methods self.trading_engine = RealTimeTradingEngine() self.trading_engine.set_trading_mode('LIVE_AUTO', hyperliquid_client) self.running = False self.position_peaks = {} # Track peak PnL for trailing stops self.partial_closures = {} # Track partial closures made self.check_interval = 30 # Check every 30 seconds # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s [GUARDIAN] %(levelname)s: %(message)s', handlers=[ logging.FileHandler('position_guardian.log'), logging.StreamHandler() ] ) self.logger = logging.getLogger(__name__) def start(self): """Démarrer le système de surveillance""" self.running = True self.logger.info("POSITION GUARDIAN STARTED - Failsafe protection active") def stop(self): """Arrêter le système de surveillance""" self.running = False self.logger.info("POSITION GUARDIAN STOPPED") def get_positions(self) -> list: """Récupérer les positions actuelles de Hyperliquid""" try: positions = self.client.get_positions() return positions if positions else [] except Exception as e: self.logger.error(f"Error fetching positions: {e}") return [] def calculate_pnl_percentage(self, position) -> float: """Calculer le PnL en pourcentage basé sur la marge""" try: raw_data = position.get('raw_position_data', {}) unrealized_pnl = float(raw_data.get('unrealizedPnl', 0)) # Get leverage information leverage_info = raw_data.get('leverage', {}) actual_leverage = 1 if isinstance(leverage_info, dict): actual_leverage = int(leverage_info.get('value', 1)) elif isinstance(leverage_info, (int, float)): actual_leverage = int(leverage_info) # Calculate margin used entry_price = position['entry_price'] size = abs(position['size']) notional_value = size * entry_price margin_used = notional_value / actual_leverage if actual_leverage > 0 else notional_value # Calculate PnL percentage pnl_percent = (unrealized_pnl / margin_used * 100) if margin_used > 0 else 0 return pnl_percent except Exception as e: self.logger.error(f"Error calculating PnL percentage: {e}") return 0.0 def should_close_position(self, position) -> tuple[bool, str, Optional[int]]: """ Déterminer si une position doit être fermée Returns: (should_close, reason, partial_percent) """ symbol = position['symbol'].replace('-PERP', '') side = position['side'] entry_price = position['entry_price'] pnl_percent = self.calculate_pnl_percentage(position) # Create unique key for position tracking position_key = f"{symbol}_{side}_{entry_price}" # Initialize tracking for this position if position_key not in self.position_peaks: self.position_peaks[position_key] = pnl_percent self.partial_closures[position_key] = { 'tp1_25pct': False, 'tp2_50pct': False } # Update peak PnL if pnl_percent > self.position_peaks[position_key]: self.position_peaks[position_key] = pnl_percent self.logger.info(f"Peak updated for {symbol}: {pnl_percent:.1f}%") peak_pnl = self.position_peaks[position_key] partial_closures = self.partial_closures[position_key] # CRITICAL: Hard stop loss at -30% (absolute priority) if pnl_percent <= -30: return True, f"EMERGENCY_STOP_LOSS_30PCT (PnL: {pnl_percent:.1f}%)", None # Take profit 1: 50% at +25% if pnl_percent >= 25 and not partial_closures['tp1_25pct']: partial_closures['tp1_25pct'] = True return True, f"TAKE_PROFIT_25PCT_50PERCENT (PnL: {pnl_percent:.1f}%)", 50 # Take profit 2: 25% at +50% (50% of remaining = 25% of original) if pnl_percent >= 50 and not partial_closures['tp2_50pct']: partial_closures['tp2_50pct'] = True return True, f"TAKE_PROFIT_50PCT_25PERCENT (PnL: {pnl_percent:.1f}%)", 50 # Trailing stop: -15% from peak if pnl_percent - peak_pnl <= -15: return True, f"TRAILING_STOP_15PCT (Peak: +{peak_pnl:.1f}%, Current: {pnl_percent:.1f}%)", None return False, "", None def close_position_full(self, symbol: str) -> bool: """Fermer complètement une position""" try: # Use the trading engine's existing closure method result = self.trading_engine.close_hyperliquid_position_by_symbol(symbol) if result.get('success'): self.logger.info(f"FULL CLOSURE SUCCESS: {symbol} - {result.get('message', '')}") return True else: self.logger.error(f"FULL CLOSURE FAILED: {symbol} - {result.get('error', 'Unknown error')}") return False except Exception as e: self.logger.error(f"FULL CLOSURE ERROR: {symbol} - {e}") return False def close_position_partial(self, symbol: str, size: float, percent: int) -> bool: """Fermer partiellement une position""" try: # Use the trading engine's existing partial closure method result = self.trading_engine.close_hyperliquid_position_partial(symbol, size) if result.get('success'): self.logger.info(f"PARTIAL CLOSURE SUCCESS: {symbol} - {percent}% closed") return True else: self.logger.error(f"PARTIAL CLOSURE FAILED: {symbol} - {result.get('error', 'Unknown error')}") return False except Exception as e: self.logger.error(f"PARTIAL CLOSURE ERROR: {symbol} - {e}") return False async def monitor_positions(self): """Boucle principale de surveillance""" while self.running: try: positions = self.get_positions() if not positions: await asyncio.sleep(self.check_interval) continue self.logger.info(f"Monitoring {len(positions)} positions...") for position in positions: symbol = position['symbol'].replace('-PERP', '') side = position['side'] size = abs(position['size']) pnl_percent = self.calculate_pnl_percentage(position) should_close, reason, partial_percent = self.should_close_position(position) if should_close: self.logger.warning(f"CLOSURE TRIGGERED: {symbol} {side} - {reason}") if partial_percent: # Partial closure quantity_to_close = size * (partial_percent / 100.0) success = self.close_position_partial(symbol, quantity_to_close, partial_percent) else: # Full closure success = self.close_position_full(symbol) # Clean up tracking for closed position if success: position_key = f"{symbol}_{side}_{position['entry_price']}" if position_key in self.position_peaks: del self.position_peaks[position_key] if position_key in self.partial_closures: del self.partial_closures[position_key] if not success: self.logger.critical(f"CRITICAL: Failed to close {symbol} - Manual intervention required!") else: # Log current status for monitoring peak = self.position_peaks.get(f"{symbol}_{side}_{position['entry_price']}", 0) self.logger.debug(f"{symbol}: PnL {pnl_percent:.1f}%, Peak {peak:.1f}%") except Exception as e: self.logger.error(f"Error in monitoring loop: {e}") await asyncio.sleep(self.check_interval) async def run(self): """Point d'entrée principal""" self.start() try: await self.monitor_positions() except KeyboardInterrupt: self.logger.info("Guardian stopped by user") finally: self.stop() async def main(): """Fonction principale""" # Initialize Hyperliquid client try: guardian = PositionGuardian(hyperliquid_client) await guardian.run() except Exception as e: print(f"Failed to start Position Guardian: {e}") logging.error(f"Failed to start Position Guardian: {e}") if __name__ == "__main__": print("Position Guardian - Independent Position Monitoring System") print("This system will monitor and automatically close positions according to stop loss rules") print("Press Ctrl+C to stop") asyncio.run(main())