Find the optimal trade route (approximately)

[ErikBjare]

See past discussion/research in https://github.com/shanefontaine/uniswap-python/issues/69

[qraxiss]

This code is written for Uniswap, it works for forks. But it doesn't work for itself, very interesting

[qraxiss]

from uniswap import Uniswap
from web3 import Web3
from config import provider

from tokens import tokens, Token

t1 : Token = [token for token in tokens.values() if token.symbol == "ETH"][0]
t2 : Token = [token for token in tokens.values() if token.symbol == "BAT"][0]

web3 = Web3(Web3.HTTPProvider(provider))

uni_v2 = Uniswap(
        address=None,
        private_key=None,
        version=2,
        web3=web3
)

uni_v3 = Uniswap(
        address=None,
        private_key=None,
        version=3,
        web3=web3
)


p1 = uni_v2.get_price_input(t1.address, t2.address, t1.qty) / t1.qty
p2 = uni_v3.get_price_input(t1.address, t2.address, t1.qty) / t1.qty
print(max(p1,p2))

p1 = uni_v2.get_price_output(t2.address, t1.address, t2.qty, route = [t2.address, t1.address]) / t2.qty
p2 = uni_v3.get_price_output(t2.address, t1.address, t2.qty) / t2.qty
print(min(p1,p2))

If there is sufficient liquidity in either the v2 pool or the v3 pool, this is the best code we have for now that gives the closest code to the real price. The reason I say the closest is that if the fee is taken from a different pool than 3000, there will be a price difference again, but in pools with a fee of 3000, he can get it perfectly. For this, you can also write a code that calculates different fee rates. But I don't know what we can do for liquidite. My web3 knowledge is insufficient here. If anyone knows how much we can get by calculating the current liquidity of the pool without asking for a price request, we can add them and reach the perfect price.

[qraxiss]

My solution is here. +- 1 in 1000 margin of error.

If a PR is requested, I can write a more professional one. But for me this is enough.

from concurrent.futures import ThreadPoolExecutor
from collections import defaultdict
from itertools import permutations
from uniswap import Uniswap
from web3 import Web3


class Token:
    def __init__(self, address, name, symbol, decimals) -> None:
        self.address = Web3.toChecksumAddress(address)
        self._raw_address = address
        self.name = name
        self.symbol = symbol
        self.decimals = decimals
        self.qty = 10 ** self.decimals


class UniswapPrice:
    sides = {
        'sell': 'get_price_output',
        'buy': 'get_price_input'
    }

    percentages = [50, 100]
    fees = [100, 300, 3000, 10000]

    def __init__(self, v2, v3):
        self.swap = {
            'v2': {
                'exchange': v2,
                'percentages': self.percentages,
                'fees': [3000]
            },
            'v3': {
                'exchange': v3,
                'percentages': self.percentages,
                'fees': self.fees
            }
        }

    def find_best(self):
        prices = [perm[0] + perm[1] for perm in permutations(
            self.pool_prices('different'), 2)] + self.pool_prices('single')
        if len(prices) == 0:
            return
        if self.side == 'sell':
            return min(prices)
        elif self.side == 'buy':
            return max(prices)

    def get_price(self, exchange, fee, qty):
        queue = [self.t0.address, self.t1.address] if self.side == 'buy' else [
            self.t1.address, self.t0.address]
        price_func = getattr(exchange, self.sides[self.side])
        return price_func(*queue, qty, fee=fee) / self.t1.qty

    def pool_prices(self, price_type):
        if price_type == 'single':
            perc = 100
        elif price_type == 'different':
            perc = 50

        return [
            fee[perc]
            for exchanges in self.futures.values()
            for fee in exchanges.values()
            if isinstance(fee[perc], float)
        ]

    def get_best_price(self, t0, t1, side):
        self.t0 = t0
        self.t1 = t1
        self.side = side

        self.get_prices()
        return self.find_best()

    def get_prices(self):
        self.futures = defaultdict(dict)

        with ThreadPoolExecutor(max_workers=10) as executor:
            for exchange_name, exchange_data in self.swap.items():
                self.futures[exchange_name] = defaultdict(dict)
                for fee in exchange_data['fees']:
                    self.futures[exchange_name][fee] = defaultdict(dict)
                    for percentage in exchange_data['percentages']:
                        self.futures[exchange_name][fee][percentage] = defaultdict(
                            dict)
                        qty = self.t0.qty * percentage // 100
                        self.futures[exchange_name][fee][percentage] = executor.submit(
                            self.get_price, exchange_data['exchange'], fee, qty
                        )

            for exchange_name, exchange_fees in self.futures.items():
                self.futures[exchange_name] = dict(exchange_fees)
                for fee, percentage_data in exchange_fees.items():
                    self.futures[exchange_name][fee] = dict(percentage_data)
                    for percentage, future in percentage_data.items():
                        try:
                            self.futures[exchange_name][fee][percentage] = future.result(
                            )
                        except:
                            pass


web3 = Web3(Web3.HTTPProvider("YOUR-JSON-RPC-URL"))

uni_v2 = Uniswap(
    address=None,
    private_key=None,
    version=2,
    web3=web3
)

uni_v3 = Uniswap(
    address=None,
    private_key=None,
    version=3,
    web3=web3
)

t0 = Token(**{
               "symbol": "ETH",
               "name": "Wrapped Ethereum",
               "decimals": 18,
               "address": "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"
           }
)


t1 = Token(**{
               "symbol": "WBTC",
               "name": "Wrapped Bitcoin",
               "decimals": 8,
               "address": "0x2260fac5e5542a773aa44fbcfedf7c193bc2c599"
           }
)


price_calculator = UniswapPrice(uni_v2, uni_v3)

print(price_calculator.get_best_price(t0, t1, 'buy'),
      price_calculator.get_best_price(t0, t1, 'sell'))

[GitUserForMaster01]

Hi, do I need to add an private key to use this example. I got "web3.exceptions.ContractLogicError: execution reverted" during execution

[ErikBjare]

@GitUserForMaster01 No, just reading prices do not require a private key.

One or more of the parameters you pass are most likely wrong, have a look at this issue: https://github.com/uniswap-python/uniswap-python/issues/356