#!/usr/bin/env python3 """ Position Tracker - Analyse les transactions on-chain pour tracker les positions Au lieu de se fier à getPositionsV2() qui n'existe pas, on scanne BSCScan """ 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__) # Event signatures (calculées avec keccak256) # OpenPosition event OPEN_POSITION_EVENT = '0x0a6da83417411689fd88436e7fa57a7cf1cf635a35194c0658314d4a037382af' # ClosePosition event (à déterminer) CLOSE_POSITION_EVENT = '0x7064b82c073f138da0ec7646ebb51d4a0061d647accb4670bc564edf0bfac41d' APOLLOX_V2_CONTRACT = "0x1b6F2d3844C6ae7D56ceb3C3643b9060ba28FEb0" # Plusieurs RPC endpoints BSC pour load balancing et fallback 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/", ] BSC_RPC_URL = BSC_RPC_URLS[0] # Default @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' # OPEN, CLOSED closed_at: Optional[str] = None close_tx_hash: Optional[str] = None pnl: Optional[float] = None class PositionTracker: """ Tracker de positions basé sur l'analyse des transactions on-chain """ 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] = {} # trade_hash -> Position self.last_scanned_block = 0 logger.info(f"📊 PositionTracker initialisé pour {wallet_address}") def scan_transactions(self, from_block: int = 0, to_block: str = 'latest') -> int: """ Scanne les transactions du wallet avec le contrat ApolloX Returns: Nombre de positions trouvées """ logger.info(f"🔍 Scan transactions depuis block {from_block} jusqu'à {to_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) # BSC limite strictement les requêtes - on utilise des petits chunks # On scanne par chunks de 500 blocks avec délais chunk_size = 500 current_block = from_block positions_found = 0 retry_delay = 1 # Délai initial entre requêtes while current_block <= to_block_num: end_block = min(current_block + chunk_size - 1, to_block_num) logger.info(f" Scanning blocks {current_block} to {end_block} (chunk: {chunk_size})...") # Filtrer les logs du contrat ApolloX filter_params = { 'fromBlock': current_block, 'toBlock': end_block, 'address': self.contract_address } try: logs = self.w3.eth.get_logs(filter_params) # Succès: réinitialiser le délai et augmenter légèrement le chunk retry_delay = 1 if chunk_size < 500: chunk_size = min(500, chunk_size + 50) except Exception as e: error_msg = str(e).lower() if 'limit exceeded' in error_msg or 'rate' in error_msg: # Rate limit: réduire drastiquement et attendre logger.warning(f" ⚠️ Rate limit détecté, pause {retry_delay}s...") time.sleep(retry_delay) chunk_size = max(100, chunk_size // 2) retry_delay = min(retry_delay * 2, 10) # Backoff exponentiel continue else: logger.warning(f" Erreur get_logs: {e}, retry avec chunk plus petit...") chunk_size = max(100, chunk_size // 2) time.sleep(1) continue # Grouper les logs par transaction tx_logs = {} for log in logs: tx_hash = log['transactionHash'].hex() if tx_hash not in tx_logs: tx_logs[tx_hash] = [] tx_logs[tx_hash].append(log) # Analyser chaque transaction for tx_hash, logs in tx_logs.items(): # Récupérer la transaction pour vérifier l'émetteur try: tx = self.w3.eth.get_transaction(tx_hash) # Seulement nos transactions if tx['from'].lower() != self.wallet_address.lower(): continue # Analyser les events self._process_transaction(tx_hash, tx, logs) positions_found += 1 except Exception as e: logger.warning(f"Erreur analyse TX {tx_hash}: {e}") continue current_block = end_block + 1 self.last_scanned_block = end_block # Petit délai entre chaque chunk pour éviter rate limit time.sleep(0.5) logger.info(f"✅ Scan terminé: {positions_found} transactions analysées") 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 transactions: {e}") import traceback logger.error(traceback.format_exc()) return 0 def _process_transaction(self, tx_hash: str, tx, logs: List): """Analyse une transaction et extrait les positions""" # Chercher l'event d'ouverture de position for log in logs: if len(log['topics']) < 2: continue event_sig = log['topics'][0].hex() # Event d'ouverture if event_sig == OPEN_POSITION_EVENT: self._process_open_position(tx_hash, tx, log, logs) # Event de fermeture elif event_sig == CLOSE_POSITION_EVENT: self._process_close_position(tx_hash, log) def _process_open_position(self, tx_hash: str, tx, open_log, all_logs): """Traite un event d'ouverture de position""" try: # Extraire le trade hash (Topic 1) trade_hash = open_log['topics'][1].hex() # Extraire la pair address (Topic 2) pair_address = '0x' + open_log['topics'][2].hex()[-40:] # Parser les données de l'event data = open_log['data'] # Décoder les données de l'event (ABI encoding) # Format approximatif: margin, qty, entryPrice, isLong, leverage, etc. # Les données sont en format uint256 (32 bytes chacun) margin = 0 leverage = 0 entry_price = 0 qty = 0 is_long = True if len(data) >= 2: # Au moins "0x" + data try: # Retirer le préfixe 0x hex_data = data[2:] if data.startswith('0x') else data # Chaque paramètre = 64 caractères hex (32 bytes) # On essaie de décoder les premiers paramètres if len(hex_data) >= 64: # Premier uint256 (possiblement margin ou qty) param1 = int(hex_data[0:64], 16) if len(hex_data) >= 128: # Deuxième uint256 param2 = int(hex_data[64:128], 16) if len(hex_data) >= 192: # Troisième uint256 param3 = int(hex_data[128:192], 16) # Les valeurs exactes dépendent de l'ABI du contrat # On stocke les valeurs brutes pour analyse if len(hex_data) >= 64: # Essayer de deviner margin (généralement en wei, diviser par 1e18) margin = param1 / 1e18 if param1 > 0 else 0 except Exception as decode_err: logger.debug(f"Erreur décodage data: {decode_err}") # Récupérer le block pour timestamp block = self.w3.eth.get_block(open_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=open_log['blockNumber'], status='OPEN' ) # Stocker la position self.positions[trade_hash] = position logger.info(f" ✅ Position ouverte: {trade_hash[:16]}... @ block {open_log['blockNumber']} (margin: ${margin:.2f})") except Exception as e: logger.error(f"Erreur process open position: {e}") import traceback logger.debug(traceback.format_exc()) def _process_close_position(self, tx_hash: str, close_log): """Traite un event de fermeture de position""" try: # Le trade hash devrait être dans les topics if len(close_log['topics']) >= 2: trade_hash = close_log['topics'][1].hex() # Trouver la position correspondante if trade_hash in self.positions: position = self.positions[trade_hash] # Récupérer le block pour timestamp block = self.w3.eth.get_block(close_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]}...") else: logger.warning(f" ⚠️ Position fermée non trouvée: {trade_hash[:16]}...") except Exception as e: logger.error(f"Erreur process close position: {e}") def get_open_positions(self) -> List[Position]: """Retourne toutes les positions ouvertes""" 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""" return [p for p in self.positions.values() if p.status == 'CLOSED'] def get_open_positions_count(self) -> int: """Nombre de positions ouvertes""" return len(self.get_open_positions()) def get_closed_positions_count(self) -> int: """Nombre de positions fermées""" return len(self.get_closed_positions()) def get_position(self, trade_hash: str) -> Optional[Position]: """Récupère une position par son trade hash""" return self.positions.get(trade_hash) def save_to_file(self, filepath: str): """Sauvegarde les positions dans un fichier JSON""" 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): """Charge les positions depuis un fichier JSON""" 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"📂 Positions chargées: {len(self.positions)} positions") except FileNotFoundError: logger.info(f"📂 Fichier non trouvé: {filepath} (nouveau tracker)") except Exception as e: logger.error(f"❌ Erreur chargement positions: {e}") if __name__ == '__main__': import os from dotenv import load_dotenv logging.basicConfig(level=logging.INFO) load_dotenv() # Récupérer l'adresse du wallet from web3 import Account private_key = os.getenv('WALLET_PRIVATE_KEY') account = Account.from_key(private_key) print("=" * 80) print("🔍 POSITION TRACKER - SCAN ON-CHAIN") print("=" * 80) print(f"\nWallet: {account.address}") tracker = PositionTracker(account.address) # Charger l'état précédent si existe tracker.load_from_file('positions_cache.json') # Scanner depuis le dernier block scanné (ou depuis 65509000 - nos premiers tests) start_block = tracker.last_scanned_block + 1 if tracker.last_scanned_block > 0 else 65509500 print(f"\n🔍 Scanning depuis block {start_block}...\n") tracker.scan_transactions(from_block=start_block) # 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: print(f" • {pos.trade_hash[:16]}... @ {pos.opened_at}") print(f"\n📝 Positions fermées: {len(closed_positions)}") for pos in closed_positions[-5:]: # Dernières 5 print(f" • {pos.trade_hash[:16]}... @ {pos.opened_at}") print("\n" + "=" * 80)