uniswap_viewer.viewer_async

  1"""
  2viewer_async.py - Asynchronous interface for accessing Uniswap V3 pool data.
  3
  4This module provides an async viewer for interacting with Uniswap V3 pools
  5using asynchronous Web3 providers (e.g., Web3 AsyncMiddleware). It allows
  6retrieving token prices, tick data, and computing tick-related values.
  7
  8Main components:
  9- `Viewer`: an async class for accessing Uniswap V3 pools.
 10- `stream_new_blocks`: async generator for tracking new block numbers.
 11- `get_decimals`: async utility to fetch token decimals from an ERC-20 token.
 12"""
 13
 14import math
 15import asyncio
 16import weakref
 17from typing import Optional, Dict, Any
 18from collections.abc import Generator
 19
 20from web3 import AsyncWeb3
 21
 22from .utils import get_abi, TICKS_KEYS, prepare_token_address
 23
 24
 25class Viewer:
 26    """
 27    Asynchronous Uniswap V3 pool viewer.
 28
 29    This class allows querying price and tick information from a Uniswap V3 pool
 30    in an async context.
 31    """
 32
 33    def __init__(self, w3: AsyncWeb3, token0: str, token1: str, 
 34                 fee: int = 3000):
 35        """
 36        Initializes the Viewer instance.
 37
 38        Args:
 39            w3: Async-enabled Web3 instance.
 40            token0 (str): First token address (must be > token1 bitwise).
 41            token1 (str): Second token address.
 42            fee (int, optional): Pool fee (bps). One of 100, 500, 3000, 10000.
 43                                 Defaults to 3000 (0.3%).
 44
 45        Raises:
 46            AssertionError: If invalid fee or token ordering.
 47        """
 48        # Prepare arguments
 49        token0 = prepare_token_address(token0)
 50        token1 = prepare_token_address(token1)
 51
 52        # Check arguments
 53        assert fee in (100, 500, 3000, 10000), \
 54            "Allowed fee must be in (100, 500, 3000, 10000) " \
 55            "that corresponds to 0.01%, 0.05%, 0.3%, 1%."
 56
 57        assert int(token0, 16) > int(token1, 16), \
 58            "token0 must go after token1 in bit representation."
 59
 60        # Attributes
 61        self._w3 = weakref.ref(w3)
 62        self._token0 = self._w3().to_checksum_address(token0)
 63        self._token1 = self._w3().to_checksum_address(token1)
 64        self._fee = fee
 65        self._decimals0 = None
 66        self._decimals1 = None
 67        self._pool_address = None
 68        self._pool_contract = None
 69
 70    async def init(self):
 71        """
 72        Initializes decimals and resolves the pool address and contract.
 73
 74        This method must be awaited before calling other methods that
 75        depend on pool context (e.g., price, tick queries).
 76
 77        Raises:
 78            AssertionError: If pool is not found on-chain.
 79        """
 80        self._decimals0 = await get_decimals(self._w3(), self._token0)
 81        self._decimals1 = await get_decimals(self._w3(), self._token1)
 82        self._pool_address = await self._get_pool_address()
 83        self._pool_contract = self._w3().eth.contract(
 84            address=self._pool_address, abi=get_abi('pool-abi')
 85        )
 86
 87    async def get_price(self, block_num: Optional[int] = None) -> float:
 88        """
 89        Asynchronously fetches the pool price as a float.
 90
 91        Args:
 92            block_num (int, optional): Specific block number, or latest if None.
 93
 94        Returns:
 95            float: Price of token0 in terms of token1.
 96        """
 97        self._check_initialized()
 98        slot0 = await self._pool_contract.functions.slot0().call(
 99            block_identifier=block_num
100        )
101        sqrtPriceX96 = slot0[0]
102        price = (sqrtPriceX96 / (2**96)) ** 2
103        price *= 10 ** (self._decimals1 - self._decimals0)
104        return price
105
106    async def get_tick_data(self, tick: int, 
107                            block_num: Optional[int] = None) -> Dict[str, Any]:
108        """
109        Retrieves detailed tick information from the pool.
110
111        Args:
112            tick (int): The tick index.
113            block_num (int, optional): Block to query from. Defaults to latest.
114
115        Returns:
116            dict: Tick data fields mapped by standard TICKS_KEYS.
117        """
118        self._check_initialized()
119        self._check_available_tick(tick)
120        tick_data = await self._pool_contract.functions.ticks(tick).call(
121            block_identifier=block_num
122        )
123        return dict(zip(TICKS_KEYS, tick_data))
124
125    def calc_tick(self, price: float, normalize: bool = False) -> int:
126        """
127        Calculates the tick index corresponding to a given price.
128
129        Args:
130            price (float): Price of token0 in token1 units.
131            normalize (bool): Apply `tick_slot` in the end to get the tick slot. 
132                              Defaults to False.
133
134        Returns:
135            int: Closest tick index.
136        """
137        price *= 10 ** (self._decimals0 - self._decimals1)
138        tick = math.floor(math.log(price) / math.log(1.0001))
139        if normalize:
140            tick = self.tick_slot(tick)
141        return tick
142
143    def tick_spacing(self) -> int:
144        """
145        Returns the tick spacing associated with the pool's fee tier.
146
147        Returns:
148            int: Tick spacing (1, 10, 60, or 200).
149        """
150        return {
151            100: 1,
152            500: 10,
153            3000: 60,
154            10000: 200,
155        }[self._fee]
156
157    def tick_slot(self, tick: int) -> int:
158        """
159        Calculates the tick slot for a given tick index.
160
161        Args:
162            tick (int): Tick index.
163
164        Returns:
165            int: Lower-bound of the spacing interval containing the tick.
166        """
167        return tick - tick % self.tick_spacing()
168
169    async def _get_pool_address(self):
170        factory_contract = self._w3().eth.contract(
171            address='0x1F98431c8aD98523631AE4a59f267346ea31F984', 
172            abi=get_abi('factory-abi'),
173        )
174        pool_address = await factory_contract.functions \
175            .getPool(self._token1, self._token0, self._fee).call()
176        assert pool_address != "0x0000000000000000000000000000000000000000", \
177            "Could not find pool for the tokens."
178        return pool_address
179
180    def _check_initialized(self):
181        assert self._pool_contract is not None, \
182            "Instance is not initialized, " \
183            "run `await instance.init()` right after creation."
184
185    def _check_available_tick(self, tick):
186        spacing = self.tick_spacing()
187        assert tick % spacing == 0, \
188            f"Tick must be multiple of {spacing} due to fee = {self._fee}, " \
189            f"{tick} given. Please, use `tick = viewer.tick_slot(tick)` or " \
190            f"`tick = viewer.calc_tick(price, normalize=True)` to normalize."
191
192
193async def stream_new_blocks(w3: AsyncWeb3, timeout: int = 1) -> Generator[int]:
194    """
195    Asynchronously yields new block numbers as they appear on-chain.
196
197    Polls the chain periodically and yields any new blocks that have appeared
198    since the last poll.
199
200    Args:
201        w3: Async Web3 instance.
202        timeout (int): Poll interval in seconds. Defaults to 1.
203
204    Yields:
205        int: New block numbers in sequence.
206    """
207    block_num_prev = None
208    while True:
209        block_num = await w3.eth.block_number
210        if block_num_prev is None:
211            yield block_num
212        else:
213            for num in range(block_num_prev + 1, block_num + 1):
214                yield num
215        block_num_prev = block_num
216        await asyncio.sleep(timeout)
217
218
219async def get_decimals(w3: AsyncWeb3, token: str) -> int:
220    """
221    Asynchronously fetches the number of decimals of an ERC-20 token.
222
223    Args:
224        w3: Async Web3 instance.
225        token (str): Ethereum address of the ERC-20 token.
226
227    Returns:
228        int: Number of decimals.
229    """
230    token_contract = w3.eth.contract(address=token, abi=get_abi('erc20-abi'))
231    decimals = await token_contract.functions.decimals().call()
232    return decimals