Skip to main content

transaction_builder.protocols.uniswap_v3.uniswap_v3_sdk

glob​

json​

math​

os​

time​

Dict​

List​

Optional​

Tuple​

get_address_by_chain_and_network​

Chain​

Network​

ISDK​

Config​

get_blocknative_tip​

get_web3_by_network_and_chain​

Contract​

DEBUG​

UniswapV3SDK Objects​

class UniswapV3SDK(ISDK)

A Uniswap SDK containing the functions that will be called on the uniswap protocol. NOTE: Common error codes from uniswap contract: https://docs.uniswap.org/contracts/v3/reference/error-codes

UNISWAP_TICK_SPACING​

MAX_UINT_128​

MAX_UINT_256​

UNISWAP_MIN_TICK​

UNISWAP_MAX_TICK​

Q96​

Q128​

DEADLINE_100_DAYS​

__init__​

def __init__(network: Network, chain: Chain)

factory_contract​

@property
def factory_contract() -> Contract

router_contract​

@property
def router_contract() -> Contract

position_manager_contract​

@property
def position_manager_contract() -> Contract

quoter_contract​

@property
def quoter_contract() -> Contract

transfer_eth​

def transfer_eth(from_address: str,
                 to_address: str,
                 amount: int,
                 nonce_counter: int = 0,
                 set_gas_override: Optional[int] = None,
                 block_identifier: Optional[int] = None) -> str

Transfer ETH from one address to another.

Parameters​

from_address: str the address to transfer from to_address: str the address to transfer to amount: int the amount of ETH to transfer

Returns​

tx_hash: str the transaction hash

wrap​

def wrap(from_address: str,
         amount: int,
         set_gas_override: Optional[int] = None,
         block_identifier: Optional[int] = None) -> str

Wrap ETH to WETH

Parameters​

amount: int the amount of ETH to wrap

Returns​

tx_hash: str the transaction hash

unwrap​

def unwrap(token_address: str,
           from_address: str,
           amount: int,
           set_gas_override: Optional[int] = None,
           block_identifier: Optional[int] = None) -> str

Unwrap WETH to ETH or other token

Parameters​

amount: int the amount of wrap token to unwrap

Returns​

tx_hash: str the transaction hash

approve​

def approve(token_address: str,
            spender_address: str,
            from_address: str,
            amount: Optional[int] = None,
            set_gas_override: Optional[int] = None,
            block_identifier: Optional[int] = None) -> str

Approve a spender to spend a certain amount of tokens on behalf of the user.

Parameters​

token_contract: Contract the address of the token to approve spender_contract: Contract the address of the spender from_address: str the address of the user amount: int the amount of tokens to approve. If not provided, defaults to the maximum uint256 value

Returns​

tx_hash: str the transaction hash

swap_in​

def swap_in(tokenIn: str,
            tokenOut: str,
            fee: int,
            recipient: str,
            amountIn: int,
            amountOutMinimum: int,
            sqrtPriceLimitX96: int,
            from_address: str,
            transfer_eth_in: bool,
            slippage: Optional[float] = None,
            set_gas_override: Optional[int] = None,
            block_identifier: Optional[int] = None)

Builds an unsigned transaction for a single-path swap on Uniswap V3 using the exactInputSingle function.

Arguments:

  • tokenIn str - The address of the input token.
  • tokenOut str - The address of the output token.
  • fee int - The fee tier of the pool, e.g., 3000 for 0.3%.
  • recipient str - The address to receive the output tokens.
  • amountIn int - The amount of input tokens to swap, in the smallest indivisible unit.
  • amountOutMinimum int - The minimum amount of output tokens to receive, in the smallest indivisible unit. this field is not used right now.
  • sqrtPriceLimitX96 int - The maximum or minimum sqrt price limit, depending on the swap direction.
  • from_address str - The address to send the transaction from.
  • transfer_eth_in bool - Whether to transfer ETH as the input token (if tokenIn is WETH).

Returns:

  • dict - The unsigned transaction dictionary.

Notes:

  • The transaction fees are calculated using the tokenOut0 method.
  • The gas limit is estimated and increased by the tokenOut1 factor.
  • If transfer_eth_in is True and tokenIn is WETH, the transaction value is set to amountIn.
  • The transaction type is set to 2 (EIP-1559).
  • The nonce is set to the current transaction count for the from_address.

quoteExactInputSingle​

def quoteExactInputSingle(token0_address: str,
                          token1_address: str,
                          fee: int,
                          amount: int,
                          sqrtPriceLimitX96: int = 0,
                          block_identifier: Optional[int] = None) -> int

A function to call the quoteExactInputSingle function from the uniswap v3 uni_quoter V2 smart contract. https://docs.uniswap.org/contracts/v3/reference/periphery/lens/Quoter

Gives the price one would obtain (without slippage). Includes price impact and fees in calculation

Parameters​

token0_address: str the address of the input token token1_address: str the address of the output token fee: int the pool fee where the swap will occur amount: int the desired input amount

Returns​

quote: int the price in units of the second token. I.e, "quote / 10**token1_decimals" is in nominal units

quoteExactInputSingleV1​

def quoteExactInputSingleV1(token0_address: str,
                            token1_address: str,
                            fee: int,
                            amount: int,
                            sqrtPriceLimitX96: int = 0,
                            block_identifier: Optional[int] = None) -> int

" Quoter function on the uniswap V1 Quoter contract

quoteExactOutputSingle​

def quoteExactOutputSingle(token0_address: str,
                           token1_address: str,
                           fee: int,
                           amount: int,
                           sqrtPriceLimitX96: int = 0,
                           block_identifier: Optional[int] = None) -> int

A function to call the quoteExactOutputSingle function from the uniswap v3 uni_quoter smart contract. https://docs.uniswap.org/contracts/v3/reference/periphery/lens/Quoter

Given the amount you want to get out, produces a quote for the amount in for a swap over a single pool

Parameters​

token0_address: str the address of the input token token1_address: str the address of the output token fee: int the pool fee where the swap will occur amount: int the desired output amount

Returns​

quote: int The amount required as the input for the swap in order to receive amountOut

quoteExactOutputSingleV1​

def quoteExactOutputSingleV1(token0_address: str,
                             token1_address: str,
                             fee: int,
                             amount: int,
                             sqrtPriceLimitX96: int = 0,
                             block_identifier: Optional[int] = None) -> int

open_lp_position​

def open_lp_position(token0_address: str,
                     token1_address: str,
                     fee: int,
                     lower_bound_price: float,
                     upper_bound_price: float,
                     token0_desired: int,
                     token1_desired: int,
                     agent_address: str,
                     amount0_min: int = 0,
                     amount1_min: int = 0,
                     slippage: Optional[float] = None,
                     set_gas_override: Optional[int] = None,
                     block_identifier: Optional[int] = None) -> str

Open an lp position on uniswap v3.

Ordering of the pool matters. E.g., if token0_address=WETH, but the pool address token0 = DAI, the amounts and tick ranges are inverted within this function.

NOTE: DOES NOT WORK WITH ETH. Do not send ETH in.​

https://docs.uniswap.org/contracts/v3/guides/providing-liquidity/mint-a-position

Arguments:

token0_address (str): token 0 address token1_address (str): token 1 address fee (int): pool fee (e.g. 3000 = 0.3%) lower_bound_price (float): the lower bound of the price to add liquidity in. Note price is in nominal units, e.g., 1WETH = 1500.001 USDT upper_bound_price (float): the upper bound of the price to add liquidity in Note price is in nominal units, e.g., 1WETH = 1500.001 USDT token0_desired (int): the amount of token0 to add into the position in units of token0 Eg., 1ETH is 1018 WEI, so token0 = 1018 for 1 ETH. token1_desired (int): the amount of token1 to add into the position agent_address (str): address of the agent

  • amount0_min int, optional - description. Defaults to 0.
  • amount1_min int, optional - description. Defaults to 0.
  • slippage - A number between 0 and 1 which determines amountmin = (1-slippage)*desired

Returns:

  • str - transaction hash of the transaction

close_lp_position​

def close_lp_position(agent_address: str,
                      position_id: int,
                      amount0_min: Optional[int] = None,
                      amount1_min: Optional[int] = None,
                      slippage: Optional[float] = None,
                      pool_address: Optional[str] = None,
                      set_gas_override: Optional[int] = None,
                      block_identifier: Optional[int] = None) -> List[str]

This function removes the liquidity from position_id, collects the fees and tokens from the removed liquidity position_id, and burns position_id.

Arguments:

agent_address (str): The address of the agent position position_id (int): the integer value of the liquidity position amount0_min (int, optional): amount0 of token0 to get out for slippage protection. Defaults to 0. amount1_min (int, optional): amount1 of token1 to get our for slippage protection. Defaults to 0.

  • slippage - (float, optional): A number between 0 and 1. Defaults to None. (1-slippage)*desired will override min amounts. pool_address (str, optional): The address of the pool, needed only for Slippage calculation. Defaults to None. If slippage isn't provided, this is not needed.

Returns:

close_lp_position_multicall​

def close_lp_position_multicall(agent_address: str,
                                position_id: int,
                                amount0_min: Optional[int] = None,
                                amount1_min: Optional[int] = None,
                                slippage: Optional[float] = None,
                                pool_address: Optional[str] = None,
                                set_gas_override: Optional[int] = None,
                                block_identifier: Optional[int] = None) -> str

remove_lp_liquidity​

def remove_lp_liquidity(agent_address: str,
                        position_id: int,
                        amount0_min: Optional[int] = None,
                        amount1_min: Optional[int] = None,
                        slippage: Optional[float] = None,
                        pool_address: Optional[str] = None,
                        set_gas_override: Optional[int] = None,
                        block_identifier: Optional[int] = None) -> str

Removes ALL liquidity from the position_id of agent_address. Partial liquidity removal is NOT implemented.

NOTE: Does not move tokens to agents wallet. To do that, call collect(). This function only decreases liquidity position on uniswapv3 internal bookkeeping.

Arguments:

agent_address (str): The address of the agent of the liquidity position position_id (int): The id of the liquidity position to be removed amount0_min (int, optional): Defaults to 0. amount1_min (int, optional): Defaults to 0. liquidity (int): The amount of liquidity to be removed.

Returns:

str: The transaction hash of the position

collect_lp​

def collect_lp(agent_address: str,
               position_id: int,
               set_gas_override: Optional[int] = None,
               block_identifier: Optional[int] = None) -> str

After having accumulated fees, or having decreased a liquidity position, the tokens are still in uniswapv3 position and need to be moved to agents wallet. This function moves/collects the maximum amount of both fees and liquidity to the agents wallet from uniswap contract.

Arguments:

  • agent_address str - agents address
  • position_id int - id of the agents position which the own.

Returns:

  • str - transaction hash of the transaction.

burn_lp_position​

def burn_lp_position(agent_address: str,
                     position_id: int,
                     set_gas_override: Optional[int] = None,
                     block_identifier: Optional[int] = None) -> str

Burn the liquidity position of agent_address at position_id NOTE: can't burn until all liquidity is removed and fees collected

Arguments:

agent_address (str): address of agent of the position position_id (int): id of the position of the agent_address

Returns:

str: transaction hash of the burn transaction

get_unclaimed_fees​

def get_unclaimed_fees(
        pool_address: str,
        position_id: int,
        block_identifier: Optional[int] = None) -> Tuple[int, int]

For a uniswapv3 liquidity position, find the amount of uncollected fees that exists in a certain position. Returns the unclaimed fees in units of base token (e.g., WEI for WETH).

Arguments:

  • pool_address str - address of the uniswap v3 pool
  • position_id int - position of the agents liquidity.

Returns:

Tuple[int, int]:

  • uncollected_fees0 - the amount of uncollected fees the agent has in the position in token0 units, e.g., WEI units for WETH
  • uncollected_fees1 - the amount of uncollected fees the agent has in the position in token1 units e.g., WEI units for WETH NOTE: https://ethereum.stackexchange.com/questions/101955/trying-to-make-sense-of-uniswap-v3-fees-feegrowthinside0lastx128-feegrowthglob https://github.com/someben/pyuv3/blob/main/pyuv3/flowint.py Check out the relevant formulas below which are from Uniswap Whitepaper Section 6.3 and 6.4

    𝑓𝑟 =𝑓𝑔−𝑓𝑏(𝑖𝑙)−𝑓𝑎(𝑖𝑢)

    𝑓𝑢 =𝑙·(𝑓𝑟(𝑡1)−𝑓𝑟(𝑡0))

get_lp_token_amounts​

def get_lp_token_amounts(
        pool_address: str,
        position_id: int,
        block_identifier: Optional[int] = None) -> Tuple[int, int]

Find the number of tokens inside a single liquidity position for an agent.

Arguments:

pool_address (str): The address of the specific pool of interest position_id (int): the id of the lp position of the agent

Returns:

Tuple[int, int]: amount0 - the number of tokens0 in the position amount1 - the number of tokens1 in the position

  • NOTE - amounts are in base units, e.g., WEI for WETH NOTE: The logic for getting the token amounts inside a specific tick bin is See 3.3.3 of LIQUIDITY MATH IN UNISWAP V3 by Atis Elsts

calculate_token0_amount​

def calculate_token0_amount(liquidity, sp, sa, sb)

Calculate the number of tokens0 inside an individual tick bin

Arguments:

  • liquidity type - liquidity in position
  • sp type - sqrt of current price
  • sa type - sqrt of lower tick price
  • sb type - sqrt of upper tick price

Returns:

number of tokens0 in tick bin

calculate_token1_amount​

def calculate_token1_amount(liquidity, sp, sa, sb)

Calculate the number of tokens1 inside an individual tick bin

Arguments:

  • liquidity type - liquidity in position
  • sp type - sqrt of current price
  • sa type - sqrt of lower tick price
  • sb type - sqrt of upper tick price

Returns:

number of tokens1 in tick bin

calculate_liquidity0​

def calculate_liquidity0(amount0, sa, sb)

calculate_liquidity1​

def calculate_liquidity1(amount1, sa, sb)

calculate_liquidity​

def calculate_liquidity(amount0, amount1, sp, sa, sb)

https://github.com/uniyj/uni-v3-peri/blob/main/atiselsts-uniswap-v3-liquidity-math/uni-v3-liquidity-math.ipynb

get_position_info​

def get_position_info(position_id: int,
                      block_identifier: Optional[int] = None) -> Tuple

get_positions​

def get_positions(agent_address: str,
                  block_identifier: Optional[int] = None) -> List[int]

get indices of agent's LP positions (i.e. position IDs)

get_pool_spot_rate​

def get_pool_spot_rate(pool_address: str,
                       inverted: bool = False,
                       block_identifier=None) -> float

get_pool_current_tick​

def get_pool_current_tick(pool_address: str) -> int

get_pool​

def get_pool(token0_address: str, token1_address: str, fee: int)

is_checksum_address​

def is_checksum_address(address: str) -> bool

to_checksum_address​

def to_checksum_address(address: str) -> str

get_pool_liquidity​

def get_pool_liquidity(pool_address: str)

get_transaction_fees​

def get_transaction_fees(block_identifier=None) -> Dict[str, int]

get_token_contract​

def get_token_contract(token_address: str) -> Contract

get_pool_contract​

def get_pool_contract(pool_address: str) -> Contract

pool_price_from_current_tick​

def pool_price_from_current_tick(token0_address: str,
                                 token1_address: str,
                                 pool_fee: int,
                                 block_identifier=None) -> Tuple[float, int]

Return the price of the current tick, in the order of the token input addresses.

Arguments:

  • token0_address str - description
  • token1_address str - description
  • pool_fee int - description

Returns:

  • float - the price of the pool in the order of the input tokens.
  • int - the current tick of the pool.

unsigned_modulo​

def unsigned_modulo(python_int: int, num_bits: int) -> int

Fixed point arithmetic on blockchain is not implemented in python. We need to modulo a python integer into an unsigned int of num_bits.

NOTE: implemented via bitwise.

Arguments:

  • python_int int - any python int.
  • num_bits int - The number of bits that the unsigned int needs to be.

Returns:

  • int - An integer in the range [0,..., 2^num_bits -1] e.g,. an unsigned int.

get_min_tick​

def get_min_tick(fee: int) -> int

get_max_tick​

def get_max_tick(fee: int) -> int

default_tick_range​

def default_tick_range(fee: int) -> Tuple[int, int]

nearest_tick​

def nearest_tick(tick: int, fee: int) -> int

Return the nearest tick on the uniswap v3 pool.

price_to_tick​

def price_to_tick(price: float, token0_decimals: int,
                  token1_decimals: int) -> int

Converts a price in nominal units to a tick value on uniswap v3

Arguments:

price (float): The price of token0 to token1 in nominal units. E.g., 1eth = 1500.01 usdt

  • token0_decimals int - decimals of token0
  • token1_decimals int - decimals of token1

Returns:

int: the tick value for the inputted price

price_to_sqrtp​

def price_to_sqrtp(price)

sqrtp_to_price​

def sqrtp_to_price(sqrtp)

tick_to_unadjusted_sqrtp​

def tick_to_unadjusted_sqrtp(tick: int) -> float

Convert a tick to an unadjusted sqrt price. Unadjusted means no token decimals multiplication has been performed.

Arguments:

  • tick int - A tick bin.

Returns:

unadjusted_price: The price that has not been adjusted yet. See 3.3.2 of LIQUIDITY MATH IN UNISWAP V3 by Atis Elsts

  • NOTE - To convert to an actual price, need to multiply by token0.decimals - token1.decimals as in tick_to_price() function

tick_to_price​

def tick_to_price(tick: int, token0_decimals: int,
                  token1_decimals: int) -> float

Determine the price given a tick value.

Arguments:

  • tick int - the tick of the pool from which to calculate price token0_decimals (int): token1_decimals (int):

tick_to_sqrt_price​

def tick_to_sqrt_price(tick: int) -> float

tick_to_sqrt_price_x_96​

def tick_to_sqrt_price_x_96(tick)

sqrt_price_x_96_to_tick​

def sqrt_price_x_96_to_tick(sqrt_price_x_96)

price_x96_to_price​

def price_x96_to_price(sqrt_price_x_96, decimals_diff=12)

trunc_int​

def trunc_int(num: int) -> int

get_next_ticks​

def get_next_ticks(pool_address: str,
                   fee: int,
                   buffer: int,
                   side_lower: bool,
                   tick: int = None)

This function calculates from either the current tick or a given tick, the next valid ticks based on the buffer.

Arguments:

  • pool_address (str): The address of the pool.
  • fee (int): The fee of the pool.
  • buffer (int): The number of bins or tick spacing to "skip".
  • side_lower (bool): The side of the buffer relative to the tick.
  • tick (int): None will default to the current tick from on-chain fetching.

lp_fees_percent_of_current_tokens​

def lp_fees_percent_of_current_tokens(
        pool_address: str,
        position_id: int,
        block_identifier: Optional[int] = None) -> Tuple[float, float]

calculate_impermanent_loss​

def calculate_impermanent_loss(price: float,
                               strike_price: float,
                               lower_limit: float,
                               upper_limit: float,
                               dex: str = "uniswap_v3") -> float

Calculate the impermanent loss for a single Uniswap V3 position.

NOTE: Impermanent loss is defined to be a negative number here.

Arguments:

  • price float - price (current)
  • strike_price float - "strike price" (i.e. price the position was opened at)
  • lower_limit float - lower bound of Uniswap V3 position range
  • upper_limit float - upper bound of Uniswap V3 position range
  • dex str - DEX in which to calculate IL (e.g. 'uniswap_v2', default == 'uniswap_v3')

Returns:

  • float - impermanent loss as a fraction. To get a percentage, multiply by 100

calculate_open_position_amount​

def calculate_open_position_amount(
    token0_amount: int,
    token1_amount: int,
    lower_price: float,
    upper_price: float,
    pool_address: str,
    slippage_spot_rate: float,
    adjust_amounts: bool = True,
    adjust_threshold: float = 0.001,
    block_identifier: Optional[int] = None
) -> Tuple[Tuple[int, int], Tuple[float, float], float]

Calculate the amount of tokens to be deposited into a Uniswap V3 position. Input a range of spot price that could be realized in the future, and output a range of liquidity amounts that could be deposited into the position.

calculate_open_position_amount_with_spot_rate​

def calculate_open_position_amount_with_spot_rate(
    token0_amount: int,
    token1_amount: int,
    lower_price: float,
    upper_price: float,
    pool_address: str,
    spot_rate: float,
    adjust_amounts: bool = True,
    adjust_threshold: float = 0.001
) -> Tuple[Tuple[int, int], Tuple[float, float]]