【Python】ZetaChain上のuniswapでswap

この記事では，ZetaChain(testnet)上に実装されたUniswapにPythonでアクセスしてswapする方法を紹介します．
環境はjupyterlabを想定しています．

解説

privateな情報は.envファイルから環境変数として読み込みます．

ADDRESS=0x...
PRIVATE_KEY=0x...

%load_ext dotenv
%dotenv .env -o -v

各種設定

ZetaChain上のtokenアドレス等を設定します．(公式リンク)

from web3 import Web3, HTTPProvider
import json, os, time

# RPCの設定
RPC_URL = "https://api.athens2.zetachain.com/evm"
w3 = Web3(HTTPProvider(RPC_URL))

# token contract address
GETH_ADDRESS = "0x91d18e54DAf4F677cB28167158d6dd21F6aB3921"
WZETA_ADDRESS = "0x5F0b1a82749cb4E2278EC87F8BF6B618dC71a8bf"

uniswapに関する情報を設定します．ABIはこちらから取得できます．

# uniswap contract address
FACTORY_ADDRESS = "0x9fd96203f7b22bCF72d9DCb40ff98302376cE09c"
ROUTER_ADDRESS = "0x2ca7d64A7EFE2D62A725E2B35Cf7230D6677FfEe"

# ABIの読み込み
with open("abis/IUniswapV2Pair.json", "r") as f:
    IUniswapV2Pair = json.loads(f.read())
with open("abis/IUniswapV2Factory.json", "r") as f:
    IUniswapV2Factory = json.loads(f.read())
with open("abis/IUniswapV2Router02.json", "r") as f:
    IUniswapV2Router02 = json.loads(f.read())
with open("abis/IERC20.json", "r") as f:
    IERC20 = json.loads(f.read())

factory = w3.eth.contract(address=FACTORY_ADDRESS , abi=IUniswapV2Factory["abi"])
router = w3.eth.contract(address=ROUTER_ADDRESS , abi=IUniswapV2Router02["abi"])

geth_wzeta_address = factory.functions.getPair(GETH_ADDRESS, WZETA_ADDRESS).call()
geth_wzeta = w3.eth.contract(address=geth_wzeta_address, abi=IUniswapV2Pair["abi"])

approve

GETHのapproveトランザクションを送信します．

# approve
my_address = os.getenv("ADDRESS")
my_private_key = os.getenv("PRIVATE_KEY")
geth = w3.eth.contract(address=GETH_ADDRESS, abi=IERC20["abi"])

geth_approve_amount = w3.to_wei('0.1', 'ether')

nonce = w3.eth.get_transaction_count(my_address)
tx_approve = geth.functions.approve(ROUTER_ADDRESS, geth_approve_amount).build_transaction({
    'chainId': 7001,
    'type': 2,
    'gas': 10 ** 6,
    'maxFeePerGas': w3.to_wei('300', 'gwei'),
    'maxPriorityFeePerGas': w3.to_wei('150', 'gwei'),
    'nonce': nonce
    })
signed_tx_approve = w3.eth.account.sign_transaction(tx_approve, private_key = my_private_key)
w3.eth.send_raw_transaction(signed_tx_approve.rawTransaction)

gETH to ZETA

# swapExactTokensForETH
amountIn = w3.to_wei('0.1', 'ether')
amountOutMin = 0

path = [GETH_ADDRESS, WZETA_ADDRESS]
to = my_address
deadline = int(time.time() + 60) # 2分以内に実行されなければrevert
nonce = w3.eth.get_transaction_count(my_address)
tx_swap = router.functions.swapExactTokensForETH(amountIn, amountOutMin, path, to, deadline).build_transaction({
    'chainId': 7001,
    'type': 2,
    'gas': 10 ** 6,
    'maxFeePerGas': w3.to_wei('300', 'gwei'),
    'maxPriorityFeePerGas': w3.to_wei('150', 'gwei'),
    'nonce': nonce, 
    })
signed_tx_swap = w3.eth.account.sign_transaction(tx_swap, private_key = my_private_key)
w3.eth.send_raw_transaction(signed_tx_swap.rawTransaction)

ZETA to gETH

# swapExactETHForTokens
value = w3.to_wei('0.1', 'ether')
amountOutMin = 0

path = [WZETA_ADDRESS, GETH_ADDRESS]
to = my_address
deadline = int(time.time() + 60 * 2)
nonce = w3.eth.get_transaction_count(my_address)
tx_swap = router.functions.swapExactETHForTokens(amountOutMin, path, to, deadline).build_transaction({
    'chainId': 7001,
    'type': 2,
    'gas': 10 ** 6,
    'maxFeePerGas': w3.to_wei('300', 'gwei'),
    'maxPriorityFeePerGas': w3.to_wei('150', 'gwei'),
    'nonce': nonce, 
    'value': value
    })
signed_tx_swap = w3.eth.account.sign_transaction(tx_swap, private_key=my_private_key)
w3.eth.send_raw_transaction(signed_tx_swap.rawTransaction)