UniswapV3 — Cycle Arbitrage Helper (Part II)

Pool State Tracking, Multi-Pool Calculation, Swap Input Generation

Continuing the development of the Cycle Arbitrage Helper, this post describes the addition of more important features.

The key improvements described here:

• Pool state tracking — the arb helper should maintain and detect the relevant pool states of every node in its swap path. To avoid unnecessary recalculation, the arb helpershould detect when the current pool states have not changed and stop early.

• Multi-pool swap calculation — the previous iteration performed two-pool arbs, but we want a helper that can analyze swaps along pool paths of arbitrary length.

• Swap input generation — the helper should provide a machine-readable report for the swap inputs at every point along the swap path. After all, the helper does not execute the arb itself, it merely simplifies execution of that arb by others.

Each feature will have a dedicated section, and we will implement them in parts.

Bonus — Class Name

The previous barebones helper is compatible with both Uniswap V2 and V3 pools, so naming it V3LpSwap is a bit clumsy. I have renamed it UniswapLpCycle, since this describes the DEX ecosystem, what contracts it uses, and the style of arbitrage it will perform.

Eventually this class will be robust enough to replace all the existing V2 helpers, as well as work for V3 pools. But I am beginning with the end in mind and trying to avoid choices that limit me.

Pool State Tracking

First, let’s identify the state we care about.

For V2, the important variables are the reserves for token0 and token1. All behavior of the pool can be predicted just by knowing the reserves, so we only need this.

For V3, the important variables are liquidity, sqrtPriceX96, and the current tick. The pool reserves do not matter, nor do the bookkeeping variables associated with fees. Therefore we will just track these three.

UniswapLpCycle will need access to these state variables, so we should provide an interface for it to use. Both LiquidityPool helper (V2) and V3LiquidityPool helper (V3) should have a common attribute name. I will call it state, and each pool will store it internally with a dictionary. This allows me to extend the dictionary later (if needed) without modifying the interface.

First, add the state attribute to the V2 helper inside its constructor:

class LiquidityPool:
    def __init__(...):

        [...]

        self.state = {
            "reserves_token0": self.reserves_token0,
            "reserves_token1": self.reserves_token1,
        }

        [...]

And inside the familiar method update_reserves:

def update_reserves(...):

    [...]
                self.state = {
                    "reserves_token0": self.reserves_token0,
                    "reserves_token1": self.reserves_token1,
                }
                return True
            else:
                return False

    [...]

    elif self._update_method == "external":

        [...]

            self.state = {
                "reserves_token0": self.reserves_token0,
                "reserves_token1": self.reserves_token1,
            }

        [...]

Now whenever the pool reserves are updated, state will be updated to match those values.

Similar deal for the V3 helper constructor:

class BaseV3LiquidityPool(ABC):

    [...]

    def __init__(...):

        [...]

        self.state = {
            "liquidity": self.liquidity,
            "sqrt_price_x96": self.sqrt_price_x96,
            "tick": self.tick,
        }

And add it to the auto_update method to keep it synced on updates:

def auto_update(
    self,
    silent: bool = True,
):
    """
    Retrieves the current slot0 and liquidity values from the LP,
    stores any that have changed, and returns a tuple with an 
    update status boolean and a dictionary holding the current state   
    values:
        - liquidity
        - sqrt_price_x96
        - tick
    """
    updated = False
    try:
        if (slot0 := self._brownie_contract.slot0()) != self.slot0:
            updated = True
            self.slot0 = slot0
            self.sqrt_price_x96 = self.slot0[0]
            self.tick = self.slot0[1]
        if (
            liquidity := self._brownie_contract.liquidity()
        ) != self.liquidity:
            updated = True
            self.liquidity = liquidity
    except:
        raise
    else:
        if updated:
            self.state = {
                "liquidity": self.liquidity,
                "sqrt_price_x96": self.sqrt_price_x96,
                "tick": self.tick,
            }
        return updated, self.state

Now within UniswapLpCycle, let’s add an attribute called pool_states and fill it with the values from each tracked helper:

class UniswapLpCycle(Arbitrage):
    def __init__(...):

        [...]

        self.pool_states = {}

And an internal method to update it when needed:

def _update_pool_states(self):
    """
    Internal method to update the `self.pool_states` state tracking dict
    """
    self.pool_states = {
        pool.address: pool.state for pool in self.swap_pools
    }

This is a dictionary comprehension that loops through self.swap_pools, adds a dictionary key for the pool address, and stores a copy of that pool’s internal state attribute.

The point of all this is to put a guard around the arbitrage calculation, so let’s modify calculate_arbitrage by adding a pre-calculation check, firing a call to update as needed, then proceeding:

def calculate_arbitrage(...):

    # short-circuit to avoid arb recalc if pool states have not changed:
    if self.pool_states == {
        pool.address: pool.state for pool in self.swap_pools
    }:
        return False, ()
    else:
        print("calculating arb...")
        self._update_pool_states()

    [...]

If the pool states have not changed, the dictionaries will match and the function can return early, skipping the calculation step.

I’ve added a debug print here just to make the behavior more obvious, but will remove it after testing.

Multi-Pool Calculation