bot3.py

bot3.py

import asyncio
import aiohttp
from web3 import Web3
from web3.middleware import geth_poa_middleware
from colorama import Fore, Back, Style, init

# Initialize colorama
init(autoreset=True)

# กำหนดค่าการเชื่อมต่อกับ Polygon network
POLYGON_RPC = 'https://polygon-mainnet.g.alchemy.com/v2/xxxxxxx'
w3 = Web3(Web3.HTTPProvider(POLYGON_RPC))
w3.middleware_onion.inject(geth_poa_middleware, layer=0)

# กำหนดที่อยู่ของ smart contract
QUICKSWAP_ROUTER = '0xa5E0829CaCEd8fFDD4De3c43696c57F7D7A678ff'
SUSHISWAP_ROUTER = '0x1b02dA8Cb0d097eB8D57A175b88c7D8b47997506'

# ABI ของ Router
ROUTER_ABI = [{"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"}]

quickswap_contract = w3.eth.contract(address=QUICKSWAP_ROUTER, abi=ROUTER_ABI)
sushiswap_contract = w3.eth.contract(address=SUSHISWAP_ROUTER, abi=ROUTER_ABI)

# Token addresses
TOKENS = {
    'WMATIC': '0x0d500B1d8E8eF31E21C99d1Db9A6444d3ADf1270',
    'USDC': '0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174',
    'WETH': '0x7ceB23fD6bC0adD59E62ac25578270cFf1b9f619'
}

async def check_connection():
    try:
        print(f"{Fore.CYAN}Checking connection to Polygon network...")
        latest_block = w3.eth.get_block('latest')
        print(f"{Fore.GREEN}Connected successfully. Latest block number: {latest_block['number']}")
        return True
    except Exception as e:
        print(f"{Fore.RED}Connection failed: {e}")
        return False

async def get_amounts_out(router_contract, amount, path):
    try:
        amounts = router_contract.functions.getAmountsOut(amount, path).call()
        return amounts[-1]
    except Exception as e:
        print(f"{Fore.RED}Error in getAmountsOut: {e}")
        return None

def format_amount(amount, decimals=18):
    return f"{amount / 10**decimals:.6f}"

async def check_arbitrage_opportunity(token_a, token_b, token_c, amount):
    if not await check_connection():
        return

    print(f"{Fore.CYAN}Checking arbitrage opportunity for {token_a}/{token_b}/{token_c}")
    print(f"{Fore.CYAN}Input amount: {format_amount(amount)} {token_a}")

    path_1 = [TOKENS[token_a], TOKENS[token_b], TOKENS[token_c], TOKENS[token_a]]
    path_2 = [TOKENS[token_a], TOKENS[token_c], TOKENS[token_b], TOKENS[token_a]]

    results = []
    for dex, contract in [("QuickSwap", quickswap_contract), ("SushiSwap", sushiswap_contract)]:
        for path, path_name in [(path_1, f"{token_a} -> {token_b} -> {token_c} -> {token_a}"),
                                (path_2, f"{token_a} -> {token_c} -> {token_b} -> {token_a}")]:
            result = await get_amounts_out(contract, amount, path)
            if result is not None:
                results.append((dex, path_name, result))

    if not results:
        print(f"{Fore.RED}Failed to fetch prices from all sources")
        return

    best_result = max(results, key=lambda x: x[2])
    worst_result = min(results, key=lambda x: x[2])

    profit = best_result[2] - amount
    profit_percentage = (profit / amount) * 100

    print(f"\n{Fore.GREEN}{Style.BRIGHT}Arbitrage Opportunity Check Results:")
    print(f"{Fore.YELLOW}Best route: {best_result[0]} - {best_result[1]}")
    print(f"{Fore.YELLOW}Worst route: {worst_result[0]} - {worst_result[1]}")
    print(f"{Fore.CYAN}Input amount: {format_amount(amount)} {token_a}")
    print(f"{Fore.CYAN}Output amount: {format_amount(best_result[2])} {token_a}")
    print(f"{Fore.CYAN}Profit: {format_amount(profit)} {token_a}")
    print(f"{Fore.CYAN}Profit percentage: {profit_percentage:.2f}%")

    if profit_percentage > 0.5:
        print(f"\n{Fore.GREEN}{Style.BRIGHT}Significant arbitrage opportunity found!")
    elif profit_percentage > 0:
        print(f"\n{Fore.YELLOW}Small arbitrage opportunity found (profit less than 0.5%)")
    else:
        print(f"\n{Fore.RED}No arbitrage opportunity (trade would result in loss)")

async def run_arbitrage_checker():
    TOKEN_A = 'WMATIC'
    TOKEN_B = 'USDC'
    TOKEN_C = 'WETH'
    AMOUNT_IN = w3.to_wei(10, 'ether')

    while True:
        try:
            print(f"\n{Fore.CYAN}{Style.BRIGHT}Starting arbitrage check...")
            await check_arbitrage_opportunity(TOKEN_A, TOKEN_B, TOKEN_C, AMOUNT_IN)
        except Exception as e:
            print(f'{Fore.RED}An error occurred in the main loop: {e}')
        
        print(f"{Fore.CYAN}Waiting 60 seconds before next check...")
        await asyncio.sleep(60)

if __name__ == '__main__':
    print(f"{Fore.GREEN}{Style.BRIGHT}Starting Improved Three-Token Arbitrage Checker...")
    asyncio.run(run_arbitrage_checker())