"""Asset pricing model for Uniswap V2 and V3 like exchanges."""
import abc
import datetime
from dataclasses import dataclass
from decimal import Decimal, ROUND_DOWN
from typing import Callable, Optional
from web3 import Web3
from tradeexecutor.state.identifier import TradingPairIdentifier
from tradeexecutor.state.types import USDollarAmount, BPS, USDollarPrice
from tradeexecutor.strategy.execution_model import ExecutionModel
from tradeexecutor.strategy.routing import RoutingModel
from tradeexecutor.strategy.universe_model import StrategyExecutionUniverse
from tradeexecutor.strategy.trading_strategy_universe import translate_trading_pair
from tradeexecutor.strategy.pricing_model import PricingModel
from tradeexecutor.strategy.trade_pricing import TradePricing
from tradeexecutor.strategy.dust import DEFAULT_DUST_EPSILON
from tradeexecutor.ethereum.routing_model import EthereumRoutingModel
from eth_defi.uniswap_v2.deployment import UniswapV2Deployment
from eth_defi.uniswap_v3.deployment import UniswapV3Deployment
from tradingstrategy.pair import PandasPairUniverse
deployment_types = (UniswapV2Deployment | UniswapV3Deployment)
#: TODO: No good data yet for the value used here
#:
LP_FEE_VALIDATION_EPSILON = 0.001
[docs]class EthereumPricingModel(PricingModel):
"""Get a price for the asset.
Needed for various aspects
- Revaluate portfolio positiosn
- Estimate buy/sell price for the live trading so we can calculate slippage
- Get the historical price in backtesting
Timestamp is passed to the pricing method. However we expect it only be honoured during
the backtesting - live execution may always use the latest price.
.. note ::
For example, in futures markets there could be different fees
on buy/sell transctions.
Used by UniswapV2LivePricing and UniswapV3LivePricing
"""
[docs] def __init__(self,
web3: Web3,
pair_universe: PandasPairUniverse,
routing_model: EthereumRoutingModel,
very_small_amount: Decimal,
epsilon: Optional[float] = LP_FEE_VALIDATION_EPSILON,
):
assert isinstance(web3, Web3)
assert isinstance(pair_universe, PandasPairUniverse), f"Expected PandasPairUniverse, got {pair_universe.__class__}"
self.web3 = web3
self.pair_universe = pair_universe
self.very_small_amount = very_small_amount
self.routing_model = routing_model
self.epsilon = epsilon
assert isinstance(self.very_small_amount, Decimal)
[docs] def get_pair_for_id(self, internal_id: int) -> Optional[TradingPairIdentifier]:
"""Look up a trading pair.
Useful if a strategy is only dealing with pair integer ids.
:return:
None if the price data is not available
"""
pair = self.pair_universe.get_pair_by_id(internal_id)
return translate_trading_pair(pair) if pair else None
def check_supported_quote_token(self, pair: TradingPairIdentifier):
assert pair.quote.address == self.routing_model.reserve_token_address, f"Quote token {self.routing_model.reserve_token_address} not supported for pair {pair}, pair tokens are {pair.base.address} - {pair.quote.address}"
[docs] def get_mid_price(self,
ts: datetime.datetime,
pair: TradingPairIdentifier) -> USDollarAmount:
"""Get the mid price from Uniswap pool.
Gets tricky, because we calculate dollar mid-price, not
quote token midprice.
Mid price is an non-trddeable price between the best ask
and the best pid.
:param ts:
Timestamp. Ignored for live pricing models.
:param pair:
Which trading pair price we query.
:return:
The mid price for the pair at a timestamp.
"""
# TODO: Use native Uniswap router functions to get the mid price
# Here we are using a hack)
bp = self.get_buy_price(ts, pair, self.very_small_amount)
sp = self.get_sell_price(ts, pair, self.very_small_amount)
return (bp.price + sp.price) / 2
[docs] def get_pair_fee(self,
ts: datetime.datetime,
pair: TradingPairIdentifier,
) -> Optional[float]:
"""Estimate the trading/LP fees for a trading pair.
This information can come either from the exchange itself (Uni v2 compatibles),
or from the trading pair (Uni v3).
The return value is used to fill the
fee values for any newly opened trades.
:param ts:
Timestamp of the trade. Note that currently
fees do not vary over time, but might
do so in the future.
:param pair:
Trading pair for which we want to have the fee.
Can be left empty if the underlying exchange is always
offering the same fee.
:return:
The estimated trading fee, expressed as %.
Returns None if the fee information is not available.
This can be different from zero fees.
"""
return pair.fee
[docs] def validate_mid_price_for_sell(self, lp_fee, mid_price, price, quantity):
"""Validate the mid price calculation for a sell trade.
Should basically have:
lp_fee = (mid_price - price)/mid_price * float(quantity)
:param lp_fee:
The fee that is paid to the LPs.
:param mid_price:
The mid price of the pair.
:param price:
The price of the trade.
:param quantity:
The quantity of the trade.
"""
raise NotImplementedError("Cannot use mid-price here")
#value = lp_fee - (mid_price - price)/mid_price * float(quantity)
#assert abs(value) < self.epsilon, f"Bad lp fee calculation: {lp_fee}, {mid_price}, {price}, {quantity}.\n" \
# f"Value {value} < epsilon {self.epsilon}\n"
[docs] def validate_mid_price_for_buy(self, lp_fee, price, mid_price, reserve):
"""Validate the mid price calculation for a buy trade.
Should basically have:
lp_fee = (price - mid_price)/price * float(reserve)
:param lp_fee:
The fee that is paid to the LPs.
:param price:
The price of the trade.
:param mid_price:
The mid price of the pair.
:param reserve:
The reserve of the trade.
"""
raise NotImplementedError("Cannot use mid-price here")
#assert lp_fee - (price - mid_price)/price * float(reserve) < self.epsilon, f"Bad lp fee calculation: {lp_fee}, {mid_price}, {price}, {reserve}"
[docs] @abc.abstractmethod
def get_uniswap(self, target_pair: TradingPairIdentifier) -> deployment_types:
"""Helper function to speed up Uniswap v2 or v3 deployment resolution."""
[docs] @abc.abstractmethod
def get_sell_price(self,
ts: datetime.datetime,
pair: TradingPairIdentifier,
quantity: Optional[Decimal]) -> TradePricing:
"""Get the sell price for an asset.
:param ts:
When to get the price.
Used in backtesting.
Live models may ignore.
:param pair:
Trading pair we are intereted in
:param quantity:
If the sel quantity is known, get the price with price impact.
:return:
Price structure for the trade.
"""
[docs] @abc.abstractmethod
def get_buy_price(self,
ts: datetime.datetime,
pair: TradingPairIdentifier,
reserve: Optional[Decimal]
) -> TradePricing:
"""Get the sell price for an asset.
:param ts:
When to get the price.
Used in backtesting.
Live models may ignore.
:param pair:
Trading pair we are intereted in
:param reserve:
If the buy token quantity quantity is known,
get the buy price with price impact.
:return:
Price structure for the trade.
"""
#: This factory creates a new pricing model for each trade cycle.
#: Pricing model depends on the trading universe that may change for each strategy tick,
#: as new trading pairs appear.
#: Thus, we need to reconstruct pricing model as the start of the each tick.
#:
#: TODO: Convert to protocol
#:
PricingModelFactory = Callable[[ExecutionModel, StrategyExecutionUniverse, RoutingModel], PricingModel]