Let's assume a pool with X and Y assets (order is important). If the current price of X is lower than the target price, then some X assets must be bought from the pool. Otherwise if the price of X is higher than the target price, some Y assets must be bought from the pool.
The basic idea is to compute the amounts of X or Y to buy, with an algorithm similar to the swap algorithm in the Fig. 4, Uniswap v3 whitepaper.
Basic idea:
- Initialize a
delta
variable to zero.
- Iterate over all tick ranges until the target price is reached:
- If the target price is in the considered tick range, calculate the
number of tokens required to reach that price, add that to
delta
,
and break the loop.
- Otherwise, add all tokens in the tick range to the
delta
(find that by looking at its liquidity), and switch to the next tick range.
The main logic is symmetrical for buying X and buying Y, but implementation details are slightly different depending on which direction the price must move.
Let's assume that these variables are already initialized:
contract
- the pool's contract
sCurrentPrice
- sqrt of the
current price
sPriceTarget
- sqrt of the target price
liquidity
- the liquidity in the current tick range of the pool
tickLower
, tickUpper
- the min and max ticks of the current tick range
sPriceUpper
, sPriceUpper
- square roots of prices
corresponding to the min and max ticks of the current range
tickSpacing
- the tick spacing in the pool.
decimalsX
, decimalsY
- the number of decimals of the X and Y tokens, for printing the result
What follows is a proof-of-concept Python code.
Let's define some helper functions first:
# amount of x in range; sp - sqrt of current price, sb - sqrt of max price
def x_in_range(L, sp, sb):
return L * (sb - sp) / (sp * sb)
# amount of y in range; sp - sqrt of current price, sa - sqrt of min price
def y_in_range(L, sp, sa):
return L * (sp - sa)
def tick_to_price(tick):
return 1.0001 ** tick
The main code:
from web3 import Web3
from collections import namedtuple
Tick = namedtuple("Tick", "liquidityGross liquidityNet feeGrowthOutside0X128 feeGrowthOutside1X128 tickCumulativeOutside secondsPerLiquidityOutsideX128 secondsOutside initialized")
# how much of X or Y tokens we need to *buy* to get to the target price?
deltaTokens = 0
if sPriceTarget > sPriceCurrent:
# too few Y in the pool; we need to buy some X to increase amount of Y in pool
while sPriceTarget > sPriceCurrent:
if sPriceTarget > sPriceUpper:
# not in the current price range; use all X in the range
x = x_in_range(liquidity, sPriceCurrent, sPriceUpper)
deltaTokens += x
# query the blockchain for liquidity in the next tick range
nextTickRange = Tick(*contract.functions.ticks(tickUpper).call())
liquidity += nextTickRange.liquidityNet
# adjust the price and the range limits
sPriceCurrent = sPriceUpper
tickLower = tickUpper
tickUpper += tickSpacing
sPriceLower = sPriceUpper
sPriceUpper = tick_to_price(tickUpper // 2)
else:
# in the current price range
x = x_in_range(liquidity, sPriceCurrent, sPriceTarget)
deltaTokens += x
sPriceCurrent = sPriceTarget
print("need to buy {:.10f} X tokens".format(deltaTokens / 10 ** decimalsX))
elif sPriceTarget < sPriceCurrent:
# too much Y in the pool; we need to buy some Y to decrease amount of Y in pool
currentTickRange = None
while sPriceTarget < sPriceCurrent:
if sPriceTarget < sPriceLower:
# not in the current price range; use all Y in the range
y = y_in_range(liquidity, sPriceCurrent, sPriceLower)
deltaTokens += y
if currentTickRange is None:
# query the blockchain for liquidityNet in the *current* tick range
currentTickRange = Tick(*contract.functions.ticks(tickLower).call())
liquidity -= currentTickRange.liquidityNet
# adjust the price and the range limits
sPriceCurrent = sPriceLower
tickUpper = tickLower
tickLower -= tickSpacing
sPriceUpper = sPriceLower
sPriceLower = tick_to_price(tickLower // 2)
# query the blockchain for liquidityNet in new current tick range
currentTickRange = Tick(*contract.functions.ticks(tickLower).call())
else:
# in the current price range
y = y_in_range(liquidity, sPriceCurrent, sPriceTarget)
deltaTokens += y
sPriceCurrent = sPriceTarget
print("need to buy {:.10f} Y tokens".format(deltaTokens / 10 ** decimalsY))
The main result is stored in the variable deltaTokens
, as value with then can be used in the exactOutputSingle
function call.
A caveat: the approach loops through all tick ranges until the target price is reached. This is fine in most situations, but this approach could be quite inefficient. Looking at the next initialized tick would be more efficient if the price difference is big and most of the potentially initialized ticks are not initialized. Another problem is that if there isn't enough liquidity in the pool, the loop would never stop, as there are no other stopping conditions at the moment.
Edit: for JavaScript and TypeScript apps, I suggest using the SwapMath
library from the Uniswap v3 SDK as the basis for your code.