Uniswap — Transaction Prediction (Part II)

Extending For V3

Feb 07, 2023

∙ Paid

I have continued to build out the UniswapTransaction helper skeleton to support more state-changing functions that we could observe in the mempool.

It’s a tedious task, but it’s made simpler by the fact that I’ve largely developed the algorithms already. We learned how to predict V2 pool behavior in the lesson covering Predicting the Effect of Mempool Transactions. I’m working on adding V2 functionality in the background, but that’s ground we’ve already covered so I will not go over it again. V3 is still fresh and relatively unexplored so I’m going to focus on it.

For a high-level overview, the UniswapTransaction helper must be able to recognize all known state-changing functions that will be sent to the V2 and V3 router contracts. These are:

V2 Router

V3 Router

All of the V2 functions have been previously defined, so I am re-implementing them and pushing updates to github.

The V3 functions, however, need to be built out.

I will build the swaps one-by-one, then finish off with multicall. Recall that a Uniswap Multicall Transaction can execute one or more state-changing transactions in a single call, so the decoder must be robust enough to process and track state changes associated with several different pools.

A further complication: the V3 router also has the ability to perform V2 swaps, both standalone and via multicall. Yikes!

It’s a big task but someone has to extract this value, so let’s go.

exactInputSingle

This is a simple function that swaps an exact amount of one token for a variable amount of another using a single pool.

Within the V3LiquidityPool function, we already have access to the ported swap function, via the hidden double-underscore method __UniswapV3Pool_swap. The function was previously set up to return just the amount0 & amount1 delta values, but it can be extended very simply to return the final state information.

Change:

return amount0, amount1

To:

return (
    amount0,
    amount1,
    state["sqrtPriceX96"],
    state["liquidity"],
    state["tick"],
)

And then define a new method called simulate_swap that accepts either a token_in or token_out, along with an associated amount:

def simulate_swap(
    self,
    token_in: Erc20Token = None,
    token_in_quantity: int = None,
    token_out: Erc20Token = None,
    token_out_quantity: int = None,
) -> Tuple[int, int, int, int, int]:

    assert (token_in and token_in_quantity) or (
        token_out and token_out_quantity
    )
    assert not (
        token_in and token_out
    ), "Incompatible options! Provide a token in/out and associated quantity, but not both"

    if token_in and token_in not in (self.token0, self.token1):
        raise LiquidityPoolError("token_in not found!")
    if token_out and token_out not in (self.token0, self.token1):
        raise LiquidityPoolError("token_out not found!")

    # determine whether the swap is token0 -> token1
    if token_in is not None and token_in_quantity:
        zeroForOne = True if token_in == self.token0 else False
    elif token_out is not None and token_out_quantity:
        zeroForOne = True if token_out == self.token1 else False

    try:
        # delegate calculations to the ported `swap` function
        (
            *_,
            end_sqrtprice,
            end_liquidity,
            end_tick,
        ) = self.__UniswapV3Pool_swap(
            zeroForOne=zeroForOne,
            amountSpecified=token_in_quantity
            if token_in_quantity
            else -token_out_quantity,
            sqrtPriceLimitX96=(
                TickMath.MIN_SQRT_RATIO + 1
                if zeroForOne
                else TickMath.MAX_SQRT_RATIO - 1
            ),
        )
    except EVMRevertError:
        raise
    else:
        return {
            "liquidity": end_liquidity,
            "sqrt_price_x96": end_sqrtprice,
            "tick": end_tick,
        }

Now inside the transaction helper, we handle the exactInputSingle swap case:

if transaction_func == "exactInputSingle":
    print(transaction_func)

    # decode with Router ABI - https://github.com/Uniswap/v3-periphery/blob/main/contracts/interfaces/ISwapRouter.sol
    # struct ExactInputSingleParams {
    #   address tokenIn;
    #   address tokenOut;
    #   uint24 fee;
    #   address recipient;
    #   uint256 deadline;
    #   uint256 amountIn;
    #   uint256 amountOutMinimum;
    #   uint160 sqrtPriceLimitX96;
    # }
    try:
        (
            tokenIn,
            tokenOut,
            fee,
            recipient,
            deadline,
            amountIn,
            amountOutMinimum,
            sqrtPriceLimitX96,
        ) = transaction_params.get("params")
    except:
        pass

    # decode with Router2 ABI - https://github.com/Uniswap/swap-router-contracts/blob/main/contracts/interfaces/IV3SwapRouter.sol
    # struct ExactInputSingleParams {
    #   address tokenIn;
    #   address tokenOut;
    #   uint24 fee;
    #   address recipient;
    #   uint256 amountIn;
    #   uint256 amountOutMinimum;
    #   uint160 sqrtPriceLimitX96;
    # }
    try:
        (
            tokenIn,
            tokenOut,
            fee,
            recipient,
            amountIn,
            amountOutMinimum,
            sqrtPriceLimitX96,
        ) = transaction_params.get("params")
    except:
        pass

    # get the V3 pool involved in the swap
    v3_pool = self.v3_pool_manager.get_pool(
        token_addresses=(tokenIn, tokenOut),
        pool_fee=fee,
    )

    print(f"pool located: {v3_pool}")
    print(f"Predicting output of swap through pool: {v3_pool}")

    starting_state = v3_pool.state
    final_state = v3_pool.simulate_swap(
        token_in=Erc20TokenHelperManager().get_erc20token(tokenIn),
        token_in_quantity=amountIn,
    )
    print(f"{starting_state=}")
    print(f"{final_state=}")

Pool Manager

You might have noticed an interesting line in that code block:

v3_pool = self.v3_pool_manager.get_pool(
    token_addresses=(tokenIn, tokenOut),
    pool_fee=fee,
)

What is this exactly?

I’ve been working hard this week, adding new features to the degenbot code base. One of these features is a set of helper managers that allow me to offload the task of creating and tracking various helpers. It can be very cumbersome to maintain the various token, pool, and arbitrage path helpers, especially when they are all pointing to each other.

For example, let’s say that I observe some transaction in the mempool. It’s a complicated Uniswap multicall that performs a series of V2 and V3 swaps between several unique pools and tokens. When I observe it, I may not have the required pool and token helper ready to simulate the transaction. I begin by looking at the tokens involved in the swap by comparing the addresses in the decoded transaction to the addresses of known token helpers. If I have those, I load them. If I don’t, I create them. Once that is ready, I identify which router the swap is using, and predict which pools that swap will go through. This means I have to know the liquidity pools deployed by the factory contract associated with that router address, as well as what tokens those pools hold. I don’t want to recreate any token or pool helpers along the way, because the bot will not know how to treat duplicates. Which state is correct? It’s a mess!

That’s the issue the managers are built to solve. Additionally, to solve the issue of duplicates, they implement a nice feature known as the Monostate Pattern. This ensures that each instance of a manager uses the same state data structure. Here’s an example:

>>> import degenbot as bot

# create a V2 LP manager and fetch a few pools

>>> uni_v2_lp_manager = bot.uniswap_managers.UniswapV2LiquidityPoolManager(
    factory_address='0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f'
)

>>> uni_v2_lp_manager.get_pool(
    pool_address='0xae461ca67b15dc8dc81ce7615e0320da1a9ab8d5'
)
• DAI (Dai Stablecoin)
• USDC (USD Coin)
DAI-USDC (V2, 0.30%)
• Token 0: DAI - Reserves: 17843041688041050871847703
• Token 1: USDC - Reserves: 17838422182290

>>> uni_v2_lp_manager.get_pool(
    pool_address='0xa478c2975ab1ea89e8196811f51a7b7ade33eb11'
)
• DAI (Dai Stablecoin)
• WETH (Wrapped Ether)
DAI-WETH (V2, 0.30%)
• Token 0: DAI - Reserves: 7455337228800814009290578
• Token 1: WETH - Reserves: 4570749232361418847112

>>> uni_v2_lp_manager.get_pool(
    pool_address='0x3041cbd36888becc7bbcbc0045e3b1f144466f5f'
)
• USDC (USD Coin)
• USDT (Tether USD)
USDC-USDT (V2, 0.30%)
• Token 0: USDC - Reserves: 5037294201600
• Token 1: USDT - Reserves: 5050682610716

# inspect the state data via the ._state attribute, then recreate the manager and verify the pools are still known

>>> uni_v2_lp_manager._state
{
    '0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f': {
        'factory_contract': 0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f,
        'pools_by_address': {
            '0x3041CbD36888bECc7bbCBc0045E3B1f144466f5f': USDC-USDT (V2, 0.30%),
            '0xA478c2975Ab1Ea89e8196811F51A7B7Ade33eB11': DAI-WETH (V2, 0.30%),
            '0xAE461cA67B15dc8dc81CE7615e0320dA1A9aB8D5': DAI-USDC (V2, 0.30%)
        },
        'pools_by_tokens': {
            ('0x6B175474E89094C44Da98b954EedeAC495271d0F', '0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48'): DAI-USDC (V2, 0.30%),
            ('0x6B175474E89094C44Da98b954EedeAC495271d0F', '0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2'): DAI-WETH (V2, 0.30%),
            ('0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48', '0xdAC17F958D2ee523a2206206994597C13D831ec7'): USDC-USDT (V2, 0.30%)
        }
    }
}

>>> del uni_v2_lp_manager

>>> uni_v2_lp_manager = bot.uniswap_managers.UniswapV2LiquidityPoolManager(
    factory_address='0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f'
)

>>> uni_v2_lp_manager._state
{
    '0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f': {
        'factory_contract': 0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f,
        'pools_by_address': {
            '0x3041CbD36888bECc7bbCBc0045E3B1f144466f5f': USDC-USDT (V2, 0.30%),
            '0xA478c2975Ab1Ea89e8196811F51A7B7Ade33eB11': DAI-WETH (V2, 0.30%),
            '0xAE461cA67B15dc8dc81CE7615e0320dA1A9aB8D5': DAI-USDC (V2, 0.30%)
        },
        'pools_by_tokens': {
            ('0x6B175474E89094C44Da98b954EedeAC495271d0F', '0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48'): DAI-USDC (V2, 0.30%),
            ('0x6B175474E89094C44Da98b954EedeAC495271d0F', '0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2'): DAI-WETH (V2, 0.30%),
            ('0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48', '0xdAC17F958D2ee523a2206206994597C13D831ec7'): USDC-USDT (V2, 0.30%)
        }
    }
}

The benefit here is not immediately obvious, but it means that I can use these Monostate helper managers across the code base to do on-demand fetching of tokens, pools, arbitrage paths, etc. without clobbering the helper state tracking dictionaries.

In fact, the execution client can now offload a lot of its business logic to the managers!

Now whenever I need a token, I can just call the get_erc20token method on the token manager. When I need a liquidity pool, I can call get_pool at the V2 or V3 pool manager. The pool managers themselves are also smart and will check if a requested helper is already available before recreating it.

Mempool Watcher Script

With this mostly ready to go, let’s build a simple watcher script that observes pending mainnet transactions to the Uniswap & Sushiswap router contracts, generates a UniswapTransaction helper, runs simulate on the transaction, then deletes it:

ethereum_pending_tx_watcher.py

Degen Code

Uniswap — Transaction Prediction (Part II)

Extending For V3

exactInputSingle

Pool Manager

Mempool Watcher Script

This post is for paid subscribers