"""Route trades to different Uniswap v2 like exchanges."""
import logging
from typing import Dict, Optional, List
from eth_typing import HexAddress
from eth_defi.tx import AssetDelta
from tradeexecutor.state.types import Percent
from tradingstrategy.chain import ChainId
from web3 import Web3
from eth_defi.gas import estimate_gas_fees
from eth_defi.uniswap_v3.deployment import UniswapV3Deployment, fetch_deployment
from eth_defi.uniswap_v3.swap import swap_with_slippage_protection
from web3.exceptions import ContractLogicError
from tradeexecutor.ethereum.tx import HotWalletTransactionBuilder
from tradeexecutor.state.identifier import TradingPairIdentifier, AssetIdentifier
from tradeexecutor.state.blockhain_transaction import BlockchainTransaction
from tradingstrategy.pair import PandasPairUniverse
from tradeexecutor.strategy.universe_model import StrategyExecutionUniverse
from tradeexecutor.ethereum.routing_state import (
EthereumRoutingState,
route_tokens, # don't remove, forwarded import
OutOfBalance, # don't remove, forwarded import
get_base_quote,
get_base_quote_intermediary
)
from tradeexecutor.ethereum.routing_model import EthereumRoutingModel
logger = logging.getLogger(__name__)
[docs]class UniswapV3RoutingState(EthereumRoutingState):
[docs] def __init__(self,
pair_universe: PandasPairUniverse,
tx_builder: Optional[HotWalletTransactionBuilder]=None,
swap_gas_limit=2_000_000):
super().__init__(pair_universe, tx_builder, swap_gas_limit)
def __repr__(self):
return f"<UniswapV3RoutingState Tx builder: {self.tx_builder} web3: {self.web3}>"
[docs] def get_uniswap_for_pair(self, address_map: dict, target_pair: TradingPairIdentifier) -> UniswapV3Deployment:
"""Get a router for a trading pair."""
return get_uniswap_for_pair(self.web3, address_map, target_pair)
[docs] def trade_on_router_two_way(self,
uniswap: UniswapV3Deployment,
target_pair: TradingPairIdentifier,
reserve_asset: AssetIdentifier,
reserve_amount: int,
max_slippage: Percent,
check_balances: False,
asset_deltas: Optional[List[AssetDelta]] = None,
notes="",
):
"""Prepare the actual swap. Same for Uniswap V2 and V3.
:param check_balances:
Check on-chain balances that the account has enough tokens
and raise exception if not.
"""
base_token, quote_token = get_base_quote(self.web3, target_pair, reserve_asset)
if check_balances:
self.check_has_enough_tokens(quote_token, reserve_amount)
raw_fee = int(target_pair.fee * 1_000_000)
logger.info(
"Creating a trade for %s, slippage tolerance %f, trade reserve %s, amount in %d",
target_pair,
max_slippage,
reserve_asset,
reserve_amount,
)
bound_swap_func = swap_with_slippage_protection(
uniswap,
recipient_address=self.tx_builder.get_token_delivery_address(),
base_token=base_token,
quote_token=quote_token,
amount_in=reserve_amount,
max_slippage=max_slippage * 10_000, # In BPS
pool_fees=[raw_fee]
)
# If the receiver (vault) has its own security policy for slippage tolerance check it here
receiver_slippage_tolerance = self.tx_builder.get_internal_slippage_tolerance()
if receiver_slippage_tolerance is not None:
# TODO: TxBuilder configures slippage tolerance as opposite of the trades
receiver_slippage_tolerance = 1 - receiver_slippage_tolerance
trade_slippage_tolerance = max_slippage
assert receiver_slippage_tolerance > trade_slippage_tolerance, f"Receiver (vault) slippage tolerance tighter than the trade slippage tolerance.\nReceiver: {receiver_slippage_tolerance * 10_000} BPS, trade: {trade_slippage_tolerance * 10_000} BPS"
return self.create_signed_transaction(
uniswap.swap_router,
bound_swap_func,
self.swap_gas_limit,
asset_deltas,
notes=notes,
)
[docs] def trade_on_router_three_way(self,
uniswap: UniswapV3Deployment,
target_pair: TradingPairIdentifier,
intermediary_pair: TradingPairIdentifier,
reserve_asset: AssetIdentifier,
reserve_amount: int,
max_slippage: float,
check_balances: False,
asset_deltas: Optional[List[AssetDelta]] = None,
notes="",
):
"""Prepare the actual swap for three way trade.
:param check_balances:
Check on-chain balances that the account has enough tokens
and raise exception if not.
"""
self.validate_pairs(target_pair, intermediary_pair)
self.validate_exchange(target_pair, intermediary_pair)
base_token, quote_token, intermediary_token = get_base_quote_intermediary(self.web3,target_pair, intermediary_pair, reserve_asset)
if check_balances:
self.check_has_enough_tokens(quote_token, reserve_amount)
# eth_defi uses raw_fees
raw_pool_fees = [int(intermediary_pair.fee * 1_000_000), int(target_pair.fee * 1_000_000)]
bound_swap_func = swap_with_slippage_protection(
uniswap,
recipient_address=self.tx_builder.get_token_delivery_address(),
base_token=base_token,
quote_token=quote_token,
pool_fees=raw_pool_fees,
amount_in=reserve_amount,
max_slippage=max_slippage * 100, # In BPS,
intermediate_token=intermediary_token,
)
return self.create_signed_transaction(
uniswap.swap_router,
bound_swap_func,
self.swap_gas_limit,
asset_deltas,
notes=notes,
)
[docs]class UniswapV3SimpleRoutingModel(EthereumRoutingModel):
"""A simple router that does not optimise the trade execution cost. Designed for uniswap-v2 forks.
- Able to trade on multiple exchanges
- Able to three-way trades through predefined intermediary hops,
either on the exchange itself or some outside exchange
"""
[docs] def __init__(self,
address_map: Dict[str, HexAddress],
allowed_intermediary_pairs: Dict[str, str],
reserve_token_address: str,
chain_id: Optional[ChainId] = None,
):
"""
:param address_map:
Defines router smart contracts to be used with each DEX.
Address map is a dict of factory, router, position_manager,
and quoter addresses
:param allowed_intermediary_pairs:
Quote token address -> pair smart contract address mapping.
Because we hold our reserves only in one currecy e.g. BUSD
and we want to trade e.g. Cake/BNB pairs, we need to whitelist
BNB as an allowed intermediary token.
This makes it possible to do BUSD -> BNB -> Cake trade.
This set is the list of pair smart contract addresses that
are allowed to be used as a hop.
:param chain_id:
Store the chain id for which these routes were generated for.
:param reserve_token_address:
Token address of our reserve currency.
Relevent for buy/sell routing.
Lowercase.
"""
super().__init__(allowed_intermediary_pairs, reserve_token_address, chain_id)
assert type(address_map) == dict
self.address_map = self.convert_address_dict_to_lower(address_map)
[docs] def create_routing_state(self,
universe: StrategyExecutionUniverse,
execution_details: dict) -> UniswapV3RoutingState:
"""Create a new routing state for this cycle.
- Connect routing to web3 and hot wallet
- Read on-chain data on what gas fee we are going to use
- Setup transaction builder based on this information
"""
return super().create_routing_state(universe, execution_details, UniswapV3RoutingState)
[docs] def make_direct_trade(
self,
routing_state: EthereumRoutingState,
target_pair: TradingPairIdentifier,
reserve_asset: AssetIdentifier,
reserve_amount: int,
max_slippage: float,
check_balances=False,
asset_deltas: Optional[List[AssetDelta]] = None,
notes="",
) -> list[BlockchainTransaction]:
return super().make_direct_trade(
routing_state,
target_pair,
reserve_asset,
reserve_amount,
max_slippage,
self.address_map,
check_balances,
asset_deltas=asset_deltas,
notes="",
)
[docs] def make_multihop_trade(
self,
routing_state: EthereumRoutingState,
target_pair: TradingPairIdentifier,
intermediary_pair: TradingPairIdentifier,
reserve_asset: AssetIdentifier,
reserve_amount: int,
max_slippage: float,
check_balances=False,
asset_deltas: Optional[List[AssetDelta]] = None,
notes="",
) -> list[BlockchainTransaction]:
return super().make_multihop_trade(
routing_state,
target_pair,
intermediary_pair,
reserve_asset,
reserve_amount,
max_slippage,
self.address_map,
check_balances,
asset_deltas=asset_deltas,
notes="",
)
[docs]def get_uniswap_for_pair(web3: Web3, address_map: dict, target_pair: TradingPairIdentifier) -> UniswapV3Deployment:
"""Get a router for a trading pair."""
assert target_pair.exchange_address, f"Exchange address missing for {target_pair}"
factory_address = Web3.to_checksum_address(target_pair.exchange_address)
assert factory_address == Web3.to_checksum_address(address_map["factory"]), "address_map[\"factory\"] and target_pair.exchange_address should be equal"
router_address = Web3.to_checksum_address(address_map["router"])
position_manager_address = Web3.to_checksum_address(address_map["position_manager"])
quoter_address = Web3.to_checksum_address(address_map["quoter"])
try:
return fetch_deployment(
web3,
factory_address,
router_address,
position_manager_address,
quoter_address
)
except ContractLogicError as e:
raise RuntimeError(f"Could not fetch deployment data for router address {router_address} (factory {factory_address}) - data is likely wrong") from e