Arbitrage

Background

Arbitrage is taking advantage of the price difference of the same asset in two different markets (pools in our example). The arbitrage agent calculates the difference in price for an asset and buys from the pool where it's valued less and sells at a pool where it's valued more.

Arbitrage ensures price efficiency across markets, as traders continuously buy and sell to capitalize on price discrepancies.

There are several forms of arbitrage — including pure arbitrage, statistical arbitrage, and triangular arbitrage — each varying in complexity and methodology. The key to successful arbitrage is speed, as price differences are often minimal and can disappear quickly due to market adjustments.

Arbitrage strategy demonstration

Arbitrage Strategy

If the last trade was too recent or the earning amount is less than the minimum amount (not enough profit to cover the trading fees), the agent doesn't execute any trades. Otherwise, the agent executes a trade, buying the token from the pool where it's valued less and selling the token in the pool where it's valued more.


How To Run

Installation

Follow our Getting Started guide to install the dojo library and other required tools.

Then clone the dojo_examples repository on GitHub using the following command on your terminal.

Terminal
git clone https://github.com/CompassLabs/dojo_examples.git

Go into the arbitrage directory to run our strategy.

Terminal
cd dojo_examples/examples/arbitrage

Running

Run the dojo_examples/start_dashboard.py script using the following command if you would like to access your dashboard.

Terminal
python start_dashboard.py

On another terminal window, copy the following command in the dojo_examples/examples/arbitrage folder.

Terminal
python run.py

This command will setup your local blockchain, contracts, accounts and agents. You can then access your Dojo dashboard at http://localhost:8051.


Step-By-Step Explanation

Initialization

We create a new class called ArbitragePolicy which inherits from the BasePolicy class. We set additional variables such as block_last_trade and min_block_dist which ensure that trades are not too recent as this would cause a higher price impact.

policy.py
class ArbitragePolicy(BasePolicy):
  """Arbitrage trading policy for a UniV3Env with two pools.
 
  :param agent: The agent which is using this policy.
  """
 
  def __init__(self, agent: BaseAgent) -> None:
      super().__init__(agent=agent)
      self.block_last_trade = -1
      self.min_block_dist = 20
      self.min_signal = 1.901
      self.tradeback_via_pool = None

Signal Calculation

This method identifies potential arbitrage opportunities between two pools by first verifying that both pools trade the same tokens. It then retrieves the token prices from each pool and calculates the price ratio between them. To ensure an accurate assessment of profitability, the method adjusts the price ratio by factoring in the transaction fees of both pools, resulting in the net arbitrage signals.

policy.py
def compute_signal(self, obs: UniV3Obs) -> Tuple[Decimal, Decimal]:
  pools = obs.pools
  pool_tokens_0 = obs.pool_tokens(pool=pools[0])
  pool_tokens_1 = obs.pool_tokens(pool=pools[1])
  assert (
      pool_tokens_0 == pool_tokens_1
  ), "This policy arbitrages same token pools with different fee levels."
 
  price_0 = obs.price(
      token=pool_tokens_0[0], unit=pool_tokens_0[1], pool=pools[0]
  )
  price_1 = obs.price(
      token=pool_tokens_0[0], unit=pool_tokens_0[1], pool=pools[1]
  )
  ratio = price_0 / price_1
  obs.add_signal(
      "Ratio",
      float(ratio),
  )
  signals = (
      ratio * (1 - obs.pool_fee(pools[0])) * (1 - obs.pool_fee(pools[1])),
      1 / ratio * (1 - obs.pool_fee(pools[0])) * (1 - obs.pool_fee(pools[1])),
  )
 
  obs.add_signal(
      "CalculatedProfit",
      float(max(signals)),
  )
 
  return signals

Trade Execution

Afterwards, we return a UniV3Trade object containing our order to buy/sell, specifying the pool, the quantity and the agent. If the last trade was too recent or no profit would be made, we return an empty list meaning no trade is being made.

policy.py
def predict(self, obs: UniV3Obs) -> List[BaseAction]:
  pools = obs.pools
  pool_tokens_0 = obs.pool_tokens(pool=pools[0])
  pool_tokens_1 = obs.pool_tokens(pool=pools[1])
  assert (
      pool_tokens_0 == pool_tokens_1
  ), "This policy arbitrages same token pools with different fee levels."
 
  # Agent will always be in USDC
  amount_0 = self.agent.quantity(pool_tokens_0[0])
  amount_1 = self.agent.quantity(pool_tokens_0[1])
 
  # Since we don't support multihop yet, we need to trade this way for now.
  if self.tradeback_via_pool is not None:
      action = UniV3Trade(
          agent=self.agent,
          pool=self.tradeback_via_pool,
          quantities=(Decimal(0), amount_1),
      )
      self.tradeback_via_pool = None
      return [action]
 
  signals = self.compute_signal(obs)
  earnings = max(signals)
  index_pool_first = signals.index(max(signals))
  pool = obs.pools[index_pool_first]
 
  # Don't trade if the last trade was too recent
  if (
      earnings < self.min_signal
      or obs.block - self.block_last_trade < self.min_block_dist
  ):
      return []
 
  # Make first trade
  self.tradeback_via_pool = (
      obs.pools[0] if index_pool_first == 1 else obs.pools[1]
  )
  self.block_last_trade = obs.block
  return [
      UniV3Trade(
          agent=self.agent,
          pool=pool,
          quantities=(amount_0, Decimal(0)),
      )
  ]

In the run.py file, we create two pools, a Uniswap environment and an agent to implement the arbitrage strategy.