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()
