#!/usr/bin/env python3 """ Position Manager - Tracking local des positions en temps réel Sauvegarde les positions au fur et à mesure que le bot trade """ import json import os from typing import Dict, List, Optional from dataclasses import dataclass, asdict from datetime import datetime import logging from threading import RLock logger = logging.getLogger(__name__) @dataclass class Position: """Position tradée""" trade_hash: str tx_hash: str symbol: str # BTCUSD_DEGEN, ETH, etc. pair_address: str is_long: bool margin: float leverage: int entry_price: float qty: float stop_loss: float take_profit: float opened_at: str block_number: int status: str = 'OPEN' # OPEN, CLOSED, LIQUIDATED closed_at: Optional[str] = None close_tx_hash: Optional[str] = None exit_price: Optional[float] = None pnl: Optional[float] = None pnl_pct: Optional[float] = None close_reason: Optional[str] = None # Raison de fermeture (LIQUIDATED, TP, SL, MANUAL, etc.) class PositionManager: """ Gestionnaire de positions local Enregistre et suit les positions en temps réel """ def __init__(self, wallet_address: str, storage_file: str = 'positions.json'): self.wallet_address = wallet_address self.storage_file = storage_file self.positions: Dict[str, Position] = {} self.lock = RLock() # Thread-safe (Reentrant lock pour éviter deadlock) # Charger les positions existantes self.load() logger.info(f"📊 PositionManager initialisé: {len(self.positions)} positions chargées") def add_position(self, trade_hash: str, tx_hash: str, symbol: str, pair_address: str, is_long: bool, margin: float, leverage: int, entry_price: float, qty: float, stop_loss: float = 0, take_profit: float = 0, block_number: int = 0) -> Position: """ Ajoute une nouvelle position ouverte Args: trade_hash: Hash unique de la position (depuis l'event) tx_hash: Hash de la transaction symbol: Symbole tradé (BTCUSD_DEGEN, ETH, etc.) pair_address: Adresse du pair token is_long: True si long, False si short margin: Montant de margin en USDT leverage: Leverage utilisé entry_price: Prix d'entrée qty: Quantité stop_loss: Prix de stop loss take_profit: Prix de take profit block_number: Numéro du block Returns: Position créée """ with self.lock: position = Position( trade_hash=trade_hash, tx_hash=tx_hash, symbol=symbol, pair_address=pair_address, is_long=is_long, margin=margin, leverage=leverage, entry_price=entry_price, qty=qty, stop_loss=stop_loss, take_profit=take_profit, opened_at=datetime.now().isoformat(), block_number=block_number, status='OPEN' ) self.positions[trade_hash] = position self.save() logger.info(f"✅ Position ajoutée: {symbol} {'LONG' if is_long else 'SHORT'} " f"{leverage}x (margin: ${margin:.2f})") logger.info(f" Trade hash: {trade_hash[:16]}...") return position def close_position(self, trade_hash: str, close_tx_hash: str, exit_price: float, pnl: float = None, block_number: int = 0) -> Optional[Position]: """ Ferme une position Args: trade_hash: Hash de la position à fermer close_tx_hash: Hash de la transaction de fermeture exit_price: Prix de sortie pnl: PnL réalisé (optionnel, sera calculé si non fourni) block_number: Numéro du block Returns: Position mise à jour ou None si non trouvée """ with self.lock: if trade_hash not in self.positions: logger.warning(f"⚠️ Position non trouvée pour fermeture: {trade_hash[:16]}...") return None position = self.positions[trade_hash] # Calculer PnL si non fourni if pnl is None and position.entry_price > 0: price_change = exit_price - position.entry_price if not position.is_long: price_change = -price_change pnl = (price_change / position.entry_price) * position.margin * position.leverage # Calculer PnL % pnl_pct = (pnl / position.margin * 100) if position.margin > 0 else 0 position.status = 'CLOSED' position.closed_at = datetime.now().isoformat() position.close_tx_hash = close_tx_hash position.exit_price = exit_price position.pnl = pnl position.pnl_pct = pnl_pct self.save() logger.info(f"✅ Position fermée: {position.symbol} {'LONG' if position.is_long else 'SHORT'}") logger.info(f" Entry: ${position.entry_price:.2f} → Exit: ${exit_price:.2f}") logger.info(f" PnL: ${pnl:.2f} ({pnl_pct:+.2f}%)") return position def get_open_positions(self) -> List[Position]: """Retourne toutes les positions ouvertes""" with self.lock: return [p for p in self.positions.values() if p.status == 'OPEN'] def get_closed_positions(self) -> List[Position]: """Retourne toutes les positions fermées""" with self.lock: return [p for p in self.positions.values() if p.status == 'CLOSED'] def get_position(self, trade_hash: str) -> Optional[Position]: """Récupère une position par son trade hash""" with self.lock: return self.positions.get(trade_hash) def get_all_positions(self) -> List[Position]: """Retourne toutes les positions""" with self.lock: return list(self.positions.values()) def get_stats(self) -> Dict: """Retourne les statistiques des positions""" with self.lock: open_pos = self.get_open_positions() closed_pos = self.get_closed_positions() total_open_margin = sum(p.margin for p in open_pos) winning_trades = [p for p in closed_pos if p.pnl and p.pnl > 0] losing_trades = [p for p in closed_pos if p.pnl and p.pnl <= 0] total_pnl = sum(p.pnl for p in closed_pos if p.pnl) avg_pnl = total_pnl / len(closed_pos) if closed_pos else 0 win_rate = (len(winning_trades) / len(closed_pos) * 100) if closed_pos else 0 return { 'open_positions': len(open_pos), 'closed_positions': len(closed_pos), 'total_positions': len(self.positions), 'total_open_margin': total_open_margin, 'winning_trades': len(winning_trades), 'losing_trades': len(losing_trades), 'win_rate': win_rate, 'total_pnl': total_pnl, 'avg_pnl': avg_pnl } def save(self): """Sauvegarde les positions dans le fichier JSON""" try: with self.lock: data = { 'wallet': self.wallet_address, 'last_updated': datetime.now().isoformat(), 'positions': {k: asdict(v) for k, v in self.positions.items()} } # Sauvegarde atomique (write + rename) temp_file = self.storage_file + '.tmp' with open(temp_file, 'w') as f: json.dump(data, f, indent=2) os.replace(temp_file, self.storage_file) except Exception as e: logger.error(f"❌ Erreur sauvegarde positions: {e}") def load(self): """Charge les positions depuis le fichier JSON""" try: if not os.path.exists(self.storage_file): logger.info(f"📂 Nouveau fichier de positions: {self.storage_file}") return with open(self.storage_file, 'r') as f: data = json.load(f) self.positions = {} for trade_hash, pos_dict in data.get('positions', {}).items(): self.positions[trade_hash] = Position(**pos_dict) logger.info(f"📂 Positions chargées depuis {self.storage_file}") except Exception as e: logger.error(f"❌ Erreur chargement positions: {e}") self.positions = {} # Singleton global pour faciliter l'accès _position_manager_instance = None def get_position_manager(wallet_address: str = None, storage_file: str = 'positions.json') -> PositionManager: """ Retourne l'instance singleton du PositionManager Args: wallet_address: Adresse du wallet (requis au premier appel) storage_file: Fichier de stockage Returns: Instance du PositionManager """ global _position_manager_instance if _position_manager_instance is None: if wallet_address is None: raise ValueError("wallet_address requis pour initialiser PositionManager") _position_manager_instance = PositionManager(wallet_address, storage_file) return _position_manager_instance if __name__ == '__main__': # Test logging.basicConfig(level=logging.INFO) manager = PositionManager('0x7476737C49d55bea691e54c2c96fD39a6bAf3049', 'test_positions.json') # Ajouter une position test pos = manager.add_position( trade_hash='0xabc123...', tx_hash='0xdef456...', symbol='BTCUSD_DEGEN', pair_address='0xD89aE81e55b1Bd46591CB5555209fdF8fFb43b76', is_long=True, margin=25.0, leverage=200, entry_price=95000.0, qty=0.5, stop_loss=94000.0, take_profit=96000.0 ) print("\n📊 Stats:") stats = manager.get_stats() for key, value in stats.items(): print(f" {key}: {value}") print("\n💼 Positions ouvertes:") for p in manager.get_open_positions(): print(f" • {p.symbol} {p.leverage}x {'LONG' if p.is_long else 'SHORT'} @ ${p.entry_price}")