"""Execution model where trade happens directly on Uniswap v2 style exchange."""
import datetime
from decimal import Decimal
import logging
from web3 import Web3
from eth_defi.hotwallet import HotWallet
from eth_defi.uniswap_v2.analysis import TradeSuccess, TradeFail
from eth_defi.uniswap_v3.deployment import UniswapV3Deployment
from eth_defi.uniswap_v3.price import UniswapV3PriceHelper, estimate_sell_received_amount
from eth_defi.uniswap_v3.analysis import analyse_trade_by_receipt
from eth_defi.uniswap_v3.deployment import mock_partial_deployment_for_analysis
from tradeexecutor.ethereum.tx import TransactionBuilder
from tradeexecutor.state.identifier import TradingPairIdentifier
#from tradeexecutor.strategy.execution_model import ExecutionModel
from tradeexecutor.ethereum.execution import EthereumExecution
logger = logging.getLogger(__name__)
[docs]class UniswapV3Execution(EthereumExecution):
"""Run order execution on a single Uniswap v3 style exchanges."""
[docs] def __init__(self,
tx_builder: TransactionBuilder,
min_balance_threshold=Decimal("0.5"),
confirmation_block_count=6,
confirmation_timeout=datetime.timedelta(minutes=5),
max_slippage: float = 0.01,
stop_on_execution_failure=True,
swap_gas_fee_limit=2_000_000,
mainnet_fork=False,
):
"""
:param tx_builder:
Hot wallet instance used for this execution
:param min_balance_threshold:
Abort execution if our hot wallet gas fee balance drops below this
:param confirmation_block_count:
How many blocks to wait for the receipt confirmations to mitigate unstable chain tip issues
:param confirmation_timeout:
How long we wait transactions to clear
:param stop_on_execution_failure:
Raise an exception if any of the trades fail top execute
:param max_slippage:
Max slippage tolerance per trade. 0.01 is 1%.
"""
assert isinstance(tx_builder, TransactionBuilder), f"Got: {tx_builder}"
super().__init__(
tx_builder,
min_balance_threshold,
confirmation_block_count,
confirmation_timeout,
max_slippage,
stop_on_execution_failure,
mainnet_fork=mainnet_fork,
)
def analyse_post_trade(self):
pass
def analyse_trade_by_receipt(
self,
web3: Web3,
*,
deployment: UniswapV3Deployment,
tx: dict,
tx_hash: str,
tx_receipt: dict,
input_args: tuple | None = None,
pair_fee: float | None = None,
) -> (TradeSuccess | TradeFail):
return analyse_trade_by_receipt(web3, deployment, tx, tx_hash, tx_receipt, input_args)
def mock_partial_deployment_for_analysis(
self,
web3: Web3,
router_address: str
) -> UniswapV3Deployment:
return mock_partial_deployment_for_analysis(web3, router_address)
[docs] def is_v3(self) -> bool:
"""Returns true if instance is related to Uniswap V3, else false.
Kind of a hack to be able to share resolve trades function amongst v2 and v3."""
return True
[docs]def get_current_price(web3: Web3, uniswap: UniswapV3Deployment, pair: TradingPairIdentifier, quantity=Decimal(1)) -> float:
"""Get a price from Uniswap v3 pool, assuming you are selling 1 unit of base token.
Does decimal adjustment.
:return: Price in quote token.
"""
quantity_raw = pair.base.convert_to_raw_amount(quantity)
out_raw = estimate_sell_received_amount(
uniswap=uniswap,
base_token_address=pair.base.checksum_address,
quote_token_address=pair.quote.checksum_address,
quantity=quantity_raw,
target_pair_fee=int(pair.fee * 1_000_000),
)
return float(pair.quote.convert_to_decimal(out_raw))