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
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 and go into the relevant directory.
git clone https://github.com/CompassLabs/dojo_examples.git
cd dojo_examples/examples/arbitrageRunning
Download the dashboard to view the simulation results.
To view example simulation data, download arbitrage.db from here and click 'Add A Simulation' on the dashboard.
To run the simulation yourself, use the following command.
python run.pyThis command will setup your local blockchain, contracts, accounts and agents. You can then access your results on your Dojo dashboard by connecting to a running simulation.
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.
class ArbitragePolicy(UniswapV3Policy):
"""Arbitrage trading policy for a UniswapV3Env with two pools.
:param agent: The agent which is using this policy.
"""
def __init__(self, agent: UniswapV3Agent) -> None: # noqa: D107
super().__init__(agent=agent)
self.block_last_trade: int = -1
self.min_block_dist: int = 20
self.min_signal: float = 1.901
self.tradeback_via_pool: Union[str, None] = NoneSignal 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.
def compute_signal(self, obs: UniswapV3Observation) -> Tuple[Decimal, Decimal]:
"""Compute strength or arbitrage signal."""
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 signalsTrade Execution
Afterwards, we return a UniswapV3Trade 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.
def predict(self, obs: UniswapV3Observation) -> list[BaseUniswapV3Action]:
"""Derive actions from observations."""
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 = UniswapV3Trade(
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: # type: ignore
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 # type: ignore
return [
UniswapV3Trade(
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.
Results
You can download the results to this example below.
We offer a dashboard desktop application for visualizing your simulation results. You can download the file for the desktop application here, or just open the results in our hosted dashboard.