如何用web3.py在以太坊打包多笔Uniswap交易与代币转账？

方案选型 #

• Multicall3：优先选择，支持任意合约调用（包括Uniswap V2兑换、ERC20多地址转账），完全覆盖你的需求场景，通用性极强。
• Uniswap V3 Router Multicall：仅支持V3路由相关操作，无法直接处理ERC20多地址转账，局限性大，不推荐。

基于Multicall3的Python实现 #

核心逻辑 #

利用Multicall3的aggregate3Value函数批量执行多个合约调用，该函数支持为每个子调用单独指定ETH金额（适配Uniswap V2的swapExactETHForTokens payable特性），且按顺序执行调用——先完成所有兑换，再执行多地址转账，确保代币到账后再转出。

代码实现 #

1. Multicall3合约ABI片段（仅需aggregate3Value函数）

multicall3_abi = [{
    "inputs": [
        {
            "components": [
                {"internalType": "address", "name": "target", "type": "address"},
                {"internalType": "bool", "name": "requireSuccess", "type": "bool"},
                {"internalType": "uint256", "name": "value", "type": "uint256"},
                {"internalType": "bytes", "name": "callData", "type": "bytes"}
            ],
            "internalType": "struct Multicall3.Call[]",
            "name": "calls",
            "type": "tuple[]"
        }
    ],
    "name": "aggregate3Value",
    "outputs": [
        {
            "components": [
                {"internalType": "bool", "name": "success", "type": "bool"},
                {"internalType": "bytes", "name": "returnData", "type": "bytes"}
            ],
            "internalType": "struct Multicall3.Result[]",
            "name": "returnData",
            "type": "tuple[]"
        }
    ],
    "stateMutability": "payable",
    "type": "function"
}]

2. 批量兑换+转账函数

import time

def batch_swap_and_transfer(swap_params_list, transfer_params_list):
    # 初始化Multicall3合约（主网地址，其他链请替换为对应地址）
    multicall3_addr = web3.to_checksum_address("0xcA11bde05977b3631167028862bE2a173976CA11")
    multicall_contract = web3.eth.contract(address=multicall3_addr, abi=multicall3_abi)
    
    # 初始化Uniswap V2 Router合约
    uni_v2_router = web3.eth.contract(
        address=web3.to_checksum_address(uni_v2_router_address), 
        abi=uni_v2_router_abi
    )
    
    total_eth = 0
    calls = []
    
    # 构建多笔Uniswap兑换调用
    for token_to_buy, eth_amount in swap_params_list:
        token_to_buy = web3.to_checksum_address(token_to_buy)
        token_to_spend = web3.to_checksum_address(chain_token)
        
        # 生成swapExactETHForTokens的调用编码
        swap_call_data = uni_v2_router.functions.swapExactETHForTokens(
            0,  # 生产环境建议设置合理minOut，比如用getAmountsOut计算后打99折
            [token_to_spend, token_to_buy],
            wallet_address,
            int(time.time()) + 600  # 交易过期时间
        ).encodeABI()
        
        eth_wei = web3.to_wei(eth_amount, 'ether')
        total_eth += eth_wei
        
        # 加入Multicall调用列表
        calls.append({
            "target": web3.to_checksum_address(uni_v2_router_address),
            "requireSuccess": True,  # 设为False可允许单调用失败不影响整体
            "value": eth_wei,
            "callData": swap_call_data
        })
    
    # 构建多笔ERC20转账调用
    for token_addr, to_addr, amount in transfer_params_list:
        token_addr = web3.to_checksum_address(token_addr)
        to_addr = web3.to_checksum_address(to_addr)
        
        # 简易ERC20合约（仅需transfer函数）
        token_contract = web3.eth.contract(address=token_addr, abi=[{
            "constant":False,
            "inputs":[
                {"internalType":"address","name":"to","type":"address"},
                {"internalType":"uint256","name":"value","type":"uint256"}
            ],
            "name":"transfer",
            "outputs":[{"internalType":"bool","name":"","type":"bool"}],
            "payable":False,
            "stateMutability":"nonpayable",
            "type":"function"
        }])
        
        # 生成transfer调用编码（注意：代币小数位非18时需调整单位，比如6位小数用'mwei'）
        transfer_call_data = token_contract.functions.transfer(
            to_addr,
            web3.to_wei(amount, 'ether')
        ).encodeABI()
        
        calls.append({
            "target": token_addr,
            "requireSuccess": True,
            "value": 0,  # 转账无需ETH
            "callData": transfer_call_data
        })
    
    # 构建交易
    nonce = web3.eth.get_transaction_count(wallet_address)
    tx = multicall_contract.functions.aggregate3Value(calls).build_transaction({
        "from": wallet_address,
        "value": total_eth,
        "nonce": nonce,
        "gas": 5000000  # 可替换为web3.eth.estimate_gas(tx)获取准确值
    })
    
    # 签名并发送
    signed_txn = web3.eth.account.sign_transaction(tx, private_key=wallet_private_key)
    tx_hash = web3.eth.send_raw_transaction(signed_txn.rawTransaction)
    return web3.to_hex(tx_hash)

3. 使用示例

# 配置：2笔兑换+2笔转账
swap_tasks = [
    ("0xTokenAddress1", 0.01),  # 0.01 ETH 兑换 Token1
    ("0xTokenAddress2", 0.02)   # 0.02 ETH 兑换 Token2
]
transfer_tasks = [
    ("0xTokenAddress1", "0xTargetWallet1", 0.005),  # 转0.005 Token1到钱包1
    ("0xTokenAddress1", "0xTargetWallet2", 0.005)   # 转0.005 Token1到钱包2
]

# 执行批量操作
tx_hash = batch_swap_and_transfer(swap_tasks, transfer_tasks)
print(f"批量交易哈希：{tx_hash}")

关键注意事项 #

• Gas优化：批量操作Gas远低于多笔单独交易，建议用web3.eth.estimate_gas(tx)预估准确Gas值，避免浪费或不足。
• minOut设置：生产环境不要设为0，可通过Uniswap V2的getAmountsOut函数计算预估输出，乘以0.99等折扣作为安全下限，防止三明治攻击。
• Multicall3地址：不同链地址可能不同，需确认对应链的官方地址（如Polygon、BSC主网地址与以太坊主网一致，部分测试链有专属地址）。
• 代币小数位：转账时需匹配目标代币的小数位数，如USDT是6位小数，需用web3.to_wei(amount, 'mwei')或直接计算amount * 10**6。

内容的提问来源于stack exchange，提问作者Symbiose