#!/usr/bin/env python3 """ Importe les positions depuis les events blockchain SOLUTION AUTOMATIQUE - Pas de copier-coller! """ from web3 import Web3 import json from datetime import datetime from typing import Dict, List import os from dotenv import load_dotenv load_dotenv() WALLET_ADDRESS = os.getenv('WALLET_ADDRESS', '0x7476737C49d55bea691e54c2c96fD39a6bAf3049') APOLLOX_CONTRACT = "0x1b6F2d3844C6ae7D56ceb3C3643b9060ba28FEb0" BSC_RPC = "https://bsc-dataseed.binance.org/" w3 = Web3(Web3.HTTPProvider(BSC_RPC)) # Event signatures trouvées OPEN_TRADE_EVENT = "0x7064b82c073f138da0ec7646ebb51d4a0061d647accb4670bc564edf0bfac41d" # Chercher events de fermeture (à trouver) # Ces signatures sont à vérifier: POSSIBLE_CLOSE_EVENTS = [ "0x0a6da83417411689fd88436e7fa57a7cf1cf635a35194c0658314d4a037382af", # Trouvé dans TX ] def get_block_range(): """Détermine la plage de blocks à scanner""" current_block = w3.eth.block_number # Scanner les 1M derniers blocks (~1 mois sur BSC) from_block = current_block - 1_000_000 print(f"📊 Range de scan:") print(f" From block: {from_block:,}") print(f" To block: {current_block:,}") print(f" Total: {current_block - from_block:,} blocks (~1 mois)\n") return from_block, current_block def scan_open_events(from_block: int, to_block: int) -> List[dict]: """Scanne les events d'ouverture""" print("🔍 Scan des events OpenTrade...") # Filtrer par wallet address (topic[1] dans OPEN_TRADE_EVENT) wallet_topic = '0x' + WALLET_ADDRESS[2:].lower().zfill(64) try: # Scanner par chunks de 5000 blocks (limite BSC RPC) chunk_size = 5000 all_logs = [] for start in range(from_block, to_block, chunk_size): end = min(start + chunk_size - 1, to_block) print(f" Scanning blocks {start:,} - {end:,}...") try: logs = w3.eth.get_logs({ 'address': Web3.to_checksum_address(APOLLOX_CONTRACT), 'topics': [OPEN_TRADE_EVENT, wallet_topic], 'fromBlock': start, 'toBlock': end }) if logs: print(f" ✅ {len(logs)} events trouvés") all_logs.extend(logs) except Exception as e: print(f" ⚠️ Erreur: {e}") continue print(f"\n✅ Total OpenTrade events: {len(all_logs)}\n") return all_logs except Exception as e: print(f"❌ Erreur: {e}") return [] def parse_open_event(log: dict) -> dict: """Parse un event OpenTrade""" # Topics topics = log['topics'] wallet = topics[1].hex() # Wallet address # topic[2] pourrait être autre chose # Data (288 bytes = 9 slots de 32 bytes) data = log['data'] if isinstance(data, bytes): data_hex = data.hex() else: data_hex = data[2:] if data.startswith('0x') else data # Parser les slots slots = [] for i in range(0, min(len(data_hex), 576), 64): # 576 = 9 * 64 slot_hex = data_hex[i:i+64] if len(slot_hex) == 64: slots.append(int(slot_hex, 16)) if len(slots) < 4: return None # Extraction des données # Slot 0: pair address # Slot 1: is_long (0 ou 1) # Slot 2: token address # Slot 3: margin (wei) # Slot 4: leverage ou price pair_address = f"0x{slots[0]:040x}" if len(slots) > 0 else None is_long = bool(slots[1]) if len(slots) > 1 else None token_address = f"0x{slots[2]:040x}" if len(slots) > 2 else None margin_wei = slots[3] if len(slots) > 3 else 0 margin = margin_wei / 1e18 # Récupérer block pour timestamp block_number = log['blockNumber'] block = w3.eth.get_block(block_number) timestamp = block['timestamp'] opened_at = datetime.fromtimestamp(timestamp).isoformat() # Récupérer TX hash tx_hash = log['transactionHash'].hex() # Trade hash depuis les autres events de cette TX receipt = w3.eth.get_transaction_receipt(tx_hash) trade_hash = None for other_log in receipt['logs']: if other_log['address'].lower() == APOLLOX_CONTRACT.lower(): if len(other_log['topics']) > 1: # L'event avec le trade_hash comme topic[1] potential_hash = other_log['topics'][1].hex() if potential_hash != wallet_topic: trade_hash = potential_hash[2:] # Retirer 0x break if not trade_hash: trade_hash = tx_hash[2:] # Fallback return { 'trade_hash': trade_hash, 'tx_hash': tx_hash, 'symbol': 'BTCUSD_DEGEN', # TODO: déterminer depuis pair_address 'pair_address': pair_address, 'is_long': is_long, 'margin': margin, 'leverage': 0, # À calculer 'entry_price': 0, # À récupérer 'qty': 0, 'stop_loss': 0, 'take_profit': 0, 'opened_at': opened_at, 'block_number': block_number, 'status': 'OPEN', # Par défaut, sera mis à jour si fermé 'closed_at': None, 'close_tx_hash': None, 'exit_price': None, 'pnl': None, 'pnl_pct': None, 'close_reason': None, 'token_address': token_address, } def main(): print("=" * 80) print("📥 IMPORT POSITIONS DEPUIS BLOCKCHAIN EVENTS") print("=" * 80) print() print(f"Wallet: {WALLET_ADDRESS}") print(f"Contract: {APOLLOX_CONTRACT}\n") # 1. Déterminer range from_block, to_block = get_block_range() # 2. Scanner events OpenTrade logs = scan_open_events(from_block, to_block) if not logs: print("❌ Aucun event trouvé") return # 3. Parser les positions print("📊 Parsing des positions...\n") positions = {} for i, log in enumerate(logs, 1): print(f" [{i}/{len(logs)}] Parsing position...") position = parse_open_event(log) if position: trade_hash = position['trade_hash'] positions[trade_hash] = position print(f" ✅ {position['symbol']}") print(f" Margin: ${position['margin']:.2f}") print(f" Opened: {position['opened_at']}") print(f" Block: {position['block_number']}") else: print(f" ⚠️ Erreur parsing") print() # 4. Sauvegarder print("=" * 80) print("💾 SAUVEGARDE") print("=" * 80) print() data = { 'positions': positions, 'last_updated': datetime.now().isoformat(), 'source': 'blockchain_events', 'scan_range': { 'from_block': from_block, 'to_block': to_block, } } with open('positions.json', 'w') as f: json.dump(data, f, indent=2) print(f"✅ {len(positions)} positions sauvegardées\n") # 5. Stats print("=" * 80) print("📊 RÉSUMÉ") print("=" * 80) total_margin = sum(p['margin'] for p in positions.values()) print(f"Total positions: {len(positions)}") print(f"Total margin: ${total_margin:.2f}") print(f"\nNote: Les positions sont toutes marquées OPEN") print(f" Scanner les events de fermeture pour mettre à jour le statut") print() print("✅ Dashboard: http://localhost:8889") if __name__ == '__main__': main()