Uniswap — Transaction Prediction (Part VII)

Transaction Simulation + Calculation Overrides = BFF

Building on the last mempool watcher, we’ll integrate the transaction simulator and our pre-generated arbitrage paths to assemble a mostly functional backrun searcher.

I’ve been refactoring some things in the degenbot repo, particularly the return value of the UniswapTransaction class and the override inputs to the UniswapLpCycle arbitrage helper class. Big thanks to reader and Discord MVP salparadi for the feedback and testing on this one!

Data Classes: Behind the Scenes

I’ve been changing some data structures in the code base to use Python data classes. They allow me a very clean way to implement “struct-like” behavior without a lot of boilerplate code. As a bonus, they perform a sort of soft data validation, since I can specify required and optional attributes inside the constructor. They’re also faster and more efficient than the typical Python dictionary.

As an example, check out the UniswapV3PoolState class:

@dataclasses.dataclass(slots=True)
class UniswapV3PoolState:
    liquidity: int
    sqrt_price_x96: int
    tick: int
    tick_bitmap: Optional[Dict] = None
    tick_data: Optional[Dict] = None

The class is wrapped by a method called dataclass which will do the work of converting my simple class skeleton into a fully-fledged object with many nice magic methods for comparison, string and repr comparison, hashing, etc.

But the key reason you should care is that the class can be referenced in several places throughout the code base and extended as needed without needing to do find & replace edits.

For an example of this class being used, check out this V3LiquidityPool method:

class V3LiquidityPool(PoolHelper):

    [...]

    def calculate_tokens_out_from_tokens_in(
        self,
        token_in: Erc20Token,
        token_in_quantity: int,
        override_state: Optional[UniswapV3PoolState] = None,
        with_remainder: bool = False,
    ) -> Union[int, Tuple[int, int]]:

        [...]

You’ll notice that the type hint for the argument override_state is an optional UniswapV3PoolState object. Whenever a pool state object is provided to the method, it will use the liquidity, sqrtPrice, and tick values (and some day can use overridden tick bitmaps and liquidity values) in its calculation.

Take another example from UniswapTransaction:

class UniswapTransaction(TransactionHelper):

    [...]

    def simulate(
        self,
        silent: bool = False,
    ) -> List[
        Union[
            Tuple[LiquidityPool, UniswapV2PoolSimulationResult],
            Tuple[V3LiquidityPool, UniswapV3PoolSimulationResult],
        ]
    ]:

        [...]

That’s a lot of type hinting, so I want to draw attention to the stuff after the arrow. The simulate method will return a list of tuples. Each tuple will contain either a LiquidityPool paired with a UniswapV2PoolSimulationResult or a V3LiquidityPool paired with a UniswapV3PoolSimulationResult. As you can imagine, the sim results are also data classes that contain attributes that matter for the respective pool.

The point isn’t to be fancy with classes, but to enable cleaner data transfer between related methods.

It shouldn’t surprise you to find that the UniswapLpCycle helper accepts a similar override:

class UniswapLpCycle(ArbitrageHelper):

    [...]

    def calculate_arbitrage(
        self,
        override_state: Optional[
            Sequence[
                Union[
                    Tuple[
                        LiquidityPool, 
                        UniswapV2PoolState,
                    ],
                    Tuple[
                        LiquidityPool, 
                        UniswapV2PoolSimulationResult,
                    ],
                    Tuple[
                        V3LiquidityPool, 
                        UniswapV3PoolState,
                    ],
                    Tuple[
                        V3LiquidityPool, 
                        UniswapV3PoolSimulationResult,
                    ],                     
                ]
            ]
        ] = None,
    ) -> Tuple[bool, Tuple[int, int]]:

        [...]

NOTE: if you’ve never seen Union in a type hint, it is a backwards-compatible way to specify an “or” relationship. So for example, Union[int,bool] means that the type for this argument can be an int or a bool. Newer versions of Python accept the more familiar | operator.

These type hints reveal that the calculate_arbitrage method accepts an override argument consisting of a sequence of tuples. These tuples should have a pool object in the first position, and a pool state or simulation result in the second position. The method accepts both as a convenience to the user, because it’s cumbersome to unwrap a simulation result, extract the future pool state, and then re-wrap it into the list of tuples. If simulation results are provided, the method will extract the pool state for you before proceeding.

The point of this backend work is to turn this big ball of spaghetti into a simple implementation (written in pseudo-code):

sim_result = TransactionHelper(mempool_tx).simulate()

backrun_result = arb_helper.calculate_arbitrage(
    override_state=sim_result
)

if backrun_arb_result.is_profitable():
    execute(arb_helper)

Critical Components

To achieve this, we need a distinct set of pieces that work together:

• An event watcher to keep pool states updated

• A mempool watcher to find pending transactions

• A set of arbitrage paths to check against those simulations

• A transaction simulator to evaluate those pending transactions

The watcher and simulator were introduced in previous posts in the series, and we’ve worked with many arbitrage path generators already.

So let’s wire them all together and see if any backruns fall out!

Before we proceed, let’s manage some expectations. This will not be a complete backrun-capable bot. It is a simple scanner that will find, calculate, and report possible backruns. It won’t maintain a pool of transactions, won’t perform any transaction reordering, and will only evaluate the pending transaction against the curent block once, before discarding it.

In short: it’s a basic example of how to integrate the components, and the full feature set will be built after the concept has been proven.

Component: Event Watcher

This is a standard component that we’ve used in several other projects. It’s lightly modified to reference the centralized pool tracker developed earlier and managed by the AllPools helper.

Component: Mempool Watcher

This is another standard component. It subscribes to new pending transactions via a websocket, and filters them for processing by the transaction simulator.

Component: Arb Path Loader

To limit complexity, we will limit the exercise to 2-pool arbitrage paths. Feel free to extend to 3-pool and beyond if you like.

On mainnet Ethereum there are two main DEX: Uniswap and Sushiswap. Both offer V2 pools and V3 pools, so we have four sets to work with.

This uses a familiar set of JSON files generated by the LP fetchers:

• ethereum_lps_sushiswapv2.json

• ethereum_lps_sushiswapv3.json

• ethereum_lps_uniswapv2.json

• ethereum_lps_uniswapv3.json

And the set of 2-pool paths:

• ethereum_arbs_2pool.json

If you don’t have these, navigate to the Discord #files channel to retrieve the latest versions.

At the time I’m writing this, the 2-pool path builder finds 15,520 total arbitrage paths between the four sets.

The function will generate four LP managers:

• univ2_lp_manager = bot.UniswapV2LiquidityPoolManager(
factory_address=(
"0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f"
)

• univ3_lp_manager = bot.UniswapV3LiquidityPoolManager(
factory_address=(
"0x1F98431c8aD98523631AE4a59f267346ea31F984"

  )

• sushiv2_lp_manager = bot.UniswapV2LiquidityPoolManager(
factory_address=(
"0xC0AEe478e3658e2610c5F7A4A2E1777cE9e4f2Ac"

  )

• sushiv3_lp_manager = bot.UniswapV3LiquidityPoolManager(
factory_address=(
"0xbACEB8eC6b9355Dfc0269C18bac9d6E2Bdc29C4F"

  )

Then arbitrage path loading becomes a simple process:

1. Identify the pool type

2. Ask the appropriate LP manager for that pool helper

3. Deliver the pool helpers to the arbitrage helper constructor

Component: Transaction Simulation / Arbitrage Path Matching

With the arb helpers built, it is a simple thing to extend the transaction simulation routine to implement the plan outlines in the pseudo-code above.

When the transaction simulator delivers a result, it contains a bundle of data in the form of a pool helper and a simulation result.

Getting from a set of pools to a set of arbitrage paths requires some finesse, but is otherwise simplified by the Python built-in set type.

A set is an unsorted collection that has a unique property. It can only ever contain one copy of a given item. For example:

>>> l = [1,2,2,3,3,3,4,4,4,4]

>>> l
[1, 2, 2, 3, 3, 3, 4, 4, 4, 4]

>>> set(l)
{1, 2, 3, 4}

You can use sets to identify when two collections share at least one item. The operation for this is called an intersection, and the set operator for an intersection is &.

For example:

>>> set([1,2,3]) & set([1])
{1}

We can use sets to identify arbitrage paths as well. If you have a set of arbitrage paths, each containing a reference to the pool helpers used in that path, you can compare the intersection of those pools to the pools in the simulation result:

>>> set(arb_helper.swap_pools) & set([pool for pool, _ in sim_results])]

If the result of the set intersection is empty, that indicates that the arbitrage helper has no pools in common with the simulation results. If the result is non-empty, the arbitrage helper will be affected by the simulated transaction.

So it becomes simple to check all of the arb helpers against the simulation, and group them in a list with a comprehension:

arb_helpers = [
    arb for arb in all_arbs
    if set(arb.swap_pools) & set([pool for pool, _ in sim_results])
]

NOTE: there are ways to improve performance (caching and pre-hashing), but this simplistic example is useful to prove the concept.

Then simply iterate over arb_helpers, perform the calculation with the simulation results, and check the results for profit:

for arb_helper in arb_helpers:
    try:
        profitable, (
            swap_amount,
            profit_amount,
        ) = arb_helper.calculate_arbitrage(
            override_state=sim_results
        )
    except:
        # ignore calculation errors
        continue

    if profitable and profit_amount >= MIN_PROFIT_ETH:
        do_something()

Example Backrun Watcher

The watcher below is useful as a passive read-only demonstration. It will find and print out lots of tiny opportunities, but do not get over-excited and start trying to turn these into real transactions.

It will display all theoretically profitable transactions, but these ignore gas costs completely. I need to write some tests to validate that all methods behave correctly when overrides are applied, and I also need to verify that applying an override does not put the helper into an inconsistent state afterwards.

That’s my focus this week, so the work will be done and then we’ll all be backrunning with ease.