[メモ] Web3.pyでgetAmountOuts, quoteExactOutputを行う

おはこんばんにちは。
DMMビットコインから482億円分流出や、TGE後のゴミ対応プロジェクトが出たり、相場が上がりそうなのに急に下げたりと。
なんか疲弊する市場ですね。

ってことで、今回は皆さん一度は憧れるDEX間アービトラージbotの監視ツールを作成したいと思います（完全に意味不明）

まず前提として、こんな監視ツールじゃ儲からないです
手数料、スリッページ、フロントラン等のriskをフル無視しています。

皆さんの心の声

まあ、逆に捉えてそれらを考慮すればワンチャン。。って感じです。

ただ、DEXbotは修羅の道。
いくら早い言語にしてもロジが優秀でも、結局は通信速度とTX承認スピードが求められます。

これを乗り越えられるなら儲かるかも？（白目）

環境構築

python3が既にインストール済みなことを確認してください。
もしくは、GoogleColabやJupiterNotebook等を使用すれば予め環境が整っている上でビルドができます。

ディレクトリを作成し、pyenv(仮想環境)を立ち上げます。

!mkdir DexWatcher
!cd DexWatcher
!python3 -m venv myvenv
!python3 source myvenv/bin/activate

必要なライブラリをインストール

!pip install web3

以上です。

単体のコード

UniswapV2単体のgetAmountOuts

v2{.}py

import json
from web3 import Web3

with open("config.json") as f:
    config = json.load(f)

web3 = Web3(Web3.HTTPProvider(config["INFURA_RPC_URL"]))

with open("UniV2Router02abi.json") as f:
    UNISWAPV2_ROUTER_ABI = json.load(f)

router_contract = web3.eth.contract(address=config["UNI_V2_ROUTER_ADDRESS"], abi=UNISWAPV2_ROUTER_ABI)

buy_path = [config["TOKEN_A_ADDRESS"], config["TOKEN_B_ADDRESS"]]
amount_to_buy_for = 1 * 10**18

token_a_amount, token_b_amount = router_contract.functions.getAmountsOut(amount_to_buy_for, buy_path).call()

print(f"for {token_a_amount/10**18} Token A you would get {token_b_amount/10**18} Token B")

token_b_to_buy = 500 * 10**18
token_a_amount, token_b_amount = router_contract.functions.getAmountsIn(token_b_to_buy, buy_path).call()
print(f"for {token_b_to_buy/10**18} Token B you have to pay {token_a_amount/10**18} Token A")

config{.}json

{
    "INFURA_RPC_URL": "https://mainnet.infura.io/v3/XXXXXXXXXXXXXXX",
    "UNI_V2_ROUTER_ADDRESS": "0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D", 
    "TOKEN_A_ADDRESS": "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2", //wETH
    "TOKEN_B_ADDRESS": "0x1f9840a85d5aF5bf1D1762F925BDADdC4201F984" //UNI
}

ABI (UniV2Router02abi) に関しては下記のetherscanからjson formatでexportしましょう

今回はWETHをTokenA、UNIをTokenBに指定しました。
数量は1ETH, 500UNIでそれぞれ2ルートで行います。

ここで少し補足ですが、ERC20トークンの種別によって*DECIMALS*と呼ばれる小数点桁数を指定している数値が異なっています。

UniTokenは18

基本的には18桁ですが、USDTやUSDCなどは6桁とのこと。
なので、input, outputによって使用するトークンが何DECIMALSなのか調べた上で随時変更しましょう

10**18 -> 10**8 -> 10**6

それはさておき、実行すると下記のようなログが出ました。

for 1.0 Token A you would get 374.5553239485418 Token B
for 500.0 Token B you have to pay 1.3354533259941952 Token A

1ETH = 374.55 UNI
500 = 1.33ETH

UniswapV3単体のquoteExactOutput

v3{.}py

import json
from web3 import Web3
import eth_abi.packed

with open("config.json") as f:
    config = json.load(f)

web3 = Web3(Web3.HTTPProvider(config["INFURA_RPC_URL"]))

with open("UniV3Quoter2abi.json") as f:
    UNISWAP_V3_QUOTER2_ABI = json.load(f)

quoter2_contract = web3.eth.contract(address=config["UNISWAP_V3_QUOTER2_ADDRESS"], abi=UNISWAP_V3_QUOTER2_ABI)

path = eth_abi.packed.encode_packed(['address','uint24','address'], [config["TOKEN_A_ADDRESS"], 3000, config["TOKEN_B_ADDRESS"]])
amount_to_buy_for = 1 * 10**18

amount_out, sqrtPriceX96After, initializedTicksCrossed, gasEstimate = quoter2_contract.functions.quoteExactOutput(path, amount_to_buy_for).call()
print(f"for {amount_to_buy_for/10**18} Token A you would get {amount_out/10**18} Token B")

path = eth_abi.packed.encode_packed(['address','uint24','address'], [config["TOKEN_B_ADDRESS"], 3000, config["TOKEN_A_ADDRESS"]])
token_b_to_buy = 500 * 10**18

amount_out, sqrtPriceX96After, initializedTicksCrossed, gasEstimate = quoter2_contract.functions.quoteExactInput(path, token_b_to_buy).call()
print(f"for {token_b_to_buy/10**18} Token B you have to pay {amount_out/10**18} Token A")

config{.}json

{
    "INFURA_RPC_URL": "https://mainnet.infura.io/v3/9921258f942545febeb8642d511a0156",
    "UNISWAP_V3_QUOTER2_ADDRESS": "0x61fFE014bA17989E743c5F6cB21bF9697530B21e",
    "TOKEN_A_ADDRESS": "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2",
    "TOKEN_B_ADDRESS": "0x1f9840a85d5aF5bf1D1762F925BDADdC4201F984",
}

abi (UniV3Quoter2abi)

https://etherscan.io/address/0x61fFE014bA17989E743c5F6cB21bF9697530B21e#code

Output (V2と同じ条件)

for 1.0 Token A you would get 376.6474772560777 Token B
for 500.0 Token B you have to pay 1.3274642508756669 Token A

1ETH = 376.64UNI
500UNI = 1.32ETH

二者間アビトラの監視コード（マージ）

UniswapV2 -> UniswapV3
UniswapV3 -> UniswapV2

双方を10秒に一回並列シミュレーションする二者間アビトラのコードです

import json
from web3 import Web3
import eth_abi.packed
import time
from concurrent.futures import ThreadPoolExecutor

with open("config.json") as f:
    config = json.load(f)

web3 = Web3(Web3.HTTPProvider(config["INFURA_RPC_URL"]))

with open("UniV2Router02abi.json") as f:
    UNISWAPV2_ROUTER_ABI = json.load(f)

with open("UniV3Quoter2abi.json") as f:
    UNISWAP_V3_QUOTER2_ABI = json.load(f)

router_contract = web3.eth.contract(address=config["UNI_V2_ROUTER_ADDRESS"], abi=UNISWAPV2_ROUTER_ABI)
quoter2_contract = web3.eth.contract(address=config["UNISWAP_V3_QUOTER2_ADDRESS"], abi=UNISWAP_V3_QUOTER2_ABI)

def simulate_arbitrage(direction):
    initial_eth = 1
    eth_amount = initial_eth * 10**18

    if direction == "v2_to_v3":
        # UniswapV2: TokenA -> TokenB
        buy_path = [config["TOKEN_A_ADDRESS"], config["TOKEN_B_ADDRESS"]]
        token_a_amount, token_b_amount = router_contract.functions.getAmountsOut(eth_amount, buy_path).call()

        # UniswapV3: TokenB -> TokenA
        path = eth_abi.packed.encode_packed(['address','uint24','address'], [config["TOKEN_B_ADDRESS"], 3000, config["TOKEN_A_ADDRESS"]])
        amount_out, sqrtPriceX96After, initializedTicksCrossed, gasEstimate = quoter2_contract.functions.quoteExactInput(path, token_b_amount).call()

        profit_loss = (amount_out - eth_amount) / 10**18
        return f"Profit/Loss (UniV2 -> UniV3): {profit_loss} ETH"
    else:
        # UniswapV3: TokenA -> TokenB
        path = eth_abi.packed.encode_packed(['address','uint24','address'], [config["TOKEN_A_ADDRESS"], 3000, config["TOKEN_B_ADDRESS"]])
        token_b_amount, sqrtPriceX96After, initializedTicksCrossed, gasEstimate = quoter2_contract.functions.quoteExactInput(path, eth_amount).call()

        # UniswapV2: TokenB -> TokenA
        sell_path = [config["TOKEN_B_ADDRESS"], config["TOKEN_A_ADDRESS"]]
        token_a_amount_out = router_contract.functions.getAmountsOut(token_b_amount, sell_path).call()[-1]

        profit_loss = (token_a_amount_out - eth_amount) / 10**18
        return f"Profit/Loss (UniV3 -> UniV2): {profit_loss} ETH"

with ThreadPoolExecutor(max_workers=2) as executor:
    while True:
        print(f"Arbitrage simulation at {time.strftime('%Y-%m-%d %H:%M:%S')}")
        futures = [executor.submit(simulate_arbitrage, direction) for direction in ["v2_to_v3", "v3_to_v2"]]
        results = [f.result() for f in futures]
        for result in results:
            print(result)
        print("---")
        time.sleep(10)

Output

余裕でマイナスです！！

10秒ごとに繰り返して、監視の基本部分はできているので、あとはパス指定さえ好みに変えていけばOKです。

三者間アビトラのexampleまで書く気力がないので、あとはお任せします！（投げやりで草）

言い訳

正直、新規ペアを取得するためにmempoolから監視してcreate pairの関数を読み取り、decodeしてパス取るのも良いのですが、ハニーポットのリスクがまだ少なからずあるので、riskが高いです。

そういや、ハニポで捕まった中華人大学生おったな。

首例发行虚拟币涉刑案引争议：撤回流动性致炒币亏损是否构成诈骗_一号专案_澎湃新闻-The Paper

そう思うと、足元には常に地雷があって競争が激しい世界で利益を継続的に取るのは不可能説が個人的にはあります。

そもそも人間辞めないとマジでムリだと思います。(丸一日継続的に研究できるタイプとか。）

とか言う自分も研究者気質が少しある方だと自負してるのですが、メインのCEXbotに1日10時間費やせても、DEXに関しては稼げる見込みがなさ過ぎて途方に暮れてます。。w

だから僕はDEXbotを辞めた

結論

柔軟に生きていこう

現場からは以上です。