#!/usr/bin/env python3 """ Import rapide avec Etherscan V2 API Un seul call pour récupérer toutes les TX du wallet """ import requests import json import os from dotenv import load_dotenv from web3 import Web3, Account from position_manager import PositionManager, Position from datetime import datetime import time load_dotenv() # Config WALLET_ADDRESS = "0x7476737C49d55bea691e54c2c96fD39a6bAf3049" API_KEY = os.getenv('BSCSCAN_API_KEY') APOLLOX_CONTRACT = "0x1b6F2d3844C6ae7D56ceb3C3643b9060ba28FEb0" # BSC RPC pour récupérer les receipts BSC_RPC = "https://bsc-dataseed.binance.org/" w3 = Web3(Web3.HTTPProvider(BSC_RPC)) # Event signatures OPEN_EVENT = '0x0a6da83417411689fd88436e7fa57a7cf1cf635a35194c0658314d4a037382af' CLOSE_EVENT = '0x7064b82c073f138da0ec7646ebb51d4a0061d647accb4670bc564edf0bfac41d' def get_all_transactions(): """Récupère toutes les TX du wallet via API Etherscan V2""" # Etherscan V2 API - unified endpoint for all chains url = "https://api.etherscan.io/v2/api" params = { 'chainid': '56', # BSC chain 'module': 'account', 'action': 'txlist', 'address': WALLET_ADDRESS, 'startblock': 65508000, # Ce matin 'endblock': 99999999, 'sort': 'asc', 'apikey': API_KEY } print(f"📡 Appel API Etherscan V2 (chainid=56)...") try: response = requests.get(url, params=params, timeout=15) if response.status_code != 200: print(f" ❌ HTTP {response.status_code}") return [] data = response.json() print(f" Status: {data.get('status')}") print(f" Message: {data.get('message')}") if data.get('status') == '1' and isinstance(data.get('result'), list): txs = data['result'] print(f" ✅ {len(txs)} transactions récupérées") return txs else: print(f" ❌ Erreur: {data.get('result', 'Unknown')}") return [] except Exception as e: print(f" ❌ Exception: {e}") return [] def filter_apollox_txs(txs): """Filtre les TX vers ApolloX""" opens = [] closes = [] for tx in txs: to_addr = tx.get('to', '').lower() from_addr = tx.get('from', '').lower() # Transactions vers ApolloX depuis notre wallet if to_addr == APOLLOX_CONTRACT.lower() and from_addr == WALLET_ADDRESS.lower(): # Check method func_name = tx.get('functionName', '') if 'openMarketTrade' in func_name or tx.get('methodId') == '0x': opens.append(tx) elif 'closeTrade' in func_name or 'executeTpSlOrLiq' in func_name: closes.append(tx) print(f" 🎯 Trouvé: {len(opens)} opens, {len(closes)} closes") return opens, closes def get_trade_hash_from_receipt(tx_hash): """Extrait le trade hash depuis le receipt""" try: receipt = w3.eth.get_transaction_receipt(tx_hash) for log in receipt['logs']: if log['address'].lower() != APOLLOX_CONTRACT.lower(): continue topics = log['topics'] if len(topics) < 2: continue event_sig = topics[0].hex().replace('0x', '').lower() target_sig = OPEN_EVENT.replace('0x', '').lower() if event_sig == target_sig: trade_hash = topics[1].hex() pair_address = '0x' + topics[2].hex()[-40:] return trade_hash, pair_address return None, None except Exception as e: print(f" ⚠️ Erreur receipt: {e}") return None, None def import_positions(): """Import toutes les positions""" print("=" * 80) print("📥 IMPORT POSITIONS - API V2") print("=" * 80) print() # Init position manager private_key = os.getenv('WALLET_PRIVATE_KEY') account = Account.from_key(private_key) manager = PositionManager(account.address, 'positions.json') # 1. Récupérer toutes les TX (1 seul call API) all_txs = get_all_transactions() if not all_txs: print("\n❌ Aucune transaction récupérée") return 0 # 2. Filtrer les TX ApolloX opens, closes = filter_apollox_txs(all_txs) if not opens: print("\n⚠️ Aucune position trouvée") return 0 # 3. Importer chaque position print(f"\n📦 Import de {len(opens)} positions...") imported = 0 for open_tx in opens: tx_hash = open_tx['hash'] # Extraire trade hash via receipt (RPC call) print(f"\n 🔍 {tx_hash[:16]}...") trade_hash, pair_address = get_trade_hash_from_receipt(tx_hash) if not trade_hash: print(f" ⏭️ Skip: no trade hash") continue # Check si déjà importée if manager.get_position(trade_hash): print(f" ⏭️ Already imported") continue # Parse les données value = float(open_tx.get('value', 0)) / 1e18 gas_used = int(open_tx.get('gasUsed', 0)) # Utiliser isoformat() pour format ISO cohérent (YYYY-MM-DDTHH:MM:SS) timestamp = datetime.fromtimestamp(int(open_tx['timeStamp'])).isoformat() block = int(open_tx['blockNumber']) # Créer la position position = Position( trade_hash=trade_hash, tx_hash=tx_hash, symbol='BTCUSD_DEGEN', pair_address=pair_address or '0xD89aE81e55b1Bd46591CB5555209fdF8fFb43b76', is_long=True, margin=value if value > 0 else 25.0, # Fallback leverage=200, entry_price=0, qty=0, stop_loss=0, take_profit=0, opened_at=timestamp, block_number=block, status='OPEN' ) manager.positions[trade_hash] = position manager.save() print(f" ✅ Imported: ${position.margin:.2f} @ {timestamp}") imported += 1 # Small delay time.sleep(0.1) # 4. Traiter les fermetures if closes: print(f"\n🔴 Traitement de {len(closes)} fermetures...") closed_count = 0 for close_tx in closes: tx_hash = close_tx['hash'] print(f"\n 🔍 {tx_hash[:16]}...") # Extraire trade hash depuis le receipt try: receipt = w3.eth.get_transaction_receipt(tx_hash) trade_hash = None for log in receipt['logs']: if log['address'].lower() == APOLLOX_CONTRACT.lower(): topics = log['topics'] if len(topics) >= 2: # Le trade hash est souvent dans topic[1] ou topic[2] potential_hash = topics[1].hex() if len(topics) > 1 else None if potential_hash and potential_hash != '0x' + '0'*64: trade_hash = potential_hash break if not trade_hash: print(f" ⏭️ Skip: no trade hash") continue # Chercher la position correspondante position = manager.get_position(trade_hash) if not position: print(f" ⏭️ Skip: position inconnue") continue if position.status != 'OPEN': print(f" ⏭️ Skip: déjà fermée") continue # Mettre à jour la position timestamp = datetime.fromtimestamp(int(close_tx['timeStamp'])).isoformat() position.status = 'CLOSED' position.closed_at = timestamp position.close_tx_hash = tx_hash # Déterminer la raison func_name = close_tx.get('functionName', '') if 'liq' in func_name.lower(): position.status = 'LIQUIDATED' position.close_reason = 'LIQUIDATION' elif 'tp' in func_name.lower(): position.close_reason = 'TAKE_PROFIT' elif 'sl' in func_name.lower(): position.close_reason = 'STOP_LOSS' else: position.close_reason = 'MANUAL' manager.save() print(f" ✅ Fermée: {position.status} @ {timestamp}") closed_count += 1 except Exception as e: print(f" ⚠️ Erreur: {e}") continue time.sleep(0.1) print(f"\n ✅ {closed_count} positions fermées") print() print("=" * 80) print(f"✅ IMPORT TERMINÉ: {imported} ouvertures, {closed_count if closes else 0} fermetures") print("=" * 80) print() print("💡 Voir sur: http://localhost:8889") return imported if __name__ == '__main__': import_positions()