This article provides an overview of [uniswap-viewer](https://github.com/fomalhaut88/uniswap-viewer) tool for Python that makes interaction with [Uniswap V3](https://blog.uniswap.org/uniswap-v3) easy, supporing synchronous and asynchronous interfaces. It natively uses [web3](https://pypi.org/project/web3/) and nothing else. In this article, we will walk through the installation of `uniswap-viewer`, creating viewer instances, writing a loop pulling the current prices and even make a simple signal based on the price inconsistency (the arbitrage condition). For advanced usage go to the [documentation page](https://fomalhaut88.github.io/uniswap-viewer/) where you will find all the entities of the library with the descriptions. In the version `0.1.10`, `uniswap-viewer` supports only viewing prices and tick information, without making orders or liquidity providing, so you cannot design your own complete trading algorithm with the help of `uniswap-viewer` only. But it can be used as a signal provider in some pipelines you are developing. ## Installation To install the last version of `uniswap-viewer`, just perform: `pip install uniswap-viewer`. Below, we are working with the version `0.1.10`. ## Preparations Before the use, you need a provider URL to an Ethereum node. In my case, I am using Infura as a provider [Infura](https://www.infura.io/). There is [an instruction](https://docs.metamask.io/developer-tools/dashboard/get-started/create-api/) how to get the necessary API key. Be sure, that you chose `Ethereum` and `Mainnet` in the options while creating. For other providers the process of getting an API key is usually quite similar. The access URL should look like this: `https://mainnet.infura.io/v3/<API_KEY>`, for example, if the API key is `d8fbd501ce2de719b29bed626da5daaf` the URL should be following: ``` https://mainnet.infura.io/v3/d8fbd501ce2de719b29bed626da5daaf ``` To check it in Python, install `web3` (via `pip install web3`) and access the current block number with the following commands: ```python >>> PROVIDER = "https://mainnet.infura.io/v3/d8fbd501ce2de719b29bed626da5daaf" >>> from web3 import Web3 >>> w3 = Web3(Web3.HTTPProvider(PROVIDER)) >>> block_num = w3.eth.block_number >>> print(block_num) 22480379 ``` ## Fetch the prices Once you're set up, you can access the node, it is time to design a simple price scrapper from Uniswap V3 contract. Uniswap keeps the pair of tokens like `USDT-WETH` for which there is information about the current prices, the liquidity and some other details. For more, follow [the official link](https://blog.uniswap.org/uniswap-v3). In this article, we will consider prices only for the demonstration purpose. Normally, asynchronous pattern is a good choice is such tasks because of the network interaction. So we’ll use the [viewer_async](https://fomalhaut88.github.io/uniswap-viewer/uniswap_viewer/viewer_async.html) module as our base. Also, we should take the tokens we will be working with. I suggest `USDT`, `WETH` (wrapped Ethereum) and `WBTC` (wrapped Bitcoin) as the most liquid ones at the moment. `uniswap-viewer` knows the addresses of the most common tokens (they all are listed [here](https://fomalhaut88.github.io/uniswap-viewer/uniswap_viewer/utils.html#TOKEN_MAP)), so if you need to create a viewer of `USDT-WBTC` pair, you should do following: ```python from web3 import AsyncWeb3 from uniswap_viewer.viewer_async import Viewer PROVIDER = "https://mainnet.infura.io/v3/d8fbd501ce2de719b29bed626da5daaf" w3 = AsyncWeb3(AsyncWeb3.AsyncHTTPProvider(PROVIDER)) viewer = Viewer(w3, "usdt", "wbtc", fee=3000) await viewer.init() price = await viewer.get_price() print(price) # The price of `WBTC` in `USDT` units ``` If `uniswap-viewer` does not have the token address you are going to work with, just pass the exact address instead of the symbol, for example: `"0xdAC17F958D2ee523a2206206994597C13D831ec7"` instead of `"usdt"`. Please, keep in mind that the token order matters. For example, above you can pass the tokens in the order `usdt, wbtc`, but not `wbtc, usdt`. According to the Uniswap V3 rules, in a pair the binary representation of the first token should be greater than of the second one. If you accidentally make a mistake with it, `uniswap-viewer` will kindly raise an error. Also notice, that we specified the fee as `3000` that corresponds to `0.3%`, it is the most common used value. In Uniswap V3, the allowed fees are 0.01%, 0.05%, 0.3%, and 1%, that should be defined with the values 100, 500, 3000, 10000 ([read more](https://docs.uniswap.org/concepts/protocol/fees)). ## Arbitrage signal From maths, if three relative prices are fully consistent (for example, `p1` is the price of `WETH` in `USDT` units, `p2` - `WBTC` in `WETH` units, `p3` - `USDT` in `WBTC` units), they should satisfy the equity: ``` p1 * p2 * p3 = 1 ``` This means, if we start, for example, with some amount of `USDT`, buy `WETH` for all `USDT`, after that we buy `WBTC` for just bought `WETH` and finally buying `USDT` back for the `WBTC`, in the end the amount of `USDT` must stay the same, neglecting the fee. Each buy means the multiplication of the amount by the corresponding relatice price, so the product of all prices must be `1`. It is convenient to work with a natural logarithm of prices, because this representation replaces product with sum and makes the work with fees easier. The consistency condition becomes following: ``` log(p1) + log(p2) + log(p3) = 0 ``` After all three boughts we considered above, we would pay the fee 3 times. So if the profit is still greater than 3 fees, we can consider it as an arbitrage signal. ``` abs((log(p1) + log(p2) + log(p3)) / 3) > fee ``` We added `abs` because if we get too small profit (smaller than 1), we can try to buy in a different diretion: on the first step, buying `WBTC` for `USDT` instead of `WETH` for `USDT` and so on. Let us form all this information into an algorithm the calculate the arbitrage signal. 1. Extract the relative prices `p1`, `p2`, `p3` of the pairs. 2. Calculate the arbitrage indicator as: `ind = (log(p1) + log(p2) + log(p3)) / 3` 3. As the signal, compare the absolute indicator with the fee: `abs(ind) > fee` ## Python script Now, we are to write a Python script that requests the net for the prices, checks the arbirage condition and prints if the signal happens. First of all, we create and initialize all the necessary viewers: ```python from uniswap_viewer.viewer_async import Viewer ... viewers = [ Viewer(w3, "weth", "wbtc", fee=3000), Viewer(w3, "usdt", "weth", fee=3000), Viewer(w3, "usdt", "wbtc", fee=3000), ] await asyncio.gather(*( viewer.init() for viewer in viewers )) ``` Notice, that the third viewer represents the pair `USDT-WBTC` instead of `WBTC-USDT` that breaks the harmony. This means, we will have to invert the price to get the correct `p3` (using the formula: `p3 = 1.0 / price`). Since we are researching the arbitrage opportunity at a fixed moment of time, we must ensure that all the prices belong to the same block number. Also it is a good idea to request the prices once on each block update, in order not to load the node too hard. To do this, `uniswap-viewer` has a generator called [stream_new_blocks](https://fomalhaut88.github.io/uniswap-viewer/uniswap_viewer/viewer_async.html#stream_new_blocks) that checks for new block and yields its number. Also, the method [Viewer.get_price](https://fomalhaut88.github.io/uniswap-viewer/uniswap_viewer/viewer_async.html#Viewer.get_price) supports the block number as an argument. So the loop looks like this: ```python from uniswap_viewer.viewer_async import stream_new_blocks ... async for block_num in stream_new_blocks(w3, timeout=8): prices = await asyncio.gather(*( viewer.get_price(block_num=block_num) for viewer in viewers )) ``` And funally, we are adding the arbitrage condition according to the algorithm we discussed above: ```python import math ... p1 = prices[0] p2 = prices[1] p3 = 1.0 / prices[2] # Due to the token reversed order ind = (math.log(p1) + math.log(p2) + math.log(p3)) / 3.0 print(block_num, ind, prices) if abs(ind) > 0.003: print(f"Signal in block {block_num}: {ind}") ``` Collecting all together, here is the full working script: ```python import math import asyncio from web3 import AsyncWeb3 from uniswap_viewer.viewer_async import Viewer, stream_new_blocks # Provider URL PROVIDER = "https://mainnet.infura.io/v3/d8fbd501ce2de719b29bed626da5daaf" async def main(): # Create a Web3 instance w3 = AsyncWeb3(AsyncWeb3.AsyncHTTPProvider(PROVIDER)) # Create the asynchronous viewers viewers = [ Viewer(w3, "weth", "wbtc", fee=3000), Viewer(w3, "usdt", "weth", fee=3000), Viewer(w3, "usdt", "wbtc", fee=3000), ] # Initialize the asynchronous viewers await asyncio.gather(*( viewer.init() for viewer in viewers )) # Loop the blocks async for block_num in stream_new_blocks(w3, timeout=8): # Request for the prices prices = await asyncio.gather(*( viewer.get_price(block_num=block_num) for viewer in viewers )) # Calculate the arbitrage indicator p1 = prices[0] p2 = prices[1] p3 = 1.0 / prices[2] # Due to the token reversed order ind = (math.log(p1) + math.log(p2) + math.log(p3)) / 3.0 # Print the prices print(block_num, ind, prices) # Print signal information if abs(ind) > 0.003: print(f"Signal in block {block_num}: {ind}") if __name__ == "__main__": loop = asyncio.new_event_loop() loop.run_until_complete(main()) ```