#!/usr/bin/env python3 """ Position Tracker RPC - Utilise eth_getLogs directement Filtre les events OpenPosition/ClosePosition du contrat ApolloX """ import json from typing import Dict, List, Optional from dataclasses import dataclass, asdict from datetime import datetime from web3 import Web3 import logging import time logger = logging.getLogger(__name__) APOLLOX_V2_CONTRACT = "0x1b6F2d3844C6ae7D56ceb3C3643b9060ba28FEb0" # Rotation entre plusieurs RPC endpoints pour éviter rate limit BSC_RPC_URLS = [ "https://bsc-dataseed1.binance.org/", "https://bsc-dataseed2.binance.org/", "https://bsc-dataseed3.binance.org/", "https://bsc-dataseed4.binance.org/", "https://bsc-dataseed1.defibit.io/", "https://bsc-dataseed2.defibit.io/", "https://bsc.rpc.blxrbdn.com", ] BSC_RPC_URL = BSC_RPC_URLS[0] # Event signatures OPEN_POSITION_EVENT = '0x0a6da83417411689fd88436e7fa57a7cf1cf635a35194c0658314d4a037382af' CLOSE_POSITION_EVENT = '0x7064b82c073f138da0ec7646ebb51d4a0061d647accb4670bc564edf0bfac41d' @dataclass class Position: """Représente une position tradée""" trade_hash: str tx_hash: str wallet: str 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' closed_at: Optional[str] = None close_tx_hash: Optional[str] = None pnl: Optional[float] = None class PositionTrackerRPC: """ Tracker de positions basé sur eth_getLogs avec filtrage par events """ def __init__(self, wallet_address: str, rpc_url: str = BSC_RPC_URL): self.wallet_address = Web3.to_checksum_address(wallet_address) self.w3 = Web3(Web3.HTTPProvider(rpc_url)) self.contract_address = Web3.to_checksum_address(APOLLOX_V2_CONTRACT) self.positions: Dict[str, Position] = {} self.last_scanned_block = 0 # RPC rotation pour éviter rate limits self.rpc_urls = BSC_RPC_URLS self.current_rpc_index = 0 logger.info(f"📊 PositionTrackerRPC initialisé pour {wallet_address}") logger.info(f" {len(self.rpc_urls)} RPC endpoints disponibles") def _rotate_rpc(self): """Passe au prochain RPC endpoint""" self.current_rpc_index = (self.current_rpc_index + 1) % len(self.rpc_urls) new_rpc = self.rpc_urls[self.current_rpc_index] self.w3 = Web3(Web3.HTTPProvider(new_rpc)) logger.info(f" 🔄 Rotation RPC -> {new_rpc}") def scan_positions(self, from_block: int = 0, to_block: str = 'latest', chunk_size: int = 2000) -> int: """ Scanne les events OpenPosition/ClosePosition du contrat Args: from_block: Block de départ to_block: Block de fin ('latest' ou numéro) chunk_size: Taille des chunks (2000 recommandé pour BSC) Returns: Nombre de positions trouvées """ logger.info(f"🔍 Scan events depuis block {from_block}...") try: # Récupérer le block actuel if to_block == 'latest': to_block_num = self.w3.eth.block_number else: to_block_num = int(to_block) logger.info(f" Block actuel: {to_block_num}") logger.info(f" Range: {to_block_num - from_block} blocks") # Scanner par chunks current_block = from_block positions_found = 0 retry_delay = 0.5 while current_block <= to_block_num: end_block = min(current_block + chunk_size - 1, to_block_num) logger.info(f" 📦 Chunk {current_block} à {end_block}...") try: # Filtrer les events OpenPosition open_logs = self._get_event_logs( event_sig=OPEN_POSITION_EVENT, from_block=current_block, to_block=end_block ) # Filtrer les events ClosePosition close_logs = self._get_event_logs( event_sig=CLOSE_POSITION_EVENT, from_block=current_block, to_block=end_block ) # Traiter les events for log in open_logs: # Vérifier si la TX est de notre wallet if self._is_our_transaction(log['transactionHash'].hex()): self._process_open_position(log) positions_found += 1 for log in close_logs: if self._is_our_transaction(log['transactionHash'].hex()): self._process_close_position(log) # Succès: avancer current_block = end_block + 1 self.last_scanned_block = end_block # Petit délai pour éviter rate limit time.sleep(retry_delay) except Exception as e: error_msg = str(e).lower() if 'limit' in error_msg or 'rate' in error_msg: logger.warning(f" ⚠️ Rate limit détecté") # Rotation RPC self._rotate_rpc() time.sleep(retry_delay) retry_delay = min(retry_delay * 1.5, 3) # Réduire chunk size chunk_size = max(200, chunk_size // 2) continue else: logger.error(f" ❌ Erreur: {e}") # Réduire chunk et continuer chunk_size = max(200, chunk_size // 2) time.sleep(1) continue logger.info(f"✅ Scan terminé:") logger.info(f" Transactions analysées: {positions_found}") logger.info(f" Positions ouvertes: {self.get_open_positions_count()}") logger.info(f" Positions fermées: {self.get_closed_positions_count()}") return positions_found except Exception as e: logger.error(f"❌ Erreur scan: {e}") import traceback logger.error(traceback.format_exc()) return 0 def _get_event_logs(self, event_sig: str, from_block: int, to_block: int) -> List: """ Récupère les logs pour un event spécifique """ filter_params = { 'fromBlock': from_block, 'toBlock': to_block, 'address': self.contract_address, 'topics': [event_sig] # Topic 0 = event signature } return self.w3.eth.get_logs(filter_params) def _is_our_transaction(self, tx_hash: str) -> bool: """ Vérifie si une transaction vient de notre wallet """ try: tx = self.w3.eth.get_transaction(tx_hash) return tx['from'].lower() == self.wallet_address.lower() except Exception as e: logger.debug(f"Erreur check TX {tx_hash}: {e}") return False def _process_open_position(self, log): """Traite un event d'ouverture de position""" try: tx_hash = log['transactionHash'].hex() # Extraire le trade hash (Topic 1) trade_hash = log['topics'][1].hex() # Extraire la pair address (Topic 2) pair_address = '0x' + log['topics'][2].hex()[-40:] # Décoder les données data = log['data'] margin = 0 leverage = 0 entry_price = 0 qty = 0 is_long = True if len(data) >= 2: try: hex_data = data[2:] if data.startswith('0x') else data if len(hex_data) >= 64: param1 = int(hex_data[0:64], 16) margin = param1 / 1e18 if param1 > 0 else 0 except Exception: pass # Récupérer le block pour timestamp block = self.w3.eth.get_block(log['blockNumber']) timestamp = datetime.fromtimestamp(block['timestamp']).isoformat() position = Position( trade_hash=trade_hash, tx_hash=tx_hash, wallet=self.wallet_address, pair_address=pair_address, is_long=is_long, margin=margin, leverage=leverage, entry_price=entry_price, qty=qty, stop_loss=0, take_profit=0, opened_at=timestamp, block_number=log['blockNumber'], status='OPEN' ) self.positions[trade_hash] = position logger.info(f" ✅ Position {trade_hash[:16]}... @ block {log['blockNumber']}") except Exception as e: logger.error(f"Erreur process open position: {e}") def _process_close_position(self, log): """Traite un event de fermeture de position""" try: tx_hash = log['transactionHash'].hex() if len(log['topics']) >= 2: trade_hash = log['topics'][1].hex() if trade_hash in self.positions: position = self.positions[trade_hash] # Récupérer le block pour timestamp block = self.w3.eth.get_block(log['blockNumber']) timestamp = datetime.fromtimestamp(block['timestamp']).isoformat() position.status = 'CLOSED' position.closed_at = timestamp position.close_tx_hash = tx_hash logger.info(f" ✅ Position fermée {trade_hash[:16]}...") except Exception as e: logger.error(f"Erreur process close position: {e}") def get_open_positions(self) -> List[Position]: return [p for p in self.positions.values() if p.status == 'OPEN'] def get_closed_positions(self) -> List[Position]: return [p for p in self.positions.values() if p.status == 'CLOSED'] def get_open_positions_count(self) -> int: return len(self.get_open_positions()) def get_closed_positions_count(self) -> int: return len(self.get_closed_positions()) def get_position(self, trade_hash: str) -> Optional[Position]: return self.positions.get(trade_hash) def save_to_file(self, filepath: str): data = { 'wallet': self.wallet_address, 'last_scanned_block': self.last_scanned_block, 'positions': {k: asdict(v) for k, v in self.positions.items()} } with open(filepath, 'w') as f: json.dump(data, f, indent=2) logger.info(f"💾 Positions sauvegardées: {filepath}") def load_from_file(self, filepath: str): try: with open(filepath, 'r') as f: data = json.load(f) self.last_scanned_block = data.get('last_scanned_block', 0) for trade_hash, pos_dict in data.get('positions', {}).items(): self.positions[trade_hash] = Position(**pos_dict) logger.info(f"📂 {len(self.positions)} positions chargées (dernier block: {self.last_scanned_block})") except FileNotFoundError: logger.info(f"📂 Nouveau tracker (pas de cache)") except Exception as e: logger.error(f"❌ Erreur chargement: {e}") if __name__ == '__main__': import os from dotenv import load_dotenv logging.basicConfig( level=logging.INFO, format='%(levelname)s - %(message)s' ) load_dotenv() from web3 import Account private_key = os.getenv('WALLET_PRIVATE_KEY') account = Account.from_key(private_key) print("=" * 80) print("🔍 POSITION TRACKER RPC - SCAN VIA ETH_GETLOGS") print("=" * 80) print(f"\nWallet: {account.address}\n") tracker = PositionTrackerRPC(account.address) # Charger cache tracker.load_from_file('positions_cache.json') # Scanner depuis le dernier block (ou depuis début du bot ~block 43500000) start_block = tracker.last_scanned_block + 1 if tracker.last_scanned_block > 0 else 43500000 tracker.scan_positions(from_block=start_block, chunk_size=2000) # Sauvegarder tracker.save_to_file('positions_cache.json') # Afficher résumé print(f"\n" + "=" * 80) print("📊 RÉSUMÉ") print("=" * 80) open_positions = tracker.get_open_positions() closed_positions = tracker.get_closed_positions() print(f"\n💼 Positions ouvertes: {len(open_positions)}") for pos in open_positions[:20]: print(f" • {pos.trade_hash[:16]}... @ {pos.opened_at}") if pos.margin > 0: print(f" Margin: ${pos.margin:.2f}") print(f"\n📝 Positions fermées: {len(closed_positions)}") for pos in closed_positions[-10:]: print(f" • {pos.trade_hash[:16]}... @ {pos.opened_at}") print("\n" + "=" * 80)