#!/usr/bin/env python3 """ Position Tracker V2 - Utilise BSCScan API pour lister les TX du wallet Plus efficace que de scanner tous les blocks """ import json import requests 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" BSC_RPC_URL = "https://bsc-dataseed.binance.org/" BSCSCAN_API = "https://api.bscscan.com/api" # 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 PositionTrackerV2: """ Tracker de positions basé sur l'API BSCScan Plus efficace que le scan block par block """ def __init__(self, wallet_address: str, bscscan_api_key: str = None, 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.bscscan_api_key = bscscan_api_key or "YourApiKeyToken" # Clé gratuite possible self.positions: Dict[str, Position] = {} self.last_scanned_block = 0 logger.info(f"📊 PositionTrackerV2 initialisé pour {wallet_address}") def fetch_wallet_transactions(self, start_block: int = 0) -> List[Dict]: """ Récupère toutes les transactions du wallet depuis BSCScan API Returns: Liste des transactions du wallet """ logger.info(f"🔍 Récupération transactions wallet depuis block {start_block}...") params = { 'module': 'account', 'action': 'txlist', 'address': self.wallet_address, 'startblock': start_block, 'endblock': 99999999, 'sort': 'asc', 'apikey': self.bscscan_api_key } try: response = requests.get(BSCSCAN_API, params=params, timeout=30) response.raise_for_status() data = response.json() if data['status'] == '1' and data['message'] == 'OK': txs = data['result'] logger.info(f" ✅ {len(txs)} transactions trouvées") return txs else: error_msg = data.get('message', 'Unknown') result_msg = data.get('result', '') logger.error(f" ❌ Erreur BSCScan API: {error_msg}") logger.error(f" Détails: {result_msg}") if 'Invalid API Key' in result_msg or 'API Key' in error_msg: logger.error(f" 💡 Vous devez obtenir une clé API BSCScan gratuite:") logger.error(f" 1. Créer compte sur https://bscscan.com/register") logger.error(f" 2. Aller dans API-KEYs: https://bscscan.com/myapikey") logger.error(f" 3. Ajouter dans .env: BSCSCAN_API_KEY=votre_clé") return [] except Exception as e: logger.error(f"❌ Erreur fetch transactions: {e}") return [] def filter_apollox_transactions(self, txs: List[Dict]) -> List[Dict]: """ Filtre les transactions qui interagissent avec le contrat ApolloX """ apollox_txs = [] for tx in txs: # Vérifier si la transaction est vers le contrat ApolloX if tx.get('to', '').lower() == self.contract_address.lower(): # Vérifier si la transaction a réussi if tx.get('txreceipt_status') == '1' or tx.get('isError') == '0': apollox_txs.append(tx) logger.info(f" 🎯 {len(apollox_txs)} transactions ApolloX trouvées") return apollox_txs def analyze_transactions(self, txs: List[Dict]) -> int: """ Analyse les transactions et extrait les positions Returns: Nombre de positions trouvées """ positions_found = 0 for tx in txs: tx_hash = tx['hash'] block_number = int(tx['blockNumber']) try: # Récupérer le receipt pour avoir les logs receipt = self.w3.eth.get_transaction_receipt(tx_hash) # Analyser les events dans les logs for log in receipt['logs']: if log['address'].lower() != self.contract_address.lower(): continue 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) positions_found += 1 # Event de fermeture elif event_sig == CLOSE_POSITION_EVENT: self._process_close_position(tx_hash, log) # Mettre à jour le dernier block scanné if block_number > self.last_scanned_block: self.last_scanned_block = block_number # Petit délai pour éviter de spam le RPC time.sleep(0.1) except Exception as e: logger.warning(f"Erreur analyse TX {tx_hash}: {e}") continue return positions_found def _process_open_position(self, tx_hash: str, tx: Dict, open_log): """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:] # Décoder les données data = open_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 as decode_err: logger.debug(f"Erreur décodage: {decode_err}") # Timestamp depuis la transaction BSCScan timestamp = datetime.fromtimestamp(int(tx['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=int(tx['blockNumber']), status='OPEN' ) self.positions[trade_hash] = position logger.info(f" ✅ Position ouverte: {trade_hash[:16]}... @ {timestamp}") except Exception as e: logger.error(f"Erreur process open position: {e}") def _process_close_position(self, tx_hash: str, close_log): """Traite un event de fermeture de position""" try: if len(close_log['topics']) >= 2: trade_hash = close_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(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]}...") except Exception as e: logger.error(f"Erreur process close position: {e}") def scan_positions(self, from_block: int = 0) -> int: """ Scanne les positions du wallet Returns: Nombre de positions trouvées """ logger.info(f"🔍 Scan positions depuis block {from_block}...") # 1. Récupérer toutes les transactions du wallet txs = self.fetch_wallet_transactions(start_block=from_block) if not txs: logger.warning("⚠️ Aucune transaction trouvée") return 0 # 2. Filtrer les transactions ApolloX apollox_txs = self.filter_apollox_transactions(txs) if not apollox_txs: logger.warning("⚠️ Aucune transaction ApolloX trouvée") return 0 # 3. Analyser les transactions positions_found = self.analyze_transactions(apollox_txs) logger.info(f"✅ Scan terminé:") logger.info(f" Positions ouvertes: {self.get_open_positions_count()}") logger.info(f" Positions fermées: {self.get_closed_positions_count()}") return positions_found 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 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"📂 Positions chargées: {len(self.positions)} positions") except FileNotFoundError: logger.info(f"📂 Fichier non trouvé: {filepath}") 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) load_dotenv() from web3 import Account private_key = os.getenv('WALLET_PRIVATE_KEY') account = Account.from_key(private_key) # Optionnel: clé API BSCScan gratuite bscscan_key = os.getenv('BSCSCAN_API_KEY', 'YourApiKeyToken') print("=" * 80) print("🔍 POSITION TRACKER V2 - SCAN VIA WALLET TX") print("=" * 80) print(f"\nWallet: {account.address}") tracker = PositionTrackerV2(account.address, bscscan_api_key=bscscan_key) # Charger cache tracker.load_from_file('positions_cache.json') # Scanner depuis le dernier block start_block = tracker.last_scanned_block + 1 if tracker.last_scanned_block > 0 else 43000000 print(f"\n🔍 Scanning depuis block {start_block}...\n") tracker.scan_positions(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[:10]: print(f" • {pos.trade_hash[:16]}... @ {pos.opened_at}") print(f" Margin: ${pos.margin:.2f}, Leverage: {pos.leverage}x") print(f"\n📝 Positions fermées: {len(closed_positions)}") for pos in closed_positions[-5:]: print(f" • {pos.trade_hash[:16]}... @ {pos.opened_at}") print("\n" + "=" * 80)