Source code for tradeexecutor.ethereum.uniswap_v3.uniswap_v3_live_pricing

"""Uniswap v3 live pricing.

Directly asks Uniswap v3 asset price from Uniswap pair contract
and JSON-RPC API.
"""
import logging
import datetime
from decimal import Decimal
from typing import Optional, Dict

from eth_defi.provider.broken_provider import get_block_tip_latency
from web3 import Web3

from tradeexecutor.ethereum.uniswap_v3.uniswap_v3_execution import UniswapV3Execution
from tradeexecutor.ethereum.uniswap_v3.uniswap_v3_routing import UniswapV3Routing, route_tokens, get_uniswap_for_pair
from tradeexecutor.ethereum.eth_pricing_model import EthereumPricingModel, LP_FEE_VALIDATION_EPSILON
from tradeexecutor.state.identifier import TradingPairIdentifier
from tradeexecutor.strategy.execution_model import ExecutionModel
from tradeexecutor.state.types import USDollarAmount
from tradeexecutor.strategy.trading_strategy_universe import TradingStrategyUniverse
from tradeexecutor.strategy.trade_pricing import TradePricing
from tradingstrategy.pair import PandasPairUniverse

from eth_defi.uniswap_v3.price import UniswapV3PriceHelper, estimate_sell_received_amount, estimate_buy_received_amount, get_onchain_price
from eth_defi.uniswap_v3.deployment import UniswapV3Deployment

logger = logging.getLogger(__name__)


[docs]class UniswapV3LivePricing(EthereumPricingModel): """Always pull the latest dollar price for an asset from Uniswap v3 deployment. Supports - Two-way BUSD -> Cake - Three-way trades BUSD -> BNB -> Cake ... within a single, uniswap v3 like exchange. .. note :: If a trade quantity/currency amount is not given uses a "default small value" that is 0.1. Depending on the token, this value might be too much/too little, as Uniswap fixed point math starts to break for very small amounts. For example, for USDC trade 10 cents is already quite low. More information - `About ask and bid <https://www.investopedia.com/terms/b/bid-and-ask.asp>`_: """
[docs] def __init__(self, web3: Web3, pair_universe: PandasPairUniverse, routing_model: UniswapV3Routing, very_small_amount=Decimal("0.10"), epsilon: Optional[float] = LP_FEE_VALIDATION_EPSILON, ): # assert isinstance(routing_model, UniswapV3SimpleRoutingModel) self.uniswap_cache: Dict[TradingPairIdentifier, UniswapV3Deployment] = {} super().__init__( web3, pair_universe, routing_model, very_small_amount, epsilon )
[docs] def get_pair_fee_multiplier(self, ts, pair): """Outdated: Uniswap V3 pairs get fees in raw format e.g. 3000 instead of 0.3% Now, all pairs are already in multiplier format, so no changes necessary. """ return super().get_pair_fee(ts, pair)
[docs] def get_uniswap(self, target_pair: TradingPairIdentifier) -> UniswapV3Deployment: """Helper function to speed up Uniswap v3 deployment resolution.""" if target_pair not in self.uniswap_cache: self.uniswap_cache[target_pair] = get_uniswap_for_pair( self.web3, self.routing_model.address_map, target_pair ) return self.uniswap_cache[target_pair]
def get_price_helper(self, target_pair: TradingPairIdentifier) -> UniswapV3PriceHelper: uniswap_v3 = self.get_uniswap(target_pair) return UniswapV3PriceHelper(uniswap_v3)
[docs] def get_sell_price(self, ts: datetime.datetime, pair: TradingPairIdentifier, quantity: Optional[Decimal], ) -> TradePricing: """Get live price on Uniswap.""" assert pair.is_spot() if quantity is None: quantity = Decimal(self.very_small_amount) assert isinstance(quantity, Decimal) target_pair, intermediate_pair = self.routing_model.route_pair(self.pair_universe, pair) base_addr, quote_addr, intermediate_addr = route_tokens(target_pair, intermediate_pair) # In three token trades, be careful to use the correct reserve token quantity_raw = target_pair.base.convert_to_raw_amount(quantity) reverse_token_order = target_pair.has_reverse_token_order() if intermediate_pair: path = [base_addr, intermediate_addr, quote_addr] fees = [intermediate_pair.fee, target_pair.fee] total_fee_pct = 1 - (1-fees[0]) * (1-fees[1]) received_raw = estimate_sell_received_amount( uniswap=self.get_uniswap(target_pair), base_token_address=base_addr, quote_token_address=quote_addr, quantity=quantity_raw, target_pair_fee=int(target_pair.fee * 1_000_000), intermediate_token_address=intermediate_addr, intermediate_pair_fee=int(intermediate_pair.fee * 1_000_000) if intermediate_pair else None, ) block_number = None else: path = [base_addr, quote_addr] fees = [target_pair.fee] total_fee_pct = 1 - (1-fees[0]) web3 = self.web3 block_number = max(1, web3.eth.block_number - get_block_tip_latency(web3)) target_pair_fee = int(target_pair.fee * 1_000_000) try: received_raw = estimate_sell_received_amount( uniswap=self.get_uniswap(target_pair), base_token_address=base_addr, quote_token_address=quote_addr, quantity=quantity_raw, target_pair_fee=target_pair_fee, intermediate_token_address=intermediate_addr, intermediate_pair_fee=int(intermediate_pair.fee * 1_000_000) if intermediate_pair else None, block_identifier=block_number, ) except Exception as e: # Add more helpful debug context raise RuntimeError(f"Could not get valid price for {target_pair}\n{base_addr}-{quote_addr} with intermediate {intermediate_addr}, quantity:{quantity} fee:{target_pair_fee}") from e if intermediate_pair: received = intermediate_pair.quote.convert_to_decimal(received_raw) path = [intermediate_pair, target_pair] else: received = target_pair.quote.convert_to_decimal(received_raw) price = float(received / quantity) path = [target_pair] price = float(received / quantity) if intermediate_pair: mid_price = price / (1 - self.get_pair_fee_multiplier(ts, target_pair)) / (1 - self.get_pair_fee_multiplier(ts, intermediate_pair)) else: # Read mid-price at the mid point of Uni v3 liquidity, # at our block number mid_price = get_onchain_price( self.web3, target_pair.pool_address, block_identifier=block_number, reverse_token_order=reverse_token_order, ) mid_price = float(mid_price) assert price <= mid_price, f"Bad pricing: {price}, {mid_price}" lp_fee = float(quantity) * total_fee_pct # self.validate_mid_price_for_sell(lp_fee, mid_price, price, quantity) return TradePricing( price=price, mid_price=mid_price, lp_fee=[lp_fee], pair_fee=fees, side=False, path=path, read_at=datetime.datetime.utcnow(), block_number=block_number, token_in=quantity, token_out=received, )
[docs] def get_buy_price(self, ts: datetime.datetime, pair: TradingPairIdentifier, reserve: Optional[Decimal], ) -> TradePricing: """Get live price on Uniswap. :param reserve: The buy size in quote token e.g. in dollars :return: Price for one reserve unit e.g. a dollar """ assert pair.is_spot() if reserve is None: reserve = Decimal(self.very_small_amount) else: assert isinstance(reserve, Decimal), f"Reserve must be decimal, got {reserve.__class__}: {reserve}" target_pair, intermediate_pair = self.routing_model.route_pair(self.pair_universe, pair) base_addr, quote_addr, intermediate_addr = route_tokens(target_pair, intermediate_pair) reverse_token_order = target_pair.has_reverse_token_order() # In three token trades, be careful to use the correct reserve token if intermediate_pair is not None: reserve_raw = intermediate_pair.quote.convert_to_raw_amount(reserve) self.check_supported_quote_token(intermediate_pair) path = [quote_addr, intermediate_addr, base_addr] fees = [intermediate_pair.fee, target_pair.fee] total_fee_pct = 1 - (1 - fees[0]) * (1-fees[1]) block_number = None token_raw_received = estimate_buy_received_amount( uniswap=self.get_uniswap(target_pair), base_token_address=base_addr, quote_token_address=quote_addr, quantity=reserve_raw, target_pair_fee=int(target_pair.fee * 1_000_000), intermediate_token_address=intermediate_addr, intermediate_pair_fee=int(intermediate_pair.fee * 1_000_000) if intermediate_pair else None, ) else: web3 = self.web3 block_number = max(1, web3.eth.block_number - get_block_tip_latency(web3)) reserve_raw = target_pair.quote.convert_to_raw_amount(reserve) self.check_supported_quote_token(pair) token_raw_received = estimate_buy_received_amount( uniswap=self.get_uniswap(target_pair), base_token_address=base_addr, quote_token_address=quote_addr, quantity=reserve_raw, target_pair_fee=int(target_pair.fee * 1_000_000), intermediate_token_address=intermediate_addr, intermediate_pair_fee=int(intermediate_pair.fee * 1_000_000) if intermediate_pair else None, block_identifier=block_number, ) path = [quote_addr, base_addr] fees = [target_pair.fee] total_fee_pct = 1 - (1 - fees[0]) token_received = target_pair.base.convert_to_decimal(token_raw_received) fee = self.get_pair_fee_multiplier(ts, pair) assert fee is not None, "Uniswap v3 fee data missing" price = float(reserve / token_received) lp_fee = float(reserve) * total_fee_pct if intermediate_pair: mid_price = price * (1 - self.get_pair_fee_multiplier(ts, intermediate_pair)) * (1 - self.get_pair_fee_multiplier(ts, target_pair)) path = [intermediate_pair, target_pair] else: # Read mid-price at the mid point of Uni v3 liquidity, # at our block number mid_price = get_onchain_price( self.web3, target_pair.pool_address, block_identifier=block_number, reverse_token_order=reverse_token_order, ) mid_price = float(mid_price) path = [target_pair] assert price >= mid_price, f"Bad pricing: {price}, {mid_price}" # self.validate_mid_price_for_buy(lp_fee, price, mid_price, reserve) return TradePricing( price=float(price), mid_price=float(mid_price), lp_fee=[lp_fee], pair_fee=fees, market_feed_delay=datetime.timedelta(seconds=0), side=True, path=path, read_at=datetime.datetime.utcnow(), block_number=block_number, token_in=reserve, token_out=token_received, )
[docs]def uniswap_v3_live_pricing_factory( execution_model: ExecutionModel, universe: TradingStrategyUniverse, routing_model: UniswapV3Routing, ) -> UniswapV3LivePricing: assert isinstance(universe, TradingStrategyUniverse) assert isinstance(execution_model, (UniswapV3Execution)), f"Execution model not compatible with this execution model. Received {execution_model}" assert isinstance(routing_model, UniswapV3Routing), f"This pricing method only works with Uniswap routing model, we received {routing_model}" web3 = execution_model.web3 return UniswapV3LivePricing( web3, universe.data_universe.pairs, routing_model )