import sys import json import requests import time from web3 import Web3 from web3._utils.events import get_event_data API_KEY = "6db42aeb3b1e4572bb72d1ae305e2e0b" RPC_URL = f"https://mainnet.infura.io/v3/{API_KEY}" w3 = Web3(Web3.HTTPProvider(RPC_URL)) PAIR_ABI = json.loads(""" [ { "anonymous": false, "inputs": [ {"indexed": true, "internalType": "address", "name": "sender", "type": "address"}, {"indexed": false, "internalType": "uint256", "name": "amount0In", "type": "uint256"}, {"indexed": false, "internalType": "uint256", "name": "amount1In", "type": "uint256"}, {"indexed": false, "internalType": "uint256", "name": "amount0Out", "type": "uint256"}, {"indexed": false, "internalType": "uint256", "name": "amount1Out", "type": "uint256"}, {"indexed": true, "internalType": "address", "name": "to", "type": "address"} ], "name": "Swap", "type": "event" }, {"constant":true,"inputs":[],"name":"token0","outputs":[{"name":"","type":"address"}],"stateMutability":"view","type":"function"}, {"constant":true,"inputs":[],"name":"token1","outputs":[{"name":"","type":"address"}],"stateMutability":"view","type":"function"} ] """) ERC20_ABI = json.loads(""" [ {"constant":true,"inputs":[],"name":"symbol","outputs":[{"name":"","type":"string"}],"stateMutability":"view","type":"function"}, {"constant":true,"inputs":[],"name":"decimals","outputs":[{"name":"","type":"uint8"}],"stateMutability":"view","type":"function"} ] """) def get_eth_price_usd(): resp = requests.get("https://api.coingecko.com/api/v3/simple/price", params={ "ids": "ethereum", "vs_currencies": "usd" }) return resp.json()["ethereum"]["usd"] def get_logs_with_auto_split(w3, params, chunk_size=500): from_block = params["fromBlock"] to_block = params["toBlock"] all_logs = [] queue = [(from_block, to_block)] while queue: start, end = queue.pop(0) p = dict(params) p["fromBlock"] = hex(start) p["toBlock"] = hex(end) retries = 0 while True: try: logs = w3.eth.get_logs(p) all_logs.extend(logs) print(f" Bloques {start}-{end}: {len(logs)} logs.") break except Exception as e: err = e.args[0] if isinstance(err, dict): code = err.get("code") if code == -32005: if start == end: print(f" Advertencia: bloque {start} individual aun asi excede el limite.") else: mid = (start + end) // 2 print(f" Demasiados resultados, subdividiendo rango {start}-{end} en {start}-{mid} y {mid+1}-{end}") queue.insert(0, (mid + 1, end)) queue.insert(0, (start, mid)) break elif "Too Many Requests" in err.get("message", ""): retries += 1 wait_time = min(2 ** retries, 60) print(f" Rate limit alcanzado, reintentando en {wait_time} segundos...") time.sleep(wait_time) else: print(f" Error al consultar bloques {start}-{end}: {e}") break else: print(f" Error al consultar bloques {start}-{end}: {e}") break time.sleep(0.2) return all_logs def main(): if len(sys.argv) < 2: print("Uso: python TransactionsWallet.py DIRECCION_CARTERA") sys.exit(1) raw_address = sys.argv[1] if raw_address.startswith("0x"): raw_address = raw_address[2:] topic_padded = "0x" + "0"*24 + raw_address.lower() print(f"Topic sender/to construido: {topic_padded}") eth_price = get_eth_price_usd() print(f"ETH/USD: ${eth_price}") latest_block = w3.eth.block_number start_block = latest_block - 5000 block_step = 2000 swap_topic = Web3.to_hex(w3.keccak(text="Swap(address,uint256,uint256,uint256,uint256,address)")) print(f"Swap topic: {swap_topic}") all_logs = [] current_block = start_block while current_block <= latest_block: end_block = min(current_block + block_step - 1, latest_block) print(f"\nConsultando bloques {current_block}-{end_block} (sender)...") logs_sender = get_logs_with_auto_split(w3, { "fromBlock": current_block, "toBlock": end_block, "topics": [ swap_topic, topic_padded ] }) print(f"Consultando bloques {current_block}-{end_block} (to)...") logs_to = get_logs_with_auto_split(w3, { "fromBlock": current_block, "toBlock": end_block, "topics": [ swap_topic, None, topic_padded ] }) all_logs.extend(logs_sender) all_logs.extend(logs_to) current_block = end_block + 1 print(f"\nTotal swaps encontrados: {len(all_logs)}") token_cache = {} profit_usd_total = 0 invested_usd = 0 timestamps = [] swap_event_abi = [entry for entry in PAIR_ABI if entry.get("name") == "Swap"][0] for log in all_logs: try: event_data = get_event_data(w3.codec, swap_event_abi, log) args = event_data["args"] sender = args["sender"] to = args["to"] amount0In = args["amount0In"] amount1In = args["amount1In"] amount0Out = args["amount0Out"] amount1Out = args["amount1Out"] pair_contract = w3.eth.contract(address=log["address"], abi=PAIR_ABI) token0 = pair_contract.functions.token0().call() token1 = pair_contract.functions.token1().call() if token0 in token_cache: symbol0, decimals0 = token_cache[token0] else: erc0 = w3.eth.contract(address=token0, abi=ERC20_ABI) symbol0 = erc0.functions.symbol().call() decimals0 = erc0.functions.decimals().call() token_cache[token0] = (symbol0, decimals0) if token1 in token_cache: symbol1, decimals1 = token_cache[token1] else: erc1 = w3.eth.contract(address=token1, abi=ERC20_ABI) symbol1 = erc1.functions.symbol().call() decimals1 = erc1.functions.decimals().call() token_cache[token1] = (symbol1, decimals1) amount0InNorm = amount0In / 10**decimals0 amount1InNorm = amount1In / 10**decimals1 amount0OutNorm = amount0Out / 10**decimals0 amount1OutNorm = amount1Out / 10**decimals1 ethValue = 0 base_symbol = None if symbol0 in ("WETH", "ETH"): base_symbol = symbol0 ethValue = amount0InNorm if amount0InNorm > 0 else amount0OutNorm elif symbol1 in ("WETH", "ETH"): base_symbol = symbol1 ethValue = amount1InNorm if amount1InNorm > 0 else amount1OutNorm else: base_symbol = "N/A" direction = "" if ethValue > 0: if (symbol0 in ("WETH", "ETH") and amount0OutNorm > 0) or (symbol1 in ("WETH", "ETH") and amount1OutNorm > 0): direction = "SELL" else: direction = "BUY" else: direction = "UNKNOWN" usd_value = ethValue * eth_price block = w3.eth.get_block(log["blockNumber"]) timestamps.append(block["timestamp"]) if direction == "SELL": profit_usd_total += usd_value elif direction == "BUY": profit_usd_total -= usd_value invested_usd += usd_value # print(f"[{log['blockNumber']}][{symbol0}/{symbol1}] Sender: {sender} To: {to} In: {amount0InNorm:.6f} {symbol0}, {amount1InNorm:.6f} {symbol1} Out: {amount0OutNorm:.6f} {symbol0}, {amount1OutNorm:.6f} {symbol1} [{direction}] Moneda base: {base_symbol} ETH: {ethValue:.6f} USD: ${usd_value:.2f}") print(f"[{log['blockNumber']}][{symbol0}/{symbol1}] In: {amount0InNorm:.6f} {symbol0}, {amount1InNorm:.6f} {symbol1} Out: {amount0OutNorm:.6f} {symbol0}, {amount1OutNorm:.6f} {symbol1} [{direction}] Moneda base: {base_symbol} ETH: {ethValue:.6f} USD: ${usd_value:.2f}") time.sleep(0.2) except Exception as e: print(f" Error procesando log en bloque {log['blockNumber']}: {e}") continue profit_pct = (profit_usd_total / invested_usd * 100) if invested_usd > 0 else 0 if timestamps: start_date = time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime(min(timestamps))) end_date = time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime(max(timestamps))) else: start_date = end_date = "N/A" print("\n==============================") print(f"Beneficio neto estimado en USD: ${profit_usd_total:.2f}") print(f"Porcentaje de beneficio: {profit_pct:.2f}%") print(f"Período: {start_date} -> {end_date}") print("==============================") if __name__ == "__main__": main()