import os import sys import json import argparse import time from decimal import Decimal from dotenv import load_dotenv from web3 import Web3 # Load environment variables load_dotenv() # --- CONFIGURATION --- # ABIs ERC20_ABI = json.loads(''' [ {"inputs": [], "name": "decimals", "outputs": [{"internalType": "uint8", "name": "", "type": "uint8"}], "stateMutability": "view", "type": "function"}, {"inputs": [], "name": "symbol", "outputs": [{"internalType": "string", "name": "", "type": "string"}], "stateMutability": "view", "type": "function"}, {"inputs": [{"internalType": "address", "name": "account", "type": "address"}], "name": "balanceOf", "outputs": [{"internalType": "uint256", "name": "", "type": "uint256"}], "stateMutability": "view", "type": "function"}, {"inputs": [{"internalType": "address", "name": "spender", "type": "address"}, {"internalType": "uint256", "name": "amount", "type": "uint256"}], "name": "approve", "outputs": [{"internalType": "bool", "name": "", "type": "bool"}], "stateMutability": "nonpayable", "type": "function"}, {"inputs": [{"internalType": "address", "name": "owner", "type": "address"}, {"internalType": "address", "name": "spender", "type": "address"}], "name": "allowance", "outputs": [{"internalType": "uint256", "name": "", "type": "uint256"}], "stateMutability": "view", "type": "function"} ] ''') WETH_ABI = json.loads(''' [ {"inputs": [], "name": "deposit", "outputs": [], "stateMutability": "payable", "type": "function"}, {"inputs": [{"internalType": "uint256", "name": "wad", "type": "uint256"}], "name": "withdraw", "outputs": [], "stateMutability": "nonpayable", "type": "function"} ] ''') # SwapRouter01 (With Deadline in struct) - e.g. Arbitrum 0xE592..., BSC SWAP_ROUTER_01_ABI = json.loads(''' [ {"inputs": [{"components": [{"internalType": "address", "name": "tokenIn", "type": "address"}, {"internalType": "address", "name": "tokenOut", "type": "address"}, {"internalType": "uint24", "name": "fee", "type": "uint24"}, {"internalType": "address", "name": "recipient", "type": "address"}, {"internalType": "uint256", "name": "deadline", "type": "uint256"}, {"internalType": "uint256", "name": "amountIn", "type": "uint256"}, {"internalType": "uint256", "name": "amountOutMinimum", "type": "uint200"}, {"internalType": "uint160", "name": "sqrtPriceLimitX96", "type": "uint160"}], "internalType": "struct ISwapRouter.ExactInputSingleParams", "name": "params", "type": "tuple"}], "name": "exactInputSingle", "outputs": [{"internalType": "uint256", "name": "amountOut", "type": "uint256"}], "stateMutability": "payable", "type": "function"} ] ''') # SwapRouter02 (NO Deadline in struct) - e.g. Base 0x2626... SWAP_ROUTER_02_ABI = json.loads(''' [ {"inputs": [{"components": [{"internalType": "address", "name": "tokenIn", "type": "address"}, {"internalType": "address", "name": "tokenOut", "type": "address"}, {"internalType": "uint24", "name": "fee", "type": "uint24"}, {"internalType": "address", "name": "recipient", "type": "address"}, {"internalType": "uint256", "name": "amountIn", "type": "uint256"}, {"internalType": "uint256", "name": "amountOutMinimum", "type": "uint256"}, {"internalType": "uint160", "name": "sqrtPriceLimitX96", "type": "uint160"}], "internalType": "struct IV3SwapRouter.ExactInputSingleParams", "name": "params", "type": "tuple"}], "name": "exactInputSingle", "outputs": [{"internalType": "uint256", "name": "amountOut", "type": "uint256"}], "stateMutability": "payable", "type": "function"} ] ''') CHAIN_CONFIG = { "ARBITRUM": { "rpc_env": "MAINNET_RPC_URL", "chain_id": 42161, "router": "0x68b3465833fb72A70ecDF485E0e4C7bD8665Fc45", # SwapRouter02 "abi_version": 2, "tokens": { "USDC": "0xaf88d065e77c8cC2239327C5EDb3A432268e5831", "WETH": "0x82aF49447D8a07e3bd95BD0d56f35241523fBab1", "ETH": "0x82aF49447D8a07e3bd95BD0d56f35241523fBab1", # Alias to WETH for wrapping "CBBTC": "0xcbB7C0000aB88B473b1f5aFd9ef808440eed33Bf" }, "default_fee": 500 }, "BASE": { "rpc_env": "BASE_RPC_URL", "chain_id": 8453, "router": "0x2626664c2603336E57B271c5C0b26F421741e481", # SwapRouter02 "abi_version": 2, "tokens": { "USDC": "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913", "WETH": "0x4200000000000000000000000000000000000006", "ETH": "0x4200000000000000000000000000000000000006", # Alias to WETH for wrapping "CBBTC": "0xcbB7C0000aB88B473b1f5aFd9ef808440eed33Bf" }, "default_fee": 500 }, "BASE_AERO": { "rpc_env": "BASE_RPC_URL", "chain_id": 8453, "router": "0xbe6D8f0D397708D99755B7857067757f97174d7d", # Aerodrome Slipstream SwapRouter "abi_version": 1, # Router requires deadline (Standard SwapRouter01 style) "tokens": { "USDC": "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913", "WETH": "0x4200000000000000000000000000000000000006", "ETH": "0x4200000000000000000000000000000000000006", "CBBTC": "0xcbB7C0000aB88B473b1f5aFd9ef808440eed33Bf" }, "default_fee": 1 # TickSpacing 1 (0.01%) }, "BSC": { "rpc_env": "BNB_RPC_URL", "chain_id": 56, "router": "0x1b81D678ffb9C0263b24A97847620C99d213eB14", # PancakeSwap V3 "abi_version": 1, "tokens": { "USDC": "0x8ac76a51cc950d9822d68b83fe1ad97b32cd580d", "WBNB": "0xbb4CdB9CBd36B01bD1cBaEBF2De08d9173bc095c", "BNB": "0xbb4CdB9CBd36B01bD1cBaEBF2De08d9173bc095c" # Alias to WBNB for wrapping }, "default_fee": 500 } } def get_web3(chain_name): config = CHAIN_CONFIG.get(chain_name.upper()) if not config: raise ValueError(f"Unsupported chain: {chain_name}") rpc_url = os.environ.get(config["rpc_env"]) if not rpc_url: raise ValueError(f"RPC URL not found in environment for {config['rpc_env']}") w3 = Web3(Web3.HTTPProvider(rpc_url)) if not w3.is_connected(): raise ConnectionError(f"Failed to connect to {chain_name} RPC") return w3, config def approve_token(w3, token_contract, spender_address, amount, private_key, my_address): """ Checks allowance and approves if necessary. Robust gas handling. """ allowance = token_contract.functions.allowance(my_address, spender_address).call() if allowance >= amount: print(f"Token already approved (Allowance: {allowance})") return True print(f"Approving token... (Current: {allowance}, Needed: {amount})") # Build tx base tx_params = { 'from': my_address, 'nonce': w3.eth.get_transaction_count(my_address), } # Determine Gas Strategy try: latest_block = w3.eth.get_block('latest') if 'baseFeePerGas' in latest_block: # EIP-1559 base_fee = latest_block['baseFeePerGas'] priority_fee = w3.to_wei(0.1, 'gwei') # Conservative priority tx_params['maxFeePerGas'] = int(base_fee * 1.5) + priority_fee tx_params['maxPriorityFeePerGas'] = priority_fee else: # Legacy tx_params['gasPrice'] = w3.eth.gas_price except Exception as e: print(f"Error determining gas strategy: {e}. Fallback to w3.eth.gas_price") tx_params['gasPrice'] = w3.eth.gas_price # Build transaction tx = token_contract.functions.approve(spender_address, 2**256 - 1).build_transaction(tx_params) # Sign and send signed_tx = w3.eth.account.sign_transaction(tx, private_key) tx_hash = w3.eth.send_raw_transaction(signed_tx.raw_transaction) print(f"Approval Tx sent: {tx_hash.hex()}") receipt = w3.eth.wait_for_transaction_receipt(tx_hash) if receipt.status == 1: print("Approval successful.") return True else: print("Approval failed.") return False def wrap_eth(w3, weth_address, amount_wei, private_key, my_address): """ Wraps native ETH/BNB to WETH/WBNB. """ print(f"Wrapping native token to wrapped version...") weth_contract = w3.eth.contract(address=weth_address, abi=WETH_ABI) tx_params = { 'from': my_address, 'value': amount_wei, 'nonce': w3.eth.get_transaction_count(my_address), } # Gas logic (Simplified) latest_block = w3.eth.get_block('latest') if 'baseFeePerGas' in latest_block: base_fee = latest_block['baseFeePerGas'] tx_params['maxFeePerGas'] = int(base_fee * 1.5) + w3.to_wei(0.1, 'gwei') tx_params['maxPriorityFeePerGas'] = w3.to_wei(0.1, 'gwei') else: tx_params['gasPrice'] = w3.eth.gas_price tx = weth_contract.functions.deposit().build_transaction(tx_params) signed_tx = w3.eth.account.sign_transaction(tx, private_key) tx_hash = w3.eth.send_raw_transaction(signed_tx.raw_transaction) print(f"Wrapping Tx sent: {tx_hash.hex()}") w3.eth.wait_for_transaction_receipt(tx_hash) print("Wrapping successful.") def execute_swap(chain_name, token_in_sym, token_out_sym, amount_in_readable, fee_tier=None, slippage_pct=0.5): """ Main function to execute swap. """ chain_name = chain_name.upper() token_in_sym = token_in_sym.upper() token_out_sym = token_out_sym.upper() w3, config = get_web3(chain_name) # Get private key private_key = os.environ.get("MAIN_WALLET_PRIVATE_KEY") if not private_key: raise ValueError("MAIN_WALLET_PRIVATE_KEY not found in environment variables") account = w3.eth.account.from_key(private_key) my_address = account.address print(f"Connected to {chain_name} as {my_address}") # Validate tokens if token_in_sym not in config["tokens"]: raise ValueError(f"Token {token_in_sym} not supported on {chain_name}") if token_out_sym not in config["tokens"]: raise ValueError(f"Token {token_out_sym} not supported on {chain_name}") token_in_addr = config["tokens"][token_in_sym] token_out_addr = config["tokens"][token_out_sym] router_addr = config["router"] abi_ver = config.get("abi_version", 1) # Initialize Contracts token_in_contract = w3.eth.contract(address=token_in_addr, abi=ERC20_ABI) token_out_contract = w3.eth.contract(address=token_out_addr, abi=ERC20_ABI) router_abi = SWAP_ROUTER_01_ABI if abi_ver == 1 else SWAP_ROUTER_02_ABI router_contract = w3.eth.contract(address=router_addr, abi=router_abi) # Decimals (ETH/BNB and their wrapped versions all use 18) decimals_in = 18 if token_in_sym in ["ETH", "BNB"] else token_in_contract.functions.decimals().call() amount_in_wei = int(Decimal(str(amount_in_readable)) * Decimal(10)**decimals_in) print(f"Preparing to swap {amount_in_readable} {token_in_sym} -> {token_out_sym}") # Handle Native Wrap if token_in_sym in ["ETH", "BNB"]: # Check native balance native_balance = w3.eth.get_balance(my_address) if native_balance < amount_in_wei: raise ValueError(f"Insufficient native balance. Have {native_balance / 10**18}, need {amount_in_readable}") # Check if we already have enough wrapped token w_balance = token_in_contract.functions.balanceOf(my_address).call() if w_balance < amount_in_wei: wrap_eth(w3, token_in_addr, amount_in_wei - w_balance, private_key, my_address) else: # Check Token Balance balance = token_in_contract.functions.balanceOf(my_address).call() if balance < amount_in_wei: raise ValueError(f"Insufficient balance. Have {balance / 10**decimals_in} {token_in_sym}, need {amount_in_readable}") # Approve approve_token(w3, token_in_contract, router_addr, amount_in_wei, private_key, my_address) # Prepare Swap Params used_fee = fee_tier if fee_tier else config["default_fee"] amount_out_min = 0 if abi_ver == 1: # Router 01 (Deadline in struct) params = ( token_in_addr, token_out_addr, used_fee, my_address, int(time.time()) + 120, # deadline amount_in_wei, amount_out_min, 0 # sqrtPriceLimitX96 ) else: # Router 02 (No Deadline in struct) params = ( token_in_addr, token_out_addr, used_fee, my_address, # No deadline here amount_in_wei, amount_out_min, 0 # sqrtPriceLimitX96 ) print(f"Swapping... Fee Tier: {used_fee} | ABI: V{abi_ver}") # Build Tx tx_build = { 'from': my_address, 'nonce': w3.eth.get_transaction_count(my_address), } # Estimate Gas try: gas_estimate = router_contract.functions.exactInputSingle(params).estimate_gas(tx_build) tx_build['gas'] = int(gas_estimate * 1.2) except Exception as e: print(f"Gas estimation failed: {e}. Using default gas limit (500k).") tx_build['gas'] = 500000 # Add Gas Price (Same robust logic as approve) if chain_name == "BSC": tx_build['gasPrice'] = w3.eth.gas_price else: try: latest_block = w3.eth.get_block('latest') if 'baseFeePerGas' in latest_block: base_fee = latest_block['baseFeePerGas'] priority_fee = w3.to_wei(0.1, 'gwei') tx_build['maxFeePerGas'] = int(base_fee * 1.5) + priority_fee tx_build['maxPriorityFeePerGas'] = priority_fee else: tx_build['gasPrice'] = w3.eth.gas_price except: tx_build['gasPrice'] = w3.eth.gas_price # Sign and Send tx_func = router_contract.functions.exactInputSingle(params) tx = tx_func.build_transaction(tx_build) signed_tx = w3.eth.account.sign_transaction(tx, private_key) tx_hash = w3.eth.send_raw_transaction(signed_tx.raw_transaction) print(f"Swap Tx Sent: {tx_hash.hex()}") receipt = w3.eth.wait_for_transaction_receipt(tx_hash) if receipt.status == 1: print("Swap Successful!") else: print("Swap Failed!") # print(receipt) # verbose if __name__ == "__main__": parser = argparse.ArgumentParser(description="Universal Swapper for Arbitrum, Base, BSC") parser.add_argument("chain", help="Chain name (ARBITRUM, BASE, BSC)") parser.add_argument("token_in", help="Token to sell (e.g. USDC)") parser.add_argument("token_out", help="Token to buy (e.g. WETH)") parser.add_argument("amount", help="Amount to swap", type=float) parser.add_argument("--fee", help="Fee tier (e.g. 500, 3000)", type=int) args = parser.parse_args() try: execute_swap(args.chain, args.token_in, args.token_out, args.amount, args.fee) except Exception as e: print(f"Error: {e}")