Мобильная версияОднако я не могу проверить контракты и получить ошибки.
Код: Выделить всё
import os
import json
import time
import logging
from web3 import Web3, exceptions
from eth_account import Account
from dotenv import load_dotenv
from web3.middleware import ExtraDataToPOAMiddleware
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
# Load environment variables
load_dotenv()
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
class ArbitrumArbitrageBot:
def __init__(self, testnet=True, dry_run=True):
self.testnet = testnet
self.dry_run = dry_run
self.initial_eth = 0
self.w3 = None
self.setup_networks()
self.setup_web3_with_fallback()
self.setup_account()
self.setup_abis()
self.load_contracts()
self.setup_parameters()
self.verify_contracts()
def setup_networks(self):
self.network_config = {
'mainnet': {
'rpc_urls': [
'https://arb1.arbitrum.io/rpc',
'https://arbitrum-one.publicnode.com',
'https://endpoints.omniatech.io/v1/arbitrum/one/public'
],
'tokens': {
'WETH': '0x82aF49447D8a07e3bd95BD0d56f35241523fBab1',
'DAI': '0xDA10009cBd5D07dd0CeCc66161FC93D7c9000da1'
},
'routers': {
'Uniswap': '0x4C60051384bd2d3C01bfc845Cf5F4b44bcbE9de5',
'Camelot': '0xc873fEcbd354f5A56E00E710B90EF4201db2448d'
}
},
'testnet': {
'rpc_urls': [
# Public Arbitrum Sepolia RPCs (no API key required)
'https://sepolia-rollup.arbitrum.io/rpc',
'https://arbitrum-sepolia.publicnode.com',
'https://arbitrum-sepolia-rpc.publicnode.com',
# Your Alchemy endpoint as fallback
'https://arb-sepolia.g.alchemy.com/v2/t'
],
'tokens': {
'WETH': '0x980B62Da83eFf3D4576C647993b0c1D7faf17c73',
'DAI': '0x9bc8388dD439fa3365B1F78A81242aDBB4677759'
},
'routers': {
'Uniswap': '0x101F443a4A2D26158236F5aD075C6591b03Fc3c5',
'Camelot': '0x171B925C51565F5D2a7d8C494ba3188D304EFD93'
}
}
}
def create_session_with_retries(self):
"""Create requests session with retry strategy"""
session = requests.Session()
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["HEAD", "GET", "POST"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)
return session
def test_rpc_connection(self, rpc_url):
"""Test if RPC endpoint is accessible"""
try:
logging.info(f"Testing RPC connection: {rpc_url}")
# Create a custom session with timeouts
session = self.create_session_with_retries()
# Test with a simple JSON-RPC call
payload = {
"jsonrpc": "2.0",
"method": "eth_chainId",
"params": [],
"id": 1
}
response = session.post(
rpc_url,
json=payload,
timeout=(10, 30), # (connect timeout, read timeout)
headers={'Content-Type': 'application/json'}
)
if response.status_code == 200:
result = response.json()
if 'result' in result:
chain_id = int(result['result'], 16)
expected_chain_id = 421614 if self.testnet else 42161
if chain_id == expected_chain_id:
logging.info(f"✓ RPC connection successful: {rpc_url}")
return True
else:
logging.warning(f"✗ Wrong chain ID {chain_id}, expected {expected_chain_id}")
return False
logging.warning(f"✗ RPC test failed: {rpc_url} - Status: {response.status_code}")
return False
except requests.exceptions.Timeout:
logging.warning(f"✗ RPC timeout: {rpc_url}")
return False
except requests.exceptions.ConnectionError as e:
logging.warning(f"✗ RPC connection error: {rpc_url} - {str(e)}")
return False
except Exception as e:
logging.warning(f"✗ RPC test error: {rpc_url} - {str(e)}")
return False
def setup_web3_with_fallback(self):
"""Setup Web3 with fallback RPC endpoints"""
network = 'testnet' if self.testnet else 'mainnet'
rpc_urls = self.network_config[network]['rpc_urls']
for rpc_url in rpc_urls:
if self.test_rpc_connection(rpc_url):
try:
# Create Web3 instance with custom request kwargs
self.w3 = Web3(Web3.HTTPProvider(
rpc_url,
request_kwargs={
'timeout': (10, 30), # (connect timeout, read timeout)
}
))
if self.testnet:
self.w3.middleware_onion.inject(ExtraDataToPOAMiddleware(), layer=0)
# Test the connection by getting chain ID
chain_id = self.w3.eth.chain_id
expected_chain_id = 421614 if self.testnet else 42161
if chain_id == expected_chain_id:
logging.info(f"✓ Web3 connected successfully to {rpc_url}")
logging.info(f"✓ Chain ID: {chain_id}")
return
else:
logging.warning(f"✗ Wrong chain ID {chain_id}, expected {expected_chain_id}")
continue
except Exception as e:
logging.warning(f"✗ Web3 setup failed for {rpc_url}: {str(e)}")
continue
raise ConnectionError(f"Failed to connect to any RPC endpoint for {'testnet' if self.testnet else 'mainnet'}")
def setup_account(self):
key_var = 'TESTNET_PRIVATE_KEY' if self.testnet else 'MAINNET_PRIVATE_KEY'
private_key = os.getenv(key_var)
if not private_key:
raise ValueError(f"Missing {key_var} in .env file")
try:
self.account = Account.from_key(private_key)
self.address = self.account.address
logging.info(f"Account address: {self.address}")
# Get initial balance with retry
self.initial_eth = self.get_eth_balance_with_retry()
logging.info(f"Initial ETH balance: {Web3.from_wei(self.initial_eth, 'ether')} ETH")
except Exception as e:
logging.error(f"Account setup failed: {str(e)}")
raise
def get_eth_balance_with_retry(self, max_retries=3):
"""Get ETH balance with retry logic"""
for attempt in range(max_retries):
try:
balance = self.w3.eth.get_balance(self.address)
return balance
except Exception as e:
logging.warning(f"Balance check attempt {attempt + 1} failed: {str(e)}")
if attempt < max_retries - 1:
time.sleep(2 ** attempt) # Exponential backoff
else:
raise e
def setup_abis(self):
# Standard ERC20 ABI with essential functions
self.erc20_abi = [
{
"constant": False,
"inputs": [
{"name": "_spender", "type": "address"},
{"name": "_value", "type": "uint256"}
],
"name": "approve",
"outputs": [{"name": "success", "type": "bool"}],
"type": "function"
},
{
"constant": True,
"inputs": [{"name": "_owner", "type": "address"}],
"name": "balanceOf",
"outputs": [{"name": "balance", "type": "uint256"}],
"type": "function"
}
]
# Uniswap V2 Router ABI (simplified)
self.uniswap_v2_abi = [
{
"inputs": [],
"name": "WETH",
"outputs": [{"internalType": "address", "name": "", "type": "address"}],
"stateMutability": "view",
"type": "function"
},
{
"inputs": [
{"internalType": "uint256", "name": "amountIn", "type": "uint256"},
{"internalType": "address[]", "name": "path", "type": "address[]"}
],
"name": "getAmountsOut",
"outputs": [{"internalType": "uint256[]", "name": "amounts", "type": "uint256[]"}],
"stateMutability": "view",
"type": "function"
},
{
"inputs": [
{"internalType": "uint256", "name": "amountOutMin", "type": "uint256"},
{"internalType": "address[]", "name": "path", "type": "address[]"},
{"internalType": "address", "name": "to", "type": "address"},
{"internalType": "uint256", "name": "deadline", "type": "uint256"}
],
"name": "swapExactETHForTokens",
"outputs": [{"internalType": "uint256[]", "name": "amounts", "type": "uint256[]"}],
"stateMutability": "payable",
"type": "function"
},
{
"inputs": [
{"internalType": "uint256", "name": "amountIn", "type": "uint256"},
{"internalType": "uint256", "name": "amountOutMin", "type": "uint256"},
{"internalType": "address[]", "name": "path", "type": "address[]"},
{"internalType": "address", "name": "to", "type": "address"},
{"internalType": "uint256", "name": "deadline", "type": "uint256"}
],
"name": "swapExactTokensForETH",
"outputs": [{"internalType": "uint256[]", "name": "amounts", "type": "uint256[]"}],
"stateMutability": "nonpayable",
"type": "function"
}
]
# Camelot Router ABI
self.camelot_abi = [
{
"inputs": [],
"name": "WETH",
"outputs": [{"internalType": "address", "name": "", "type": "address"}],
"stateMutability": "view",
"type": "function"
},
{
"inputs": [
{"internalType": "uint256", "name": "amountIn", "type": "uint256"},
{"internalType": "address[]", "name": "path", "type": "address[]"}
],
"name": "getAmountsOut",
"outputs": [{"internalType": "uint256[]", "name": "amounts", "type": "uint256[]"}],
"stateMutability": "view",
"type": "function"
},
{
"inputs": [
{"internalType": "uint256", "name": "amountOutMin", "type": "uint256"},
{"internalType": "address[]", "name": "path", "type": "address[]"},
{"internalType": "address", "name": "to", "type": "address"},
{"internalType": "address", "name": "referrer", "type": "address"},
{"internalType": "uint256", "name": "deadline", "type": "uint256"}
],
"name": "swapExactETHForTokensSupportingFeeOnTransferTokens",
"outputs": [],
"stateMutability": "payable",
"type": "function"
},
{
"inputs": [
{"internalType": "uint256", "name": "amountIn", "type": "uint256"},
{"internalType": "uint256", "name": "amountOutMin", "type": "uint256"},
{"internalType": "address[]", "name": "path", "type": "address[]"},
{"internalType": "address", "name": "to", "type": "address"},
{"internalType": "address", "name": "referrer", "type": "address"},
{"internalType": "uint256", "name": "deadline", "type": "uint256"}
],
"name": "swapExactTokensForETHSupportingFeeOnTransferTokens",
"outputs": [],
"stateMutability": "nonpayable",
"type": "function"
}
]
def load_contracts(self):
network = 'testnet' if self.testnet else 'mainnet'
self.tokens = {
token: Web3.to_checksum_address(addr)
for token, addr in self.network_config[network]['tokens'].items()
}
self.contracts = {
'Uniswap': self.w3.eth.contract(
address=Web3.to_checksum_address(self.network_config[network]['routers']['Uniswap']),
abi=self.uniswap_v2_abi
),
'Camelot': self.w3.eth.contract(
address=Web3.to_checksum_address(self.network_config[network]['routers']['Camelot']),
abi=self.camelot_abi
)
}
def verify_contracts(self):
"""Verify contracts with better error handling"""
working_contracts = {}
for dex, router in self.contracts.items():
try:
logging.info(f"Verifying {dex} contract at {router.address}...")
# Test WETH function
try:
weth_addr = router.functions.WETH().call()
logging.info(f"✓ {dex} WETH address: {weth_addr}")
except Exception as e:
logging.error(f"✗ {dex} WETH() call failed: {str(e)}")
continue
# Test price check with smaller amount to avoid liquidity issues
try:
test_amount = Web3.to_wei(0.001, 'ether') # Smaller test amount
amounts = router.functions.getAmountsOut(
test_amount,
[weth_addr, self.tokens['DAI']]
).call()
logging.info(f"✓ {dex} price check successful: {amounts}")
working_contracts[dex] = router
except Exception as price_error:
logging.warning(f"⚠ {dex} price check failed: {str(price_error)}")
# Still add to working contracts as price check might fail due to low liquidity
working_contracts[dex] = router
except Exception as e:
logging.error(f"✗ Contract verification failed for {dex}: {str(e)}")
if not working_contracts:
logging.error("No working DEX contracts found!")
else:
logging.info(f"✓ {len(working_contracts)} DEX contracts verified")
self.contracts = working_contracts
def setup_parameters(self):
self.slippage = 0.015 # 1.5%
self.trade_amount_eth = Web3.to_wei(0.001, 'ether') # Smaller amount for testing
self.max_gas_gwei = 10 if self.testnet else 100 # Lower gas for testnet
self.min_profit_eth = Web3.to_wei(0.0001, 'ether') # Smaller minimum profit
self.max_loss_eth = Web3.to_wei(0.0005, 'ether')
def get_gas_price(self):
try:
base_gas = self.w3.eth.gas_price
adjusted_gas = int(base_gas * 1.1) # 10% increase instead of 20%
max_gas = Web3.to_wei(self.max_gas_gwei, 'gwei')
return min(adjusted_gas, max_gas)
except Exception as e:
logging.error(f"Gas price error: {e}")
return Web3.to_wei(2 if self.testnet else 20, 'gwei')
def get_eth_balance(self):
return self.get_eth_balance_with_retry()
def get_token_balance(self, token_address):
try:
contract = self.w3.eth.contract(address=token_address, abi=self.erc20_abi)
return contract.functions.balanceOf(self.address).call()
except Exception as e:
logging.error(f"Token balance check failed: {e}")
return 0
def approve_token(self, token_address, spender):
if self.dry_run:
logging.info(f"✓ Dry run: Would approve {token_address}")
return True
contract = self.w3.eth.contract(address=token_address, abi=self.erc20_abi)
try:
tx = contract.functions.approve(
spender,
2**256 - 1
).build_transaction({
'from': self.address,
'gas': 100000,
'gasPrice': self.get_gas_price(),
'nonce': self.w3.eth.get_transaction_count(self.address)
})
signed_tx = self.account.sign_transaction(tx)
tx_hash = self.w3.eth.send_raw_transaction(signed_tx.rawTransaction)
receipt = self.w3.eth.wait_for_transaction_receipt(tx_hash)
return receipt.status == 1
except Exception as e:
logging.error(f"Approval failed: {e}")
return False
def get_price(self, router, path, amount_in):
try:
return router.functions.getAmountsOut(amount_in, path).call()[-1]
except exceptions.ContractLogicError as e:
logging.warning(f"Price check failed (contract logic): {e}")
return None
except exceptions.BadFunctionCallOutput:
logging.warning("Price check failed (ABI/Contract mismatch)")
return None
except Exception as e:
logging.warning(f"Price check failed (unexpected): {e}")
return None
def execute_swap(self, dex_name, router, path, amount_in, min_amount_out, is_eth_to_token=True):
if self.dry_run:
logging.info(f"✓ Dry run: Would swap on {dex_name}")
return True
deadline = int(time.time()) + 300
tx_args = {
'from': self.address,
'gas': 300000,
'gasPrice': self.get_gas_price(),
'nonce': self.w3.eth.get_transaction_count(self.address)
}
try:
if dex_name == 'Camelot':
if is_eth_to_token:
tx = router.functions.swapExactETHForTokensSupportingFeeOnTransferTokens(
min_amount_out,
path,
self.address,
'0x0000000000000000000000000000000000000000', # referrer
deadline
).build_transaction({**tx_args, 'value': amount_in})
else:
tx = router.functions.swapExactTokensForETHSupportingFeeOnTransferTokens(
amount_in,
min_amount_out,
path,
self.address,
'0x0000000000000000000000000000000000000000', # referrer
deadline
).build_transaction(tx_args)
else: # Uniswap
if is_eth_to_token:
tx = router
Подробнее здесь: https://stackoverflow.com/questions/796 ... on-but-get
