uniswap_v3.price

Documentation for eth_defi.uniswap_v3.price Python module.

Uniswap v3 price calculations.

Helpers to calculate

Example:

import os
from decimal import Decimal

from eth_defi.provider.multi_provider import create_multi_provider_web3
from eth_defi.uniswap_v3.constants import UNISWAP_V3_DEPLOYMENTS
from eth_defi.uniswap_v3.deployment import fetch_deployment
from eth_defi.uniswap_v3.pool import fetch_pool_details
from eth_defi.uniswap_v3.price import get_onchain_price, estimate_buy_received_amount
from eth_defi.uniswap_v3.tvl import fetch_uniswap_v3_pool_tvl


def main():
    # You can pass your own endpoint in an environment variable
    json_rpc_url = os.environ.get("JSON_RPC_POLYGON", "https://polygon-rpc.com")

    # Search pair contract addresses using Trading Strategy search: https://tradingstrategy.ai/search
    # This one is:
    # https://tradingstrategy.ai/trading-view/polygon/uniswap-v3/eth-usdc-fee-5
    pool_address = os.environ.get("PAIR_ADDRESS", "0x45dda9cb7c25131df268515131f647d726f50608")

    # Create web3 connection instance
    web3 = create_multi_provider_web3(json_rpc_url)

    contract_details = UNISWAP_V3_DEPLOYMENTS["polygon"]
    uniswap = fetch_deployment(
        web3,
        factory_address=contract_details["factory"],
        router_address=contract_details["router"],
        position_manager_address=contract_details["position_manager"],
        quoter_address=contract_details["quoter"],
    )

    # Get Pool contract ABI file, prepackaged in eth_defi Python package
    # and convert it to a wrapped Python object
    pool = fetch_pool_details(web3, pool_address)

    inverse = True

    # Manually resolve token order from random Uniswap v3 order
    if inverse:
        base_token = pool.token1
        quote_token = pool.token0
    else:
        base_token = pool.token0
        quote_token = pool.token1

    # Print out pool details
    # token0 and token1 will be always in a random order
    # and may inverse the price
    print("-" * 80)
    print("Uniswap pool details")
    print("Chain", web3.eth.chain_id)
    print("Pool", pool_address)
    print("Token0", pool.token0.symbol)
    print("Token1", pool.token1.symbol)
    print("Base token", base_token.symbol)
    print("Quote token", quote_token.symbol)
    print("Fee (BPS)", pool.get_fee_bps())
    print("-" * 80)

    inverse = True  # Is price inverted for output

    # Record the block number close to our timestamp
    block_num = web3.eth.get_block_number()

    # Use get_onchain_price() to get a human readable price
    # in Python Decimal
    mid_price = get_onchain_price(
        web3,
        pool.address,
    )

    if inverse:
        mid_price = 1 / mid_price

    target_pair_fee_bps = 5

    # Attempt to buy ETH wit $1,000,000.50
    swap_amount = Decimal("1_000_000.50")
    swap_amount_raw = quote_token.convert_to_raw(swap_amount)

    received_amount_raw = estimate_buy_received_amount(
        uniswap=uniswap,
        base_token_address=base_token.address,
        quote_token_address=quote_token.address,
        quantity=swap_amount_raw,
        target_pair_fee=target_pair_fee_bps * 100,  # Uniswap v3 units
        block_identifier=block_num,
    )

    received_amount = base_token.convert_to_decimals(received_amount_raw)

    quoted_price = received_amount / swap_amount

    if inverse:
        quoted_price = 1 / quoted_price

    price_impact = (quoted_price - mid_price) / mid_price

    tvl_quote = fetch_uniswap_v3_pool_tvl(
        pool,
        quote_token,
        block_identifier=block_num,
    )

    tvl_base = fetch_uniswap_v3_pool_tvl(
        pool,
        base_token,
        block_identifier=block_num,
    )

    print(f"Block: {block_num:,}")
    print(f"Swap size: {swap_amount:,.2f} {quote_token.symbol}")
    print(f"Pool base token TVL: {tvl_base:,.2f} {base_token.symbol}")
    print(f"Pool quote token TVL: {tvl_quote:,.2f} {quote_token.symbol}")
    print(f"Mid price {base_token.symbol} / {quote_token.symbol}: {mid_price:,.2f}")
    print(f"Quoted amount to received: {received_amount:,.2f} {base_token.symbol}")
    print(f"Quoted price (no execution slippage): {quoted_price:,.2f} {quote_token.symbol}")
    print(f"Price impact: {price_impact * 100:.2f}%")


if __name__ == "__main__":
    main()

See Uniswap v3 historical price estimation tutorial for more information.

Functions

estimate_buy_received_amount(uniswap, ...[, ...])

Estimate how much we receive for buying with a certain quote token amount.

estimate_sell_received_amount(uniswap, ...)

Estimate how much we receive for selling a certain base token amount.

get_onchain_price(web3, pool_contract_address, *)

Get the current price of a Uniswap v3 pool.

Classes

UniswapV3PriceHelper

Internal helper class for price calculations.

Exceptions

QuotingFailed

QuoterV2 pukes ouk revert.
exception QuotingFailed

Bases: Exception

QuoterV2 pukes ouk revert.

__init__(*args, **kwargs)
__new__(**kwargs)
add_note()

Exception.add_note(note) – add a note to the exception

with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

class UniswapV3PriceHelper

Bases: object

Internal helper class for price calculations.

__init__(uniswap_v3)
Parameters

uniswap_v3 (eth_defi.uniswap_v3.deployment.UniswapV3Deployment) –

get_amount_out(amount_in, path, fees, *, slippage=0, block_identifier=None)

Get how much token we are going to receive.

Example:

# Estimate how much DAI we will receive for 1000 WETH
# using the route of 2 pools: WETH/USDC 0.3% and USDC/DAI 1%
# with slippage tolerance is 0.5%
price_helper = UniswapV3PriceHelper(uniswap_v3_deployment)
amount_out = price_helper.get_amount_out(
    1000,
    [
        weth.address,
        usdc.address,
        dai.address,
    ],
    [
        3000,
        10000,
    ],
    slippage=50,
)
Parameters

  • amount_in (int) – Amount of input asset.

  • path (list[eth_typing.evm.HexAddress]) – List of token addresses how to route the trade

  • fees (list[int]) – List of trading fees of the pools in the route

  • slippage (float) – Slippage express in bps

  • block_identifier (int | None) – A specific block to estimate price

Return type

int

get_amount_in(amount_out, path, fees, *, slippage=0, block_identifier=None)

Get how much token we are going to spend.

Parameters

  • amount_in – Amount of output asset.

  • path (list[eth_typing.evm.HexAddress]) – List of token addresses how to route the trade

  • fees (list[int]) – List of trading fees of the pools in the route

  • slippage (float) – Slippage express in bps

  • block_identifier (int | None) – A specific block to estimate price

  • amount_out (int) –

Return type

int

estimate_buy_received_amount(uniswap, base_token_address, quote_token_address, quantity, target_pair_fee, *, slippage=0, intermediate_token_address=None, intermediate_pair_fee=None, block_identifier=None, verbose=False)

Estimate how much we receive for buying with a certain quote token amount.

Example:

# Estimate the price of buying 1650 USDC worth of ETH
eth_received = estimate_buy_received_amount(
    uniswap_v3,
    weth.address,
    usdc.address,
    1650 * 10**18,  # Must be raw token units
    500,  # 100 Uniswap v3 fee units = 1 BPS, this is 5 BPS
)

assert eth_received / (10**18) == pytest.approx(0.9667409780905836)

# Calculate price of ETH as $ for our purchase
price = (1650 * 10**18) / eth_received
assert price == pytest.approx(Decimal(1706.7653460381143))
Parameters
Returns

Expected base token amount to receive

Raises

TokenDetailError – If we have an issue with ERC-20 contracts

Return type

int | tuple[int, int]

estimate_sell_received_amount(uniswap, base_token_address, quote_token_address, quantity, target_pair_fee, *, slippage=0, intermediate_token_address=None, intermediate_pair_fee=None, block_identifier=None, verbose=False)

Estimate how much we receive for selling a certain base token amount.

See example in eth_defi.uniswap_v3.price.

Parameters
Returns

Expected quote token amount to receive

Raises

TokenDetailError – If we have an issue with ERC-20 contracts

Return type

int | tuple[int, int]

get_onchain_price(web3, pool_contract_address, *, block_identifier=None, reverse_token_order=False)

Get the current price of a Uniswap v3 pool.

Reads Uniswap v3 “slot 0” price.

  • This is the current price according to Uniswap team explanation, which we assume is the mid-price

  • See mid price

To read the latest ETH-USDC price on Polygon:

import os
from web3 import Web3, HTTPProvider

from eth_defi.uniswap_v3.price import get_onchain_price

json_rpc_url = os.environ["JSON_RPC_POLYGON"]
web3 = Web3(HTTPProvider(json_rpc_url))

# ETH-USDC 5 BPS pool address
# https://tradingstrategy.ai/trading-view/polygon/uniswap-v3/eth-usdc-fee-5
pool_address = "0x45dda9cb7c25131df268515131f647d726f50608"

price = get_onchain_price(web3, pool_address, reverse_token_order=True)
print(f"ETH price is {price:.2f} USD")
Parameters

  • web3 (web3.main.Web3) – Web3 instance

  • pool_contract_address (str) – Contract address of the Uniswap v3 pool

  • block_identifier (int | None) –

    A specific block to query price.

    Block number or block hash.

  • reverse_token_order (bool) –

    For switching the pair ticker around to make it human readable.

    • If set, assume quote token is token0, and the human price is 1/price

    • If not set assumes base token is token0

Returns

Current price in human-readable Decimal format.