uniswap_viewer.viewer_sync

  1"""
  2viewer_sync.py - Synchronous interface for accessing Uniswap V3 pool data.
  3
  4This module provides a synchronous viewer for querying Uniswap V3 pools using
  5web3.py. It allows retrieving token prices, tick data, and computing 
  6tick-related values such as tick index and tick slots.
  7
  8Main components:
  9- `Viewer`: a class for interacting with a Uniswap V3 pool.
 10- `stream_new_blocks`: generator function to yield new block numbers as they 
 11                       appear.
 12- `get_decimals`: helper function to fetch token decimals from an ERC-20 
 13                  contract.
 14"""
 15
 16import math
 17import time
 18import weakref
 19from typing import Optional, Dict, Any
 20from collections.abc import Generator
 21
 22from web3 import Web3
 23
 24from .utils import get_abi, TICKS_KEYS, prepare_token_address
 25
 26
 27class Viewer:
 28    """
 29    A synchronous Uniswap V3 pool viewer.
 30
 31    This class enables price and tick queries for a specific Uniswap V3 pool
 32    defined by a token pair and a fee tier.
 33    """
 34
 35    def __init__(self, w3: Web3, token0: str, token1: str, fee: int = 3000):
 36        """
 37        Initializes the Viewer instance.
 38
 39        Args:
 40            w3 (Web3): Web3 instance.
 41            token0 (str): Address of the first token (must be > token1).
 42            token1 (str): Address of the second token.
 43            fee (int, optional): Pool fee in basis points. Allowed values are 
 44                                 100 (0.01%), 500 (0.05%), 3000 (0.3%), 10000 
 45                                 (1%). Defaults to 3000 (0.3%).
 46
 47        Raises:
 48            AssertionError: If fee is invalid or token0 is not > token1 in bits.
 49        """
 50        # Prepare arguments
 51        token0 = prepare_token_address(token0)
 52        token1 = prepare_token_address(token1)
 53
 54        # Check arguments
 55        assert fee in (100, 500, 3000, 10000), \
 56            "Allowed fee must be in (100, 500, 3000, 10000) " \
 57            "that corresponds to 0.01%, 0.05%, 0.3%, 1%."
 58
 59        assert int(token0, 16) > int(token1, 16), \
 60            "token0 must go after token1 in bit representation."
 61
 62        # Attributes
 63        self._w3 = weakref.ref(w3)
 64        self._token0 = self._w3().to_checksum_address(token0)
 65        self._token1 = self._w3().to_checksum_address(token1)
 66        self._fee = fee
 67        self._decimals0 = None
 68        self._decimals1 = None
 69        self._pool_address = None
 70        self._pool_contract = None
 71
 72    def init(self):
 73        """
 74        Initializes the pool contract by resolving its address from the factory.
 75
 76        Must be called before using methods like `get_price` or `get_tick_data`.
 77
 78        Raises:
 79            AssertionError: If no pool exists for the provided tokens and fee.
 80        """
 81        self._decimals0 = get_decimals(self._w3(), self._token0)
 82        self._decimals1 = get_decimals(self._w3(), self._token1)
 83        self._pool_address = self._get_pool_address()
 84        self._pool_contract = self._w3().eth.contract(
 85            address=self._pool_address, abi=get_abi('pool-abi')
 86        )
 87
 88    def get_price(self, block_num: Optional[int] = None) -> float:
 89        """
 90        Returns the current pool price as a floating-point value.
 91
 92        The result is adjusted for token decimals.
 93
 94        Args:
 95            block_num (int, optional): Specific block number. If None, uses 
 96                                       latest.
 97
 98        Returns:
 99            float: The spot price of token0 denominated in token1.
100        """
101        self._check_initialized()
102        slot0 = self._pool_contract.functions.slot0().call(
103            block_identifier=block_num
104        )
105        sqrtPriceX96 = slot0[0]
106        price = (sqrtPriceX96 / (2**96)) ** 2
107        price *= 10 ** (self._decimals1 - self._decimals0)
108        return price
109
110    def get_tick_data(self, tick: int, 
111                      block_num: Optional[int] = None) -> Dict[str, Any]:
112        """
113        Fetches tick data for a given tick index from the pool.
114
115        Args:
116            tick (int): The tick index.
117            block_num (int, optional): Specific block number. If None, uses 
118                                       latest.
119
120        Returns:
121            dict: A dictionary of tick data fields, keyed by TICKS_KEYS.
122        """
123        self._check_initialized()
124        self._check_available_tick(tick)
125        tick_data = self._pool_contract.functions.ticks(tick).call(
126            block_identifier=block_num
127        )
128        return dict(zip(TICKS_KEYS, tick_data))
129
130    def calc_tick(self, price: float, normalize: bool = False) -> int:
131        """
132        Calculates the tick index corresponding to a given price.
133
134        Args:
135            price (float): Price of token0 in token1 units.
136            normalize (bool): Apply `tick_slot` in the end to get the tick slot. 
137                              Defaults to False.
138
139        Returns:
140            int: Closest tick index.
141        """
142        price *= 10 ** (self._decimals0 - self._decimals1)
143        tick = math.floor(math.log(price) / math.log(1.0001))
144        if normalize:
145            tick = self.tick_slot(tick)
146        return tick
147
148    def tick_spacing(self) -> int:
149        """
150        Returns the tick spacing for the pool based on the fee tier.
151
152        Returns:
153            int: Tick spacing (1, 10, 60, or 200).
154        """
155        return {
156            100: 1,
157            500: 10,
158            3000: 60,
159            10000: 200,
160        }[self._fee]
161
162    def tick_slot(self, tick: int) -> int:
163        """
164        Calculates the tick slot for a given tick.
165
166        A tick slot is the lower bound of the spacing interval that contains the 
167        tick.
168
169        Args:
170            tick (int): The tick index.
171
172        Returns:
173            int: The tick slot index.
174        """
175        return tick - tick % self.tick_spacing()
176
177    def _get_pool_address(self) -> str:
178        factory_contract = self._w3().eth.contract(
179            address='0x1F98431c8aD98523631AE4a59f267346ea31F984', 
180            abi=get_abi('factory-abi'),
181        )
182        pool_address = factory_contract.functions \
183            .getPool(self._token1, self._token0, self._fee).call()
184        assert pool_address != "0x0000000000000000000000000000000000000000", \
185            "Could not find pool for the tokens."
186        return pool_address
187
188    def _check_initialized(self):
189        assert self._pool_contract is not None, \
190            "Instance is not initialized, " \
191            "run `instance.init()` right after creation."
192
193    def _check_available_tick(self, tick):
194        spacing = self.tick_spacing()
195        assert tick % spacing == 0, \
196            f"Tick must be multiple of {spacing} due to fee = {self._fee}, " \
197            f"{tick} given. Please, use `tick = viewer.tick_slot(tick)` or " \
198            f"`tick = viewer.calc_tick(price, normalize=True)` to normalize."
199
200
201def stream_new_blocks(w3: Web3, timeout: int = 1) -> Generator[int]:
202    """
203    Yields new block numbers as they appear on-chain.
204
205    This generator polls the chain at regular intervals to detect new blocks
206    and yields all blocks that were mined since the last check.
207
208    Args:
209        w3 (Web3): An initialized Web3 instance.
210        timeout (int): Time delay in seconds between polling. Defaults to 1.
211
212    Yields:
213        int: Block numbers in increasing order.
214    """
215    block_num_prev = None
216    while True:
217        block_num = w3.eth.block_number
218        if block_num_prev is None:
219            yield block_num
220        else:
221            yield from range(block_num_prev + 1, block_num + 1)
222        block_num_prev = block_num
223        time.sleep(timeout)
224
225
226def get_decimals(w3: Web3, token: str) -> int:
227    """
228    Returns the number of decimals for a given ERC-20 token.
229
230    Args:
231        w3 (Web3): Web3 instance.
232        token (str): Ethereum address of the token.
233
234    Returns:
235        int: Number of decimal places used by the token.
236    """
237    token_contract = w3.eth.contract(address=token, abi=get_abi('erc20-abi'))
238    decimals = token_contract.functions.decimals().call()
239    return decimals