Degen Code

Share this post

Project: UniswapV3 Mempool Watcher

www.degencode.com

Project: UniswapV3 Mempool Watcher

🦄 👀 🤖

BowTiedDevil
Oct 22, 2022
∙ Paid
11
Share this post

Project: UniswapV3 Mempool Watcher

www.degencode.com
7
Share

The majority of transactions on UniswapV3 go through the Router, and the majority of those are using the Multicall functionality.

If we want to use this DEX for fun profit, we need to build a means to observe and decode multicall transactions that take this submission path.

This project will be limited in scope, since implementing a full UniswapV3-compatible bot is complex. I prefer to build the individual pieces, test in isolation, then integrate them.

Our goal is to build a bot that:

  • Connects to mainnet Ethereum via websocket

  • Sets up an eth_subscribe watcher to receive new pending transactions

  • Decodes and prints the function and parameters associated with any observed UniswapV3 transaction

To understand the techniques in this lesson, be sure you’ve reviewed and understood the previous lesson on Asychronous Websocket Listeners, the lesson on Listening to Mempool Transactions, and the explorations of the UniswapV3 Router and Multicall contracts.

The bot will be presented in several sections that concentrate on implementing a focused piece. Complete source code will be presented at the end.

This will be an asynchronous bot that uses asyncio coroutines to achieve concurrency.

Make sure that you have an up-to-date websockets module installed in your virtual environment via pip. To ensure the latest, run pip install --upgrade websockets and confirm that you have major version 10 at minimum. I am using version 10.3 at the time of writing.

Imports and Setup

Import some modules, establish a connection to our RPC via Brownie, and launch the watch_pending_transactions coroutine (to be defined soon):

FILENAME: ethereum_univ3_watcher.py

import asyncio
import brownie
import itertools
import json
import os
import sys
import web3
import websockets


BROWNIE_NETWORK = "mainnet-local-ws"
WEBSOCKET_URI = "ws://localhost:8546"
ETHERSCAN_API_KEY = "[edit me]"

# Create a reusable web3 object (no arguments to WebsocketProvider
# will default to localhost on the default port)
w3 = web3.Web3(web3.WebsocketProvider())

os.environ["ETHERSCAN_TOKEN"] = ETHERSCAN_API_KEY

try:
    brownie.network.connect(BROWNIE_NETWORK)
except:
    sys.exit(
        "Could not connect! Verify your Brownie network settings using 'brownie networks list'"
    )

asyncio.run(watch_pending_transactions())

Connect and Watch

We begin by defining a asynchronous coroutine called watch_pending_transactions that established an RPC subscription to "newPendingTransactions", listens to the websocket for new messages, filters the messages to identify transactions to the UniswapV3 router contracts (Both Router and Router 2), then prints them.

The addresses we care about:

  • Router: 0xE592427A0AEce92De3Edee1F18E0157C05861564

  • Router2: 0x68b3465833fb72a70ecdf485e0e4c7bd8665fc45

async def watch_pending_transactions():

    v3_routers = {
        "0xE592427A0AEce92De3Edee1F18E0157C05861564": {
            "name": "UniswapV3: Router"
        },
        "0x68b3465833fb72A70ecDF485E0e4C7bD8665Fc45": {
            "name": "UniswapV3: Router 2"
        },
    }

    for router_address in v3_routers.keys():
        try:
            router_contract = brownie.Contract(
                router_address
            )
        except:
            router_contract = brownie.Contract.from_explorer(
                router_address
            )
        else:
            v3_routers[router_address]["abi"] = router_contract.abi
            v3_routers[router_address]["web3_contract"] = w3.eth.contract(
                address=router_address,
                abi=router_contract.abi,
            )

        try:
            factory_address = w3.toChecksumAddress(router_contract.factory())
            factory_contract = brownie.Contract(factory_address)
        except:
            factory_contract = brownie.Contract.from_explorer(factory_address)
        else:
            v3_routers[router_address]["factory_address"] = factory_address
            v3_routers[router_address]["factory_contract"] = factory_contract

    print("Starting pending TX watcher loop")

    async for websocket in websockets.connect(uri=WEBSOCKET_URI):

        try:
            await websocket.send(
                json.dumps(
                    {
                        "id": 1,
                        "method": "eth_subscribe",
                        "params": ["newPendingTransactions"],
                    }
                )
            )
        except websockets.WebSocketException:
            print("(pending_transactions) reconnecting...")
            continue
        except Exception as e:
            print(e)
            continue
        else:
            subscribe_result = json.loads(await websocket.recv())
            print(subscribe_result)

        while True:

            try:
                message = json.loads(await websocket.recv())
            except websockets.WebSocketException as e:
                print("(pending_transactions inner) reconnecting...")
                print(e)
                break  # escape the loop to reconnect
            except Exception as e:
                print(e)
                break

            try:
                pending_tx = dict(
                    w3.eth.get_transaction(
                        message.get("params").get("result")
                    )
                )
            except:
                # ignore any transaction that cannot be found
                continue

            # skip post-processing unless the TX was sent to 
            # an address on our watchlist
            if pending_tx.get("to") not in v3_routers.keys():
                continue
            else:
                try:
                    # decode the TX using the ABI
                    decoded_tx = (
                        v3_routers.get(
                            w3.toChecksumAddress(pending_tx.get("to"))
                        )
                        .get("web3_contract")
                        .decode_function_input(pending_tx.get("input"))
                    )
                except Exception as e:
                    continue
                else:
                    func, func_args = decoded_tx

            print(f'func: {func.fn_name}')
            print(f'args: {func_args}')

Run this and you’ll start to see a steady stream of UniswapV3 mempool transactions:

(.venv) devil@hades:~/bots$ /home/devil/bots/.venv/bin/python /home/devil/bots/ethereum_univ3_watcher.py

Starting pending TX watcher loop
{'jsonrpc': '2.0', 'id': 1, 'result': '0x6296f025a6c697b2d0257ba28a4734f0'}

func: multicall
args: {'deadline': 1666462787, 'data': [b"G+C\xf3\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x03x-\xac\xe9\xd9\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x0f\xb9}\xa4\x84\x015\xd3\xdb\x1f\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x80\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x8f[v\xe8\xf7\x15R;\xf3\x88\xff\xda\xa4\x1e\xde\r\xe4\x87d\xb8\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x02\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xc0*\xaa9\xb2#\xfe\x8d\n\x0e\\O'\xea\xd9\x08<ul\xc2\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00&\xa9\xc0\xc6\xf2\x8a\xe2\xe6\x92p\xbc9\xf1=\xb8z'\xdbL\xe5"]}
func: multicall
args: {'deadline': 1666462787, 'data': [b"Bq*g\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01>R\xb9\xab\xe0\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x02tmW\xaf\xab\x01\xd0\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x80\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xdc\xcd\x9d&}\x19\x94\x89\x82\x03\x0b\x9f\xa9\x07\x0eK\x14\x17\xdd!\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x02\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xc0*\xaa9\xb2#\xfe\x8d\n\x0e\\O'\xea\xd9\x08<ul\xc2\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x8a5\ni \xc0\xe6\x82\x16\n\x02\xfd#\xee\x86\x90o\x1aU\xf3", b'\x12!\x0e\x8a']}
func: multicall
args: {'deadline': 1666462787, 'data': [b"\x04\xe4Z\xaf\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x0c\xba`\xca^\xf4\xd4/\x92\xa5\x07\n\x8f\xed\xd1;\xe9>(a\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xc0*\xaa9\xb2#\xfe\x8d\n\x0e\\O'\xea\xd9\x08<ul\xc2\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'\x10\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x02\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x05\xb7Z\xb6A$\xc1\x18\x1c]\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x06a@\xe5\xb7[\xc8\xa8\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00", b"I@K|\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x06a@\xe5\xb7[\xc8\xa8\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00e/[\xf6\xa0'x\x8c\xdb\xf1jj?){\x9b5\x94\xc1\xf9"]}
func: multicall
args: {'deadline': 1666462163, 'data': [b"G+C\xf3\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00#\x86\xf2o\xc1\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00A\xbe\x16\x08?\x1e\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x80\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x99%\xe3x\x1c[\x8d\xdb\xd6L\x887\x94\xb4\xd1v8\x17z'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x02\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xc0*\xaa9\xb2#\xfe\x8d\n\x0e\\O'\xea\xd9\x08<ul\xc2\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x07\x01z\xb5i\x1b\xe5\x0eUa\xe9\xd6\xb6R\xb65OH\x9d\x80"]}
func: multicall
args: {'deadline': 1666462787, 'data': [b"G+C\xf3\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\r\xe0\xb6\xb3\xa7d\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x05\\\xdcR\x9c\xc1!\xfa\xf8\xb7\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x80\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00t\xbd\x05\x12=\x15\xec=w-[$-\xe4\x17\x96#\xdc;d\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x02\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xc0*\xaa9\xb2#\xfe\x8d\n\x0e\\O'\xea\xd9\x08<ul\xc2\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xa8\xc8\xcf\xb1A\xa3\xbbY\xfe\xa1\xe2\xeaky\xb5\xec\xbc\xd7\xb6\xca"]}

[...]

Neat, though not immediately useful.

Now let’s improve the function by decoding the function names and arguments, instead of just printing them. Start with the four single-purpose swapping functions:

  • exactInputSingle

  • exactInput

  • exactOutputSingle

  • exactOutput

Replace this code block:

print(f'func: {func.fn_name}')
print(f'args: {func_args}')

With this:

if func.fn_name == "multicall":
    print(func.fn_name)
    print(func_args.get("params"))
elif func.fn_name == "exactInputSingle":
    print(func.fn_name)
    print(func_args.get("params"))
elif func.fn_name == "exactInput":
    print(func.fn_name)
    print(func_args.get("params"))
elif func.fn_name == "exactOutputSingle":
    print(func.fn_name)
    print(func_args.get("params"))
elif func.fn_name == "exactOutput":
    print(func.fn_name)
    print(func_args.get("params"))
else:
    print(f"other function: {func.fn_name}")
    continue

Running this, we see some more readable function info:

(.venv) devil@hades:~/bots$ /home/devil/bots/.venv/bin/python /home/devil/bots/ethereum_univ3_watcher.py

Starting pending TX watcher loop
{'jsonrpc': '2.0', 'id': 1, 'result': '0xc18082aeac4cd977b13c7eec3ed128df'}

exactInputSingle
('0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2', '0x761D38e5ddf6ccf6Cf7c55759d5210750B5D60F3', 10000, '0xac778E227d7a901098C1626653C55f510eF88e83', 1620917607, 7056610575000000, 41082914344639370021254586, 0)

exactInput
(b"\xc0*\xaa9\xb2#\xfe\x8d\n\x0e\\O'\xea\xd9\x08<ul\xc2\x00\x01\xf4\xa0\xb8i\x91\xc6!\x8b6\xc1\xd1\x9dJ.\x9e\xb0\xce6\x06\xebH\x00\x0b\xb8kLz^?\x0b\x99\xfc\xd8>\x9c\x08\x9b\xdd\xd6\xc7\xfc\xe5\xc6\x11", '0xFCA96792f58053C6B2B399dD93c7d63f220d8131', 1627477281, 30000000000000000, 949176846757898446)

exactInputSingle
('0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2', '0x761D38e5ddf6ccf6Cf7c55759d5210750B5D60F3', 3000, '0x452a8D14999039dD338a1B3dbd00e180DD8Cf988', 1635667624, 2800000000000000, 5650482210574539262239092, 0)

exactInputSingle
('0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2', '0x95aD61b0a150d79219dCF64E1E6Cc01f0B64C4cE', 10000, '0xE502F62192caF93A49fb8264974B7775aB0FFa2F', 1620534299, 2000000000000000, 541746494200261181199796, 0)

exactInputSingle
('0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2', '0x95aD61b0a150d79219dCF64E1E6Cc01f0B64C4cE', 3000, '0x8B9CC1ff7d4d5E1443C9688270eB598BacD61F8B', 1620662876, 44000000000000000, 6085054860184434415693067, 0)

exactInputSingle
('0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2', '0xd1420AF453Fd7bf940573431D416cAcE7fF8280c', 3000, '0x9d1a2886d80576261bCFcacFe86CD773F1EB6b23', 1666462133, 120001706730658130, 1117715804039971864576, 0)

[...]

You’ll notice that the function arguments for exactInputSingle are perfectly readable and require no further processing.

Expanding exactInput

The arguments for exactInput are a little more obscure, so we can do some work to clean those up. From the Router exploration, we learned that the argument to exactInput is a struct named ExactInputParams, defined as:

  • struct ExactInputParams {

    bytes path;

    address recipient;

    uint256 deadline;

    uint256 amountIn;

    uint256 amountOutMinimum;

    }

The tricky part of decoding the path argument is manipulating all the bytes. Recall from the Router lesson that path is a concatenated string of bytes, alternating between the pool address and fee values:

Now we will decode path by manipulating the byte string.

Replace this code block:

elif func.fn_name == "exactInput":
    print(func.fn_name)
    print(func_args.get("params"))

With this:

This post is for paid subscribers

Already a paid subscriber? Sign in
© 2024 BowTiedDevil
Privacy ∙ Terms ∙ Collection notice
Start WritingGet the app
Substack is the home for great culture
Share