import os
from urllib.parse import urlencode
import requests
from eth_account import Account
from eth_account.messages import encode_typed_data
from web3 import Web3
API_BASE = "https://rialto-trade-api.rialto.xyz"
CHAIN_ID = 4663
SELL_TOKEN = "0x5fc5360D0400a0Fd4f2af552ADD042D716F1d168"
BUY_TOKEN = "0xc93a8c440CEa26D7445dF01729f193b27965099f"
SELL_AMOUNT = "0.532262"
SLIPPAGE_BPS = 50
SWAP_FEE_BPS = 50
ERC20_ABI = [
{
"name": "approve",
"type": "function",
"stateMutability": "nonpayable",
"inputs": [
{"name": "spender", "type": "address"},
{"name": "value", "type": "uint256"},
],
"outputs": [{"name": "", "type": "bool"}],
}
]
def require_env(name: str) -> str:
value = os.getenv(name)
if not value:
raise RuntimeError(f"missing {name}")
return value
def tx_fees(w3: Web3) -> dict:
latest = w3.eth.get_block("latest")
priority = w3.to_wei("0.01", "gwei")
return {
"maxPriorityFeePerGas": priority,
"maxFeePerGas": int(latest["baseFeePerGas"]) * 2 + priority,
}
def patch_permit2_signature(tx_data: str, signature_offset: int, signature: bytes) -> str:
data = bytearray(bytes.fromhex(tx_data.removeprefix("0x")))
if len(signature) != 65:
raise RuntimeError("Permit2 signature must be 65 bytes")
data[signature_offset : signature_offset + 65] = signature
return "0x" + data.hex()
api_key = require_env("INTEGRATOR_API_KEY")
private_key = require_env("PRIVATE_KEY")
w3 = Web3(Web3.HTTPProvider(require_env("RPC_URL")))
taker = Account.from_key(private_key).address
params = {
"sell_token": SELL_TOKEN,
"buy_token": BUY_TOKEN,
"sell_amount": SELL_AMOUNT,
"taker": taker,
"slippage_bps": SLIPPAGE_BPS,
"chain_id": CHAIN_ID,
"swapFeeBps": SWAP_FEE_BPS,
}
response = requests.get(
f"{API_BASE}/quote?{urlencode(params)}",
headers={"Authorization": f"Bearer {api_key}"},
timeout=30,
)
response.raise_for_status()
quote = response.json()
print("quote_id:", quote["quote_id"])
print("buy_amount:", quote["buy_amount"])
print("min_buy_amount:", quote["min_buy_amount"])
print("integrator_fee:", quote.get("integrator_fee"))
issues = quote.get("issues") or {}
if issues.get("balance"):
raise RuntimeError(f"insufficient balance: {issues['balance']}")
if issues.get("allowance"):
allowance = issues["allowance"]
token = Web3.to_checksum_address(quote["sell_token"])
spender = Web3.to_checksum_address(allowance["spender"])
amount = int(quote["sell_amount"])
approve_tx = w3.eth.contract(address=token, abi=ERC20_ABI).functions.approve(
spender, amount
).build_transaction(
{
"from": taker,
"chainId": CHAIN_ID,
"nonce": w3.eth.get_transaction_count(taker),
**tx_fees(w3),
}
)
approve_tx["gas"] = int(w3.eth.estimate_gas(approve_tx) * 1.2)
signed_approval = Account.sign_transaction(approve_tx, private_key)
approval_hash = w3.eth.send_raw_transaction(signed_approval.raw_transaction)
receipt = w3.eth.wait_for_transaction_receipt(approval_hash)
if receipt.status != 1:
raise RuntimeError(f"approval reverted: {approval_hash.hex()}")
tx = quote["tx"]
data = tx["data"]
if quote.get("permit2"):
typed_data = {
"domain": quote["permit2"]["domain"],
"types": quote["permit2"]["types"],
"primaryType": quote["permit2"]["primaryType"],
"message": quote["permit2"]["message"],
}
signed_permit = Account.sign_message(
encode_typed_data(full_message=typed_data),
private_key,
)
data = patch_permit2_signature(
tx["data"],
int(tx["signature_offset"]),
bytes(signed_permit.signature),
)
swap_tx = {
"from": taker,
"to": Web3.to_checksum_address(tx["to"]),
"data": data,
"value": int(tx.get("value", "0")),
"chainId": CHAIN_ID,
"nonce": w3.eth.get_transaction_count(taker),
**tx_fees(w3),
}
gas_estimate = w3.eth.estimate_gas(swap_tx)
swap_tx["gas"] = int(gas_estimate * 1.3)
signed_swap = Account.sign_transaction(swap_tx, private_key)
swap_hash = w3.eth.send_raw_transaction(signed_swap.raw_transaction)
print("swap_tx:", swap_hash.hex())