Calculating price using Web3py and Uniswap pair smart contracts: How to deal with fixed point numbers?

Question

I'm trying to calculate the price of a token in Uniswap using web3.py and a Uniswap pair contract, but I must be doing something wrong with the math. Example: I would like to calculate the price of one ETH denominated in USDC. I first connect to the pair contract USDC/ETH (0xb4e16d0168e52d35cacd2c6185b44281ec28c9dc) to get access to the smart contracts functions via web3.py.

As explained here (https://uniswap.org/docs/v2/smart-contract-integration/building-an-oracle/) I'm storing the result of price1CumulativeLast() and the associated timestamp at two distinct points in time (one minute between each function call). I then use the formula

(price0CumulativeLATEST — price0CumulativeFIRST) / (timestampOfLATEST — timestampOfFIRST)

(see: https://medium.com/@epheph/using-uniswap-v2-oracle-with-storage-proofs-3530e699e1d3) to compute the price of token1 (ETH) denominated in token0 (USDC).

The Uniswap docs say that price1CumulativeLast() returns a Q112Q112 fixed point number which is why I think I cannot make sense of the numbers.

I've tried searching for Python functions to easily convert a fixed point Q112Q112 to float but haven't found a working solution so I guess I must getting something fundamental wrong with the math or units used in Ethereum smart contracts or Uniswap specifically.

def calculate_price():
'''
Call price1CumulativeLast() three times with one minute interval between each call.
'''
results = []
for x in range(3):
    try:
        temp_dict = {}
        start_ts = w3.eth.getBlock(w3.eth.block_number).timestamp
        token1_price_start = contract.functions.price1CumulativeLast().call()
        time.sleep(60*1)
        end_ts = w3.eth.getBlock(w3.eth.block_number).timestamp
        token1_price_end = contract.functions.price1CumulativeLast().call()
        temp_dict['start_ts'] = start_ts
        temp_dict['token1_price_start'] = token1_price_start
        temp_dict['end_ts'] = end_ts
        temp_dict['token1_price_end'] = token1_price_end
        results.append(temp_dict)
    except:
        continue
return results

This gives me:

    results = [{'start_ts': 1623002172,
  'token1_price_start': 183015811459414492033193017518027,
  'end_ts': 1623002242,
  'token1_price_end': 183016664977333417354464783721666},
 {'start_ts': 1623002242,
  'token1_price_start': 183016664977333417354464783721666,
  'end_ts': 1623002250,
  'token1_price_end': 183016664977333417354464783721666},
 {'start_ts': 1623002250,
  'token1_price_start': 183016664977333417354464783721666,
  'end_ts': 1623002355,
  'token1_price_end': 183018525945544514538790080485913}]

I now insert two timestamps and two prices into the formula to recalculate the price of token1 denominated in token0 over an interval of one minute:

   price = (results[0]['token1_price_end'] - results[0]['token1_price_start']) / (results[0]['end_ts'] - results[0]['start_ts'])
format(price, '.10f')

This returns the following string:

'12193113127504589866663936.0000000000'

At the time of writing this should be around 2800 (1 ETH = 2800 USDC) but I don't know how to get to that number from here. What am I doing wrong?

Answer 1

Here is an example code that calculates Uniswap pair contract price in few ways using web3-ethereum-defi library.

I don't think price1CumulativeLast needs to be involved.

The example code

• Gets the price entries using Uniswap v2 pair reserve0 and reserve1

• Price after the latest trade

• Time-weighted average price (TWAP)

"""Show live BNB/BUSD price from PancakeSwap pool.

- Show the latest price

- Show the TWAP price

Also

- Uses HTTP polling method

- Adjusts for minor chain reorgs / unstable chain tip

To run:

.. code-block:: python

    export BNB_CHAIN_JSON_RPC="https://bsc-dataseed.binance.org/"
    python scripts/live-price.py

"""
import datetime
import os
import time

from web3 import Web3, HTTPProvider
from web3.middleware import geth_poa_middleware

from eth_defi.price_oracle.oracle import PriceOracle, time_weighted_average_price
from eth_defi.uniswap_v2.oracle import update_live_price_feed
from eth_defi.uniswap_v2.pair import fetch_pair_details


def main():
    json_rpc_url = os.environ["BNB_CHAIN_JSON_RPC"]

    web3 = Web3(HTTPProvider(json_rpc_url))
    web3.middleware_onion.clear()
    web3.middleware_onion.inject(geth_poa_middleware, layer=0)

    # https://tradingstrategy.ai/trading-view/binance/pancakeswap-v2/bnb-busd
    pair_contract_address = "0x58F876857a02D6762E0101bb5C46A8c1ED44Dc16"

    reverse_token_order = False

    pair_details = fetch_pair_details(web3, pair_contract_address)

    print(f"Displaying live and TWAP price for {pair_details.token0.symbol} - {pair_details.token1.symbol}")

    price_ticker = f"{pair_details.token0.symbol}/{pair_details.token1.symbol}"

    oracle = PriceOracle(
        time_weighted_average_price,
        max_age=datetime.timedelta(minutes=15),   # Crash if we data gets more stale than 15 minutes
        min_duration=datetime.timedelta(minutes=1),
    )

    # How fast BNB Smart chain ticks
    block_time = 3.0

    initial_fetch_safety_margin = 1.2

    # To back fill the oracle buffer,
    # unitially fetch data for the latest time window blocks plus 20% safety margin
    initial_fetch_block_count = int(oracle.target_time_window / datetime.timedelta(seconds=block_time) * initial_fetch_safety_margin)

    print(f"Starting initial data fetch of {initial_fetch_block_count} blocks")
    update_live_price_feed(
        oracle,
        web3,
        pair_contract_address,
        reverse_token_order=reverse_token_order,
        lookback_block_count=initial_fetch_block_count)

    print(f"Starting live price feed, TWAP time window is set to {oracle.target_time_window}")
    while True:
        stats = update_live_price_feed(
            oracle,
            web3,
            pair_contract_address,
            reverse_token_order=reverse_token_order)

        last_price = oracle.get_newest().price
        twap = oracle.calculate_price()

        oldest = oracle.get_oldest()
        newest = oracle.get_newest()

        print(f"Block {oracle.last_refreshed_block_number:,} at {oracle.last_refreshed_at} current price:{last_price:.4f} {price_ticker} TWAP:{twap:.4f} {price_ticker}")
        print(f"    Oracle data updates: {stats}, trades in TWAP buffer:{len(oracle.buffer)}, oldest:{oldest.timestamp}, newest:{newest.timestamp} ")
        time.sleep(block_time)


if __name__ == "__main__":
    main()

For more information see here.