使用 Python 构建 Uniswap 和 MetaMask 三明治交易机器人教程

import time
import json
import requests
from web3 import Web3
from web3.middleware import geth_poa_middleware
from web3.exceptions import InvalidAddress
from eth_account import Account

# 以太坊节点提供者URL
provider_url = 'https://mainnet.infura.io/v3/YOUR_INFURA_PROJECT_ID'

# Uniswap合约地址
router_address = '0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D'
token_1_address = '0xTOKEN_1_ADDRESS' # 示例代币1地址
token_2_address = '0xTOKEN_2_ADDRESS' # 示例代币2地址

# MetaMask钱包私钥和密码短语
wallet_private_key = 'YOUR_WALLET_PRIVATE_KEY'
wallet_passphrase = 'YOUR_WALLET_PASSPHRASE'

# 初始化Web3提供者
web3 = Web3(Web3.HTTPProvider(provider_url))
web3.middleware_onion.inject(geth_poa_middleware, layer=0)

# 加载钱包
wallet = Account.from_key(wallet_private_key)

# 设置合约地址
router_contract = web3.eth.contract(address=Web3.toChecksumAddress(router_address), abi=json.loads(UNISWAP_ROUTER_ABI))

# 指定代币和交易数量
token_1_amount = web3.toWei(1, 'ether') # 交易代币1的数量
token_2_amount = web3.toWei(1, 'ether') # 交易代币2的数量
trade_slippage = 0.05 # 滑点容差(5%)

# 执行三明治交易的函数
def execute_sandwich_trade():
    # 从Uniswap获取最新代币价格
    token_1_price = get_token_price(token_1_address)
    token_2_price = get_token_price(token_2_address)
    
    # 计算目标价格
    buy_price = token_1_price - (token_2_price * (1 + trade_slippage))
    sell_price = token_1_price + (token_2_price * (1 - trade_slippage))
    
    # 模拟检查市场价格
    # 请替换为您自己获取实时市场价格的逻辑
    token_1_market_price = get_token_price(token_1_address)
    token_2_market_price = get_token_price(token_2_address)
    
    # 执行买入交易
    if token_1_market_price <= buy_price:
        print("正在执行买入交易…")
        execute_uniswap_trade(token_2_address, token_2_amount, token_1_address, token_1_amount)
    
    # 执行卖出交易
    elif token_1_market_price >= sell_price:
        print("正在执行卖出交易…")
        execute_uniswap_trade(token_1_address, token_1_amount, token_2_address, token_2_amount)

# 从Uniswap获取代币价格的函数
def get_token_price(token_address):
    try:
        response = requests.get(f'https://api.coingecko.com/api/v3/simple/token_price/ethereum?contract_addresses={token_address}&vs_currencies=usd')
        data = response.json()
        token_price = data[token_address.lower()]['usd']
        return token_price
    except Exception as e:
        print(f"获取代币价格时出错: {e}")
        return None

# 在Uniswap上执行交易的函数
def execute_uniswap_trade(from_token, from_amount, to_token, to_amount):
    try:
        # 批准代币转移
        approve_tx = router_contract.functions.approve(
            Web3.toChecksumAddress(router_address),
            from_amount
        ).buildTransaction({
            'from': wallet.address,
            'gas': 100000,
            'gasPrice': web3.toWei('5', 'gwei'),
            'nonce': web3.eth.getTransactionCount(wallet.address),
        })
        
        signed_approve_tx = web3.eth.account.sign_transaction(approve_tx, private_key=wallet.privateKey)
        approve_tx_hash = web3.eth.send_raw_transaction(signed_approve_tx.rawTransaction)
        web3.eth.wait_for_transaction_receipt(approve_tx_hash)
        
        # 执行交换
        swap_tx = router_contract.functions.swapExactTokensForTokens(
            from_amount,
            to_amount,
            [from_token, to_token],
            wallet.address,
            int(time.time()) + 10000
        ).buildTransaction({
            'from': wallet.address,
            'gas': 200000,
            'gasPrice': web3.toWei('5', 'gwei'),
            'nonce': web3.eth.getTransactionCount(wallet.address),
        })
        
        signed_swap_tx = web3.eth.account.sign_transaction(swap_tx, private_key=wallet.privateKey)
        swap_tx_hash = web3.eth.send_raw_transaction(signed_swap_tx.rawTransaction)
        web3.eth.wait_for_transaction_receipt(swap_tx_hash)
        
        print("交易执行成功。")
    except InvalidAddress as e:
        print(f"合约地址无效: {e}")
    except Exception as e:
        print(f"执行交易时出错: {e}")

# 主循环
while True:
    execute_sandwich_trade()
    time.sleep(10) # 等待10秒后执行下一次交易