Sonic: Swap farming for points walkthrough in Python

import json
import sys
from web3 import Web3
from eth_abi import decode
from hexbytes import HexBytes

# Sonic Blaze Testnet configuration
RPC_URL = "SONIC_BLAZE_NODE"
ROUTER_ADDRESS = "0x086d426f8b653b88a2d6d03051c8b4ab8783be2b"
SWAP_METHOD_ID = "0xddba27a7"  # Method ID for the swap function
SWAP_METHOD_ID_BYTES = bytes.fromhex(SWAP_METHOD_ID[2:])  # Convert to bytes for comparison

# Define start block directly in the script from which to start scanning
START_BLOCK = 24069000  # Replace with your desired start block

# Token addresses to track
TOKENS = {
    "DIAM": "0x30BF3761147Ef0c86E2f84c3784FBD89E7954670",
    "CORAL": "0xAF93888cbD250300470A1618206e036E11470149",
    "FLE": "0x9Fa14D267d331c9E8BB7979bcDC212136915eCE8",
    "MACH": "0x50971F8978C431D560ff658a83a8a03fdf199055",
    "OBS": "0x3e6eE2F3f33766294C7148bc85c7d145E70cBD9A",
    "ONYX": "0xE73c4f6A0A3B0EF8337fD080b76C08172b3eB958"
}

# Normalize addresses to lowercase for comparison
ROUTER_ADDRESS = ROUTER_ADDRESS.lower()
TOKENS = {k: v.lower() for k, v in TOKENS.items()}

# Connect to the Sonic Blaze Testnet
w3 = Web3(Web3.HTTPProvider(RPC_URL))
if not w3.is_connected():
    print("Failed to connect to the Sonic Blaze Testnet")
    sys.exit(1)

print(f"Connected to Sonic Blaze Testnet. Chain ID: {w3.eth.chain_id}")

def extract_addresses_from_input(input_data):
    """
    Extract all Ethereum addresses from the input data.
    This is a more robust approach that doesn't rely on specific offsets.
    """
    addresses = []

    # Remove '0x' prefix if present
    if input_data.startswith('0x'):
        data = input_data[2:]
    else:
        data = input_data

    # Scan through the data looking for potential addresses
    # An Ethereum address is 20 bytes (40 hex chars)
    for i in range(0, len(data) - 40, 2):  # Step by 2 because each byte is 2 hex chars
        # Check if this could be an address (look for leading zeros that pad addresses in ABI encoding)
        potential_address_with_padding = data[i:i+64]
        if len(potential_address_with_padding) == 64 and potential_address_with_padding.startswith('000000000000000000000000'):
            address = '0x' + potential_address_with_padding[24:64]
            # Basic validation: check if it looks like an address
            if all(c in '0123456789abcdefABCDEF' for c in address[2:]):
                addresses.append(address.lower())

    return addresses

def decode_swap_input(input_data):
    """
    Decode the input data of a swap transaction
    """
    try:
        # Extract all addresses from the input data
        addresses = extract_addresses_from_input(input_data)

        if not addresses:
            print("No addresses found in input data")
            return None

        # Filter addresses to find our tokens
        token_addresses = []
        for addr in addresses:
            if addr.lower() in TOKENS.values():
                token_addresses.append(addr.lower())

        # The last address is typically the LP address
        lp_address = None
        for addr in reversed(addresses):
            if addr.lower() not in TOKENS.values():
                lp_address = addr.lower()
                break

        return {
            "path": token_addresses,
            "lp_address": lp_address
        }
    except Exception as e:
        print(f"Error decoding input data: {e}")
        return None

def scan_blocks(start_block):
    """
    Scan blocks for swap transactions involving the specified tokens.
    Continues until LP addresses are found for all tokens.
    """
    # Track LP addresses for each token
    token_lp_map = {token_addr: set() for token_addr in TOKENS.values()}

    # Keep track of tokens that still need LP addresses
    tokens_remaining = set(TOKENS.values())

    print(f"Scanning blocks starting from {start_block}...")
    print(f"Looking for LP addresses for {len(tokens_remaining)} tokens...")

    block_num = start_block

    while tokens_remaining:
        if block_num % 100 == 0:
            print(f"Processing block {block_num}...")

        try:
            block = w3.eth.get_block(block_num, full_transactions=True)

            for tx in block.transactions:
                # Check if transaction is to the router contract
                if hasattr(tx, 'to') and tx.to and tx.to.lower() == ROUTER_ADDRESS:
                    # Check if it's a swap transaction by comparing the first 4 bytes (method ID)
                    if tx.input[:4] == SWAP_METHOD_ID_BYTES:
                        # Get transaction hash
                        tx_hash = tx.hash.hex()

                        decoded = decode_swap_input(tx.input.hex())

                        if decoded and decoded["lp_address"] and decoded["path"]:
                            # Check if any of our tracked tokens are in the path
                            for token_addr in decoded["path"]:
                                if token_addr in tokens_remaining:
                                    token_name = next((name for name, addr in TOKENS.items() if addr == token_addr), token_addr)

                                    # Add LP address to the map
                                    token_lp_map[token_addr].add(decoded["lp_address"])

                                    # Print only if this is the first LP address found for this token
                                    if len(token_lp_map[token_addr]) == 1:
                                        print(f"Block {block_num}, TX {tx_hash}: Found LP for {token_name}: {decoded['lp_address']}")
                                        # Remove token from remaining list
                                        tokens_remaining.remove(token_addr)
                                        print(f"Remaining tokens to find: {len(tokens_remaining)}")

            # Move to the next block
            block_num += 1

        except Exception as e:
            print(f"Error processing block {block_num}: {e}")
            # Continue to the next block even if there's an error
            block_num += 1

    print(f"Found LP addresses for all tokens after scanning to block {block_num-1}")
    return token_lp_map

def main():
    token_lp_map = scan_blocks(START_BLOCK)

    print("\n--- Summary of LP Addresses ---")
    for token_addr, lp_addresses in token_lp_map.items():
        token_name = next((name for name, addr in TOKENS.items() if addr == token_addr), token_addr)
        if lp_addresses:
            print(f"\n{token_name} ({token_addr}):")
            for lp in lp_addresses:
                print(f"  LP: {lp}")
        else:
            print(f"\n{token_name} ({token_addr}): No LP addresses found")

if __name__ == "__main__":
    main()

Connected to Sonic Blaze Testnet. Chain ID: 57054
Scanning blocks starting from 24069000...
Looking for LP addresses for 6 tokens...
Processing block 24069000...
Block 24069018, TX 02fd4c396e9d146eab07b6427a712644fac14cdbc85ea52615f5144484df7c81: Found LP for CORAL: 0x0db7087e050ef022ec1d0432e983ff42506cc990
Remaining tokens to find: 5
Block 24069039, TX b4995dabb7c74ea1bf01834b6c1b305b426905d27b9bd82dc2a55c44050069d7: Found LP for MACH: 0xf88e70b3e1f848c55781297329093e8b15969908
Remaining tokens to find: 4
Block 24069078, TX 9a9680eb2487b8d6d4bd226a624104bdec76a2dff08ae2d90219aa68f2e70642: Found LP for FLE: 0xce86c8d81d24dcfe54d29409afeffde81852b8ad
Remaining tokens to find: 3
Block 24069085, TX a86f4b4f3d46e0a1fae4c369ae5d2541cc43c5d83ee889205d98da5283d401fa: Found LP for ONYX: 0x7d5be487743f73264d6aa4ae202b6103078cd1a8
Remaining tokens to find: 2
Processing block 24069100...
Processing block 24069200...
Block 24069235, TX c6bcd8d9d783aef33819343b46924b33f67b9115a5ab45b5eaa56500caaf1174: Found LP for DIAM: 0x9785f7336ccfdd863ffc8179761f51f81e45bdf4
Remaining tokens to find: 1
Processing block 24069300...
Processing block 24069400...
Processing block 24069500...
Processing block 24069600...
Block 24069659, TX 3a0f5f962ec06a8b91dd99128311f1e2ecc760de48fb1b7732bd04c8db43f6c7: Found LP for OBS: 0xd7d04d8a33b6e6eb42a2e0e273e0fe1f23f818fd
Remaining tokens to find: 0
Found LP addresses for all tokens after scanning to block 24069659

--- Summary of LP Addresses ---

DIAM (0x30bf3761147ef0c86e2f84c3784fbd89e7954670):
  LP: 0x9785f7336ccfdd863ffc8179761f51f81e45bdf4

CORAL (0xaf93888cbd250300470a1618206e036e11470149):
  LP: 0x0db7087e050ef022ec1d0432e983ff42506cc990

FLE (0x9fa14d267d331c9e8bb7979bcdc212136915ece8):
  LP: 0xce86c8d81d24dcfe54d29409afeffde81852b8ad

MACH (0x50971f8978c431d560ff658a83a8a03fdf199055):
  LP: 0xf88e70b3e1f848c55781297329093e8b15969908

OBS (0x3e6ee2f3f33766294c7148bc85c7d145e70cbd9a):
  LP: 0xd7d04d8a33b6e6eb42a2e0e273e0fe1f23f818fd

ONYX (0xe73c4f6a0a3b0ef8337fd080b76c08172b3eb958):
  LP: 0x7d5be487743f73264d6aa4ae202b6103078cd1a8

import json
import time
import sys
from web3 import Web3
from eth_abi import encode
from hexbytes import HexBytes
import requests

# Sonic Blaze Testnet configuration
RPC_URL = "SONIC_BLAZE_NODE"
ROUTER_ADDRESS = "0x086d426f8b653b88a2d6d03051c8b4ab8783be2b"
SWAP_METHOD_ID = "0xddba27a7"  # Method ID for the swap signature

ROUTER_ABI = [
    {
        "inputs": [
            {"name": "amountIn", "type": "uint256"},
            {"name": "path_offset", "type": "uint256"},
            {"name": "to_offset", "type": "uint256"},
            {"name": "path", "type": "address[]"},
            {"name": "lp", "type": "address[]"}
        ],
        "name": "swap",
        "outputs": [{"name": "amounts", "type": "uint256[]"}],
        "stateMutability": "nonpayable",
        "type": "function",
        "function_selector": "0xddba27a7"
    }
]

# Token addresses
TOKENS = {
    "ONYX": "0xE73c4f6A0A3B0EF8337fD080b76C08172b3eB958",
    "CORAL": "0xAF93888cbD250300470A1618206e036E11470149",
    "OBS": "0x3e6eE2F3f33766294C7148bc85c7d145E70cBD9A"
}

# LP addresses for each swap pair
LP_ADDRESSES = {
    "ONYX_CORAL": "0x7D5bE487743F73264D6aA4Ae202B6103078cD1a8",
    "CORAL_OBS": "0xD7D04d8A33b6E6EB42a2e0e273e0fe1F23f818fD",
    "OBS_ONYX": "0xCE1c63381b03bd5f227C1cCfa71c5E93154f336F"
}

# Swap configuration
# IMPORTANT: Replace with your own private key
PRIVATE_KEY = "PRIVATE_KEY"
CYCLES = 4        # Number of swap cycles (0 for infinite)
AMOUNT = 1    # Amount to swap each time
DELAY = 10        # Delay between swaps in seconds

# ERC20 ABI for token balance checking and approvals
ERC20_ABI = [
    {
        "constant": True,
        "inputs": [{"name": "_owner", "type": "address"}],
        "name": "balanceOf",
        "outputs": [{"name": "balance", "type": "uint256"}],
        "type": "function"
    },
    {
        "constant": False,
        "inputs": [
            {"name": "_spender", "type": "address"},
            {"name": "_value", "type": "uint256"}
        ],
        "name": "approve",
        "outputs": [{"name": "", "type": "bool"}],
        "type": "function"
    },
    {
        "constant": True,
        "inputs": [
            {"name": "_owner", "type": "address"},
            {"name": "_spender", "type": "address"}
        ],
        "name": "allowance",
        "outputs": [{"name": "", "type": "uint256"}],
        "type": "function"
    }
]

# Connect to the Sonic Blaze Testnet
w3 = Web3(Web3.HTTPProvider(RPC_URL))
if not w3.is_connected():
    print("Failed to connect to the Sonic Blaze Testnet")
    sys.exit(1)

# Get the chain ID
CHAIN_ID = w3.eth.chain_id
print(f"Connected to Sonic Blaze Testnet. Chain ID: {CHAIN_ID}")

# Convert all addresses to checksum format
ROUTER_ADDRESS = Web3.to_checksum_address(ROUTER_ADDRESS)
TOKENS = {k: Web3.to_checksum_address(v) for k, v in TOKENS.items()}
LP_ADDRESSES = {k: Web3.to_checksum_address(v) for k, v in LP_ADDRESSES.items()}

# Derive wallet address from private key
account = w3.eth.account.from_key(PRIVATE_KEY)
WALLET_ADDRESS = account.address
print(f"Using wallet: {WALLET_ADDRESS}")

def get_token_balance(token_name):
    """
    Get the balance of a specific token for the wallet
    """
    token_address = TOKENS[token_name]
    token_contract = w3.eth.contract(address=token_address, abi=ERC20_ABI)
    balance_wei = token_contract.functions.balanceOf(WALLET_ADDRESS).call()
    balance_eth = w3.from_wei(balance_wei, 'ether')
    return balance_eth

def check_all_balances():
    """
    Check and display balances for all tokens
    """
    print("\n--- Current Token Balances ---")
    for token_name in TOKENS.keys():
        balance = get_token_balance(token_name)
        print(f"{token_name}: {balance}")
    print("-----------------------------")

def check_and_approve_token(token_name, amount):
    """
    Check if the router has enough allowance to spend the token
    If not, approve the router to spend the token
    """
    token_address = TOKENS[token_name]
    token_contract = w3.eth.contract(address=token_address, abi=ERC20_ABI)

    # Convert amount to Wei
    amount_wei = w3.to_wei(amount, 'ether')

    # Check current allowance
    current_allowance = token_contract.functions.allowance(WALLET_ADDRESS, ROUTER_ADDRESS).call()

    if current_allowance < amount_wei:
        print(f"Approving {token_name} for the router...")

        # Approve a large amount to avoid frequent approvals
        # Using max uint256 value (2^256 - 1)
        max_amount = 2**256 - 1

        # Build the approval transaction
        tx = token_contract.functions.approve(
            ROUTER_ADDRESS,
            max_amount
        ).build_transaction({
            'from': WALLET_ADDRESS,
            'gas': 100000,
            'gasPrice': w3.eth.gas_price,
            'nonce': w3.eth.get_transaction_count(WALLET_ADDRESS),
            'chainId': CHAIN_ID
        })

        # Sign and send the transaction
        signed_tx = w3.eth.account.sign_transaction(tx, PRIVATE_KEY)
        tx_hash = w3.eth.send_raw_transaction(signed_tx.raw_transaction)

        print(f"Approval transaction sent: {tx_hash.hex()}")

        # Wait for the transaction to be mined
        receipt = w3.eth.wait_for_transaction_receipt(tx_hash)

        if receipt.status == 1:
            print(f"{token_name} approved successfully!")
        else:
            print(f"Failed to approve {token_name}!")
            return False

    return True

def create_swap_transaction(from_token, to_token, amount, lp_address):
    # Convert amount to Wei (assuming 18 decimals for all tokens)
    amount_wei = w3.to_wei(amount, 'ether')

    # Create the path array with the token addresses
    path = [TOKENS[from_token], TOKENS[to_token]]

    # Function selector
    function_selector = SWAP_METHOD_ID[2:]  # Remove '0x' prefix

    # Encode parameters
    # 1. amountIn (uint256)
    amount_hex = f"{amount_wei:064x}"

    # 2. path_offset (uint256) - fixed at 0x60 (96 in decimal)
    path_offset_hex = "0000000000000000000000000000000000000000000000000000000000000060"

    # 3. to_offset (uint256) - fixed at 0xc0 (192 in decimal)
    to_offset_hex = "00000000000000000000000000000000000000000000000000000000000000c0"

    # 4. path_length (uint256) - 2 tokens
    path_length_hex = "0000000000000000000000000000000000000000000000000000000000000002"

    # 5. path tokens (address[])
    path_tokens_hex = ""
    for token in path:
        path_tokens_hex += f"000000000000000000000000{token[2:].lower()}"

    # 6. lp_length (uint256) - 1 LP
    lp_length_hex = "0000000000000000000000000000000000000000000000000000000000000001"

    # 7. lp_address (address)
    lp_address_hex = f"000000000000000000000000{lp_address[2:].lower()}"

    # Combine all parts
    data = f"0x{function_selector}{amount_hex}{path_offset_hex}{to_offset_hex}{path_length_hex}{path_tokens_hex}{lp_length_hex}{lp_address_hex}"

    # Debug print
    print(f"Transaction data: {data}")
    print(f"From token: {TOKENS[from_token]}")
    print(f"To token: {TOKENS[to_token]}")
    print(f"LP address: {lp_address}")

    # Create transaction dictionary
    tx = {
        'from': WALLET_ADDRESS,
        'to': ROUTER_ADDRESS,
        'value': 0,
        'gas': 500000,
        'gasPrice': w3.eth.gas_price,
        'nonce': w3.eth.get_transaction_count(WALLET_ADDRESS),
        'data': data,
        'chainId': CHAIN_ID
    }

    return tx

def sign_and_send_transaction(tx):
    """
    Sign and send a transaction
    """
    try:
        # Sign the transaction
        signed_tx = w3.eth.account.sign_transaction(tx, PRIVATE_KEY)

        # Send the transaction
        tx_hash = w3.eth.send_raw_transaction(signed_tx.raw_transaction)

        print(f"Transaction sent: {tx_hash.hex()}")

        # Wait for the transaction to be mined
        receipt = w3.eth.wait_for_transaction_receipt(tx_hash)

        if receipt.status == 1:
            print("Transaction successful!")
        else:
            print("Transaction failed!")
            # Try to get more information about the failure
            try:
                # Try to get the transaction trace
                trace_data = {
                    "jsonrpc": "2.0",
                    "method": "trace_transaction",
                    "params": [tx_hash.hex()],
                    "id": 1
                }

                # Make a direct RPC call to get trace data
                response = requests.post(RPC_URL, json=trace_data)
                if response.status_code == 200:
                    trace_result = response.json()
                    if 'result' in trace_result and trace_result['result']:
                        for trace in trace_result['result']:
                            if 'error' in trace:
                                print(f"Transaction error: {trace['error']}")
                    else:
                        print("No trace data available")
                else:
                    print(f"Failed to get trace data: {response.status_code}")
            except Exception as e:
                print(f"Error getting trace data: {e}")

        return receipt
    except Exception as e:
        print(f"Error sending transaction: {e}")
        return None

def perform_round_robin_swaps(cycles=CYCLES, amount=AMOUNT, delay=DELAY):
    """
    Perform continuous round-robin swaps between ONYX, CORAL, and OBS

    Args:
        cycles: Number of complete cycles to perform (0 for infinite)
        amount: Amount of tokens to swap each time
        delay: Delay in seconds between swaps
    """
    # Define the swap sequence
    swap_sequence = [
        {"from": "ONYX", "to": "CORAL", "lp": LP_ADDRESSES["ONYX_CORAL"]},
        {"from": "CORAL", "to": "OBS", "lp": LP_ADDRESSES["CORAL_OBS"]},
        {"from": "OBS", "to": "ONYX", "lp": LP_ADDRESSES["OBS_ONYX"]}
    ]

    # Approve all tokens before starting
    print("Checking and approving tokens...")
    for token_name in TOKENS.keys():
        if not check_and_approve_token(token_name, amount):
            print(f"Failed to approve {token_name}. Aborting.")
            return

    # Wait a bit after approvals to ensure they're processed
    print("Waiting 5 seconds for approvals to be fully processed...")
    time.sleep(5)

    cycle_count = 0

    # Check initial balances
    check_all_balances()

    try:
        while cycles == 0 or cycle_count < cycles:
            print(f"\n--- Starting cycle {cycle_count + 1} ---")

            for swap in swap_sequence:
                from_token = swap['from']
                to_token = swap['to']

                # Check balance before swap
                from_balance_before = get_token_balance(from_token)
                to_balance_before = get_token_balance(to_token)

                print(f"\nSwapping {amount} {from_token} to {to_token} via LP {swap['lp']}")
                print(f"Balance before swap: {from_token}: {from_balance_before}, {to_token}: {to_balance_before}")

                # Check if we have enough balance
                if from_balance_before < amount:
                    print(f"Insufficient {from_token} balance. Skipping this swap.")
                    continue

                # Create and send the swap transaction
                tx = create_swap_transaction(from_token, to_token, amount, swap['lp'])
                receipt = sign_and_send_transaction(tx)

                # Check balance after swap
                from_balance_after = get_token_balance(from_token)
                to_balance_after = get_token_balance(to_token)

                print(f"Balance after swap: {from_token}: {from_balance_after}, {to_token}: {to_balance_after}")
                print(f"Change: {from_token}: {from_balance_after - from_balance_before}, {to_token}: {to_balance_after - to_balance_before}")

                # Wait between swaps
                if delay > 0:
                    print(f"Waiting {delay} seconds before next swap...")
                    time.sleep(delay)

            cycle_count += 1
            print(f"Completed cycle {cycle_count}")

            # Check all balances at the end of each cycle
            check_all_balances()

    except KeyboardInterrupt:
        print("\nSwap process interrupted by user")
        # Show final balances
        check_all_balances()
    except Exception as e:
        print(f"\nError during swap process: {e}")
        # Show balances even if there was an error
        check_all_balances()

def main():
    print("=== Sonic Token Round-Robin Swap ===")
    print("This script will continuously swap between ONYX, CORAL, and OBS tokens")
    print(f"Configured for {CYCLES} cycles, {AMOUNT} tokens per swap, {DELAY}s delay between swaps")

    # Perform the swaps
    perform_round_robin_swaps()

    print("\nSwap process completed")

if __name__ == "__main__":
    main()