import logging

import os

import time

from collections import namedtuple

from collections.abc import Iterable, Sequence

from typing import (

Any,

)

from eth_typing import HexStr

from eth_typing.evm import Address, ChecksumAddress

from hexbytes import HexBytes

from web3 import Web3

from web3._utils.abi import map_abi_data

from web3._utils.normalizers import BASE_RETURN_NORMALIZERS

from web3.contract import Contract

from web3.contract.contract import ContractFunction

from web3.exceptions import BadFunctionCallOutput, ContractLogicError

from web3.types import (

Nonce,

TxParams,

TxReceipt,

Wei,

)

from .constants import (

ETH_ADDRESS,

MAX_TICK,

MAX_UINT_128,

MIN_TICK,

WETH9_ADDRESS,

_factory_contract_addresses_v1,

_factory_contract_addresses_v2,

_netid_to_name,

_router_contract_addresses_v2,

_tick_bitmap_range,

_tick_spacing,

)

from .decorators import check_approval, supports

from .exceptions import InsufficientBalance, InvalidToken

from .fee import validate_fee_tier

from .token import ERC20Token

from .types import AddressLike

from .util import (

_addr_to_str,

_load_contract,

_load_contract_erc20,

_str_to_addr,

_validate_address,

chunks,

encode_sqrt_ratioX96,

is_same_address,

nearest_tick,

realised_fee_percentage,

)

logger = logging.getLogger(__name__)

class Uniswap:

"""

Wrapper around Uniswap contracts.

"""

address: AddressLike

version: int

w3: Web3

netid: int

netname: str

default_slippage: float

use_estimate_gas: bool

def __init__(

self,

address: AddressLike | str | None,

private_key: str | None,

provider: str | None = None,

web3: Web3 | None = None,

version: int = 1,

default_slippage: float = 0.01,

use_estimate_gas: bool = True,

# use_eip1559: bool = True,

factory_contract_addr: str | None = None,

router_contract_addr: str | None = None,

enable_caching: bool = False,

) -> None:

"""

:param address: The public address of the ETH wallet to use.

:param private_key: The private key of the ETH wallet to use.

:param provider: Can be optionally set to a Web3 provider URI. If none set, will fall back to the PROVIDER environment variable, or web3 if set.

:param web3: Can be optionally set to a custom Web3 instance.

:param version: Which version of the Uniswap contracts to use.

:param default_slippage: Default slippage for a trade, as a float (0.01 is 1%). WARNING: slippage is untested.

:param factory_contract_addr: Can be optionally set to override the address of the factory contract.

:param router_contract_addr: Can be optionally set to override the address of the router contract (v2 only).

:param enable_caching: Optionally enables middleware caching RPC method calls.

"""

self.address = _str_to_addr(

address or "0x0000000000000000000000000000000000000000"

)

self.private_key = (

private_key

or "0x0000000000000000000000000000000000000000000000000000000000000000"

)

self.version = version

if self.version not in [1, 2, 3]:

raise ValueError(

f"Invalid version '{self.version}', only 1, 2 or 3 supported"

) # pragma: no cover

self.default_slippage = default_slippage

self.use_estimate_gas = use_estimate_gas

if web3:

self.w3 = web3

else:

# Initialize web3. Extra provider for testing.

if not provider:

provider = os.environ["PROVIDER"]

self.w3 = Web3(Web3.HTTPProvider(provider, request_kwargs={"timeout": 60}))

self.netid = int(self.w3.net.version)

if self.netid in _netid_to_name:

self.netname = _netid_to_name[self.netid]

else:

raise ValueError(f"Unknown netid: {self.netid}") # pragma: no cover

logger.info(f"Using {self.w3} ('{self.netname}', netid: {self.netid})")

self.last_nonce: Nonce = self.w3.eth.get_transaction_count(self.address)

# This code automatically approves you for trading on the exchange.

# max_approval is to allow the contract to exchange on your behalf.

# max_approval_check checks that current approval is above a reasonable number

# The program cannot check for max_approval each time because it decreases

# with each trade.

max_approval_hex = f"0x{64 * 'f'}"

self.max_approval_int = int(max_approval_hex, 16)

max_approval_check_hex = f"0x{15 * '0'}{49 * 'f'}"

self.max_approval_check_int = int(max_approval_check_hex, 16)

if self.version == 1:

if factory_contract_addr is None:

factory_contract_addr = _factory_contract_addresses_v1[self.netname]

self.factory_contract = _load_contract(

self.w3,

abi_name="uniswap-v1/factory",

address=_str_to_addr(factory_contract_addr),

)

elif self.version == 2:

if router_contract_addr is None:

router_contract_addr = _router_contract_addresses_v2[self.netname]

self.router_address: AddressLike = _str_to_addr(router_contract_addr)

if factory_contract_addr is None:

factory_contract_addr = _factory_contract_addresses_v2[self.netname]

self.factory_contract = _load_contract(

self.w3,

abi_name="uniswap-v2/factory",

address=_str_to_addr(factory_contract_addr),

)

# Documented here: https://uniswap.org/docs/v2/smart-contracts/router02/

self.router = _load_contract(

self.w3,

abi_name="uniswap-v2/router02",

address=self.router_address,

)

elif self.version == 3:

# https://github.com/Uniswap/uniswap-v3-periphery/blob/main/deploys.md

factory_contract_address = _str_to_addr(

"0x1F98431c8aD98523631AE4a59f267346ea31F984"

)

self.factory_contract = _load_contract(

self.w3, abi_name="uniswap-v3/factory", address=factory_contract_address

)

quoter_addr = _str_to_addr("0xb27308f9F90D607463bb33eA1BeBb41C27CE5AB6")

self.router_address = _str_to_addr(

"0xE592427A0AEce92De3Edee1F18E0157C05861564"

)

self.quoter = _load_contract(

self.w3, abi_name="uniswap-v3/quoter", address=quoter_addr

)

self.router = _load_contract(

self.w3, abi_name="uniswap-v3/router", address=self.router_address

)

self.positionManager_addr = _str_to_addr(

"0xC36442b4a4522E871399CD717aBDD847Ab11FE88"

)

self.nonFungiblePositionManager = _load_contract(

self.w3,

abi_name="uniswap-v3/nonFungiblePositionManager",

address=self.positionManager_addr,

)

if self.netname == "arbitrum":

multicall2_addr = _str_to_addr(

"0x50075F151ABC5B6B448b1272A0a1cFb5CFA25828"

)

else:

multicall2_addr = _str_to_addr(

"0x5BA1e12693Dc8F9c48aAD8770482f4739bEeD696"

)

self.multicall2 = _load_contract(

self.w3, abi_name="uniswap-v3/multicall", address=multicall2_addr

)

else:

raise ValueError(

f"Invalid version '{self.version}', only 1, 2 or 3 supported"

)

if hasattr(self, "factory_contract"):

logger.info(f"Using factory contract: {self.factory_contract}")

# ------ Market --------------------------------------------------------------------

def get_price_input(

self,

token0: AddressLike, # input token

token1: AddressLike, # output token

qty: int,

fee: int | None = None,

route: list[AddressLike] | None = None,

fees: list[int] | None = None,

) -> int:

"""Given `qty` amount of the input `token0`, returns the maximum output amount of output `token1`."""

fee = validate_fee_tier(fee=fee, version=self.version)

if is_same_address(token0, ETH_ADDRESS) and self.version == 2:

return self._get_eth_token_input_price(token1, Wei(qty), fee)

elif is_same_address(token1, ETH_ADDRESS) and self.version == 2:

return self._get_token_eth_input_price(token0, qty, fee)

else:

if is_same_address(token0, ETH_ADDRESS):

token0 = self.get_weth_address()

if is_same_address(token1, ETH_ADDRESS):

token1 = self.get_weth_address()

return self._get_token_token_input_price(

token0, token1, qty, fee, route, fees

)

def get_price_output(

self,

token0: AddressLike,

token1: AddressLike,

qty: int,

fee: int | None = None,

route: list[AddressLike] | None = None,

fees: list[int] | None = None,

) -> int:

"""Returns the minimum amount of `token0` required to buy `qty` amount of `token1`."""

fee = validate_fee_tier(fee=fee, version=self.version)

if is_same_address(token0, ETH_ADDRESS) and self.version == 2:

return self._get_eth_token_output_price(token1, qty, fee)

elif is_same_address(token1, ETH_ADDRESS) and self.version == 2:

return self._get_token_eth_output_price(token0, Wei(qty), fee)

else:

if is_same_address(token0, ETH_ADDRESS):

token0 = self.get_weth_address()

if is_same_address(token1, ETH_ADDRESS):

token1 = self.get_weth_address()

return self._get_token_token_output_price(

token0, token1, qty, fee, route, fees

)

def _get_eth_token_input_price(

self,

token: AddressLike, # output token

qty: Wei,

fee: int,

) -> Wei:

"""Public price (i.e. amount of output token received) for ETH to token trades with an exact input."""

if self.version == 1:

ex = self._exchange_contract(token)

price: Wei = ex.functions.getEthToTokenInputPrice(qty).call()

elif self.version == 2:

price = self.router.functions.getAmountsOut(

qty, [self.get_weth_address(), token]

).call()[-1]

elif self.version == 3:

price = self._get_token_token_input_price(

self.get_weth_address(), token, qty, fee=fee

) # type: ignore

else:

raise ValueError # pragma: no cover

return price

def _get_token_eth_input_price(

self,

token: AddressLike, # input token

qty: int,

fee: int,

) -> int:

"""Public price (i.e. amount of ETH received) for token to ETH trades with an exact input."""

if self.version == 1:

ex = self._exchange_contract(token)

price: int = ex.functions.getTokenToEthInputPrice(qty).call()

elif self.version == 2:

price = self.router.functions.getAmountsOut(

qty, [token, self.get_weth_address()]

).call()[-1]

elif self.version == 3:

price = self._get_token_token_input_price(

token, self.get_weth_address(), qty, fee=fee

)

else:

raise ValueError # pragma: no cover

return price

def _get_token_token_input_price(

self,

token0: AddressLike, # input token

token1: AddressLike, # output token

qty: int,

fee: int,

route: list[AddressLike] | None = None,

fees: list[int] | None = None,

) -> int:

"""

Public price (i.e. amount of output token received) for token to token trades with an exact input.

:param fee: (v3 only) The pool's fee in hundredths of a bip, i.e. 1e-6 (3000 is 0.3%)

"""

if route is None and self.version == 2:

# If one of the tokens are WETH, delegate to appropriate call.

# See: https://github.com/shanefontaine/uniswap-python/issues/22

if is_same_address(token0, self.get_weth_address()):

return int(self._get_eth_token_input_price(token1, Wei(qty), fee))

elif is_same_address(token1, self.get_weth_address()):

return int(self._get_token_eth_input_price(token0, qty, fee))

route = [token0, self.get_weth_address(), token1]

logger.warning(f"No route specified, assuming route: {route}")

if self.version == 2:

price: int = self.router.functions.getAmountsOut(qty, route).call()[-1]

elif self.version == 3:

if route and not fees:

raise ValueError(

"V3 custom routes require both 'route' and 'fees' parameters"

)

if route and fees:

# NOTE: to support custom routes we need to support the Path data encoding: https://github.com/Uniswap/uniswap-v3-periphery/blob/main/contracts/libraries/Path.sol

# result: tuple = self.quoter.functions.quoteExactInput(route, qty).call()

encoded_v3_path = self.encode_v3_path(route, fees)

price = self.quoter.functions.quoteExactInput(

encoded_v3_path, qty

).call()

else:

# FIXME: How to calculate this properly? See https://docs.uniswap.org/reference/libraries/SqrtPriceMath

sqrtPriceLimitX96 = 0

price = self.quoter.functions.quoteExactInputSingle(

token0, token1, fee, qty, sqrtPriceLimitX96

).call()

else:

raise ValueError("function not supported for this version of Uniswap")

return price

def _get_eth_token_output_price(

self,

token: AddressLike, # output token

qty: int,

fee: int | None = None,

) -> Wei:

"""Public price (i.e. amount of ETH needed) for ETH to token trades with an exact output."""

fee = validate_fee_tier(fee=fee, version=self.version)

if self.version == 1:

ex = self._exchange_contract(token)

price: Wei = ex.functions.getEthToTokenOutputPrice(qty).call()

elif self.version == 2:

route = [self.get_weth_address(), token]

price = self.router.functions.getAmountsIn(qty, route).call()[0]

elif self.version == 3:

price = Wei(

self._get_token_token_output_price(

self.get_weth_address(), token, qty, fee=fee

)

)

else:

raise ValueError # pragma: no cover

return price

def _get_token_eth_output_price(

self,

token: AddressLike,

qty: Wei,

fee: int | None = None, # input token

) -> int:

"""Public price (i.e. amount of input token needed) for token to ETH trades with an exact output."""

fee = validate_fee_tier(fee=fee, version=self.version)

if self.version == 1:

ex = self._exchange_contract(token)

price: int = ex.functions.getTokenToEthOutputPrice(qty).call()

elif self.version == 2:

route = [token, self.get_weth_address()]

price = self.router.functions.getAmountsIn(qty, route).call()[0]

elif self.version == 3:

price = self._get_token_token_output_price(

token, self.get_weth_address(), qty, fee=fee

)

else:

raise ValueError # pragma: no cover

return price

@supports([2, 3])

def _get_token_token_output_price(

self,

token0: AddressLike, # input token

token1: AddressLike, # output token

qty: int,

fee: int | None = None,

route: list[AddressLike] | None = None,

fees: list[int] | None = None,

) -> int:

"""

Public price (i.e. amount of input token needed) for token to token trades with an exact output.

:param fee: (v3 only) The pool's fee in hundredths of a bip, i.e. 1e-6 (3000 is 0.3%)

"""

fee = validate_fee_tier(fee=fee, version=self.version)

if not route and self.version == 2:

# If one of the tokens are WETH, delegate to appropriate call.

# See: https://github.com/shanefontaine/uniswap-python/issues/22

if is_same_address(token0, self.get_weth_address()):

return int(self._get_eth_token_output_price(token1, qty, fee))

elif is_same_address(token1, self.get_weth_address()):

return int(self._get_token_eth_output_price(token0, Wei(qty), fee))

route = [token0, self.get_weth_address(), token1]

logger.warning(f"No route specified, assuming route: {route}")

if self.version == 2:

price: int = self.router.functions.getAmountsIn(qty, route).call()[0]

elif self.version == 3:

if route and fees:

# NOTE: to support custom routes we need to support the Path data encoding: https://github.com/Uniswap/uniswap-v3-periphery/blob/main/contracts/libraries/Path.sol

# result: tuple = self.quoter.functions.quoteExactOutput(route, qty).call()

encoded_v3_path = self.encode_v3_path(route, fees, is_exact_out=True)

price = self.quoter.functions.quoteExactOutput(

encoded_v3_path, qty

).call()

else:

# FIXME: How to calculate this properly?

# - https://docs.uniswap.org/reference/libraries/SqrtPriceMath

# - https://github.com/Uniswap/uniswap-v3-sdk/blob/main/src/swapRouter.ts

sqrtPriceLimitX96 = 0

price = self.quoter.functions.quoteExactOutputSingle(

token0, token1, fee, qty, sqrtPriceLimitX96

).call()

else:

raise ValueError # pragma: no cover

return price

# ------ Make Trade ----------------------------------------------------------------

@check_approval

def make_trade(

self,

input_token: AddressLike,

output_token: AddressLike,

qty: int | Wei,

recipient: AddressLike | None = None,

fee: int | None = None,

slippage: float | None = None,

fee_on_transfer: bool = False,

route: list[AddressLike] | None = None,

fees: list[int] | None = None,

) -> HexBytes:

"""Make a trade by defining the qty of the input token."""

if not isinstance(qty, int):

raise TypeError("swapped quantity must be an integer")

fee = validate_fee_tier(fee=fee, version=self.version)

if slippage is None:

slippage = self.default_slippage

if input_token == output_token:

raise ValueError

if self.version != 3 and fees:

fees = None

if self.version == 3 and route and not fees:

raise ValueError(

"V3 custom routes require both 'route' and 'fees' parameters"

)

if input_token == ETH_ADDRESS and (route is None or fees is None):

return self._eth_to_token_swap_input(

output_token, Wei(qty), recipient, fee, slippage, fee_on_transfer

)

elif output_token == ETH_ADDRESS and (route is None or fees is None):

return self._token_to_eth_swap_input(

input_token, qty, recipient, fee, slippage, fee_on_transfer

)

else:

# if is_same_address(input_token, ETH_ADDRESS):

# input_token = self.get_weth_address()

# if is_same_address(output_token, ETH_ADDRESS):

# output_token = self.get_weth_address()

return self._token_to_token_swap_input(

input_token,

output_token,

qty,

recipient,

fee,

slippage,

fee_on_transfer,

route,

fees,

)

@check_approval

def make_trade_output(

self,

input_token: AddressLike,

output_token: AddressLike,

qty: int | Wei,

recipient: AddressLike | None = None,

fee: int | None = None,

slippage: float | None = None,

route: list[AddressLike] | None = None,

fees: list[int] | None = None,

) -> HexBytes:

"""Make a trade by defining the qty of the output token."""

fee = validate_fee_tier(fee=fee, version=self.version)

if slippage is None:

slippage = self.default_slippage

if input_token == output_token:

raise ValueError

if self.version != 3 and fees:

fees = None

if input_token == ETH_ADDRESS and (route is None or fees is None):

balance = self.get_eth_balance()

need = self._get_eth_token_output_price(output_token, qty, fee)

if balance < need:

raise InsufficientBalance(balance, need)

return self._eth_to_token_swap_output(

output_token, qty, recipient, fee, slippage

)

elif output_token == ETH_ADDRESS and (route is None or fees is None):

return self._token_to_eth_swap_output(

input_token, Wei(qty), recipient, fee, slippage

)

else:

# if is_same_address(input_token, ETH_ADDRESS):

# input_token = self.get_weth_address()

# if is_same_address(output_token, ETH_ADDRESS):

# output_token = self.get_weth_address()

return self._token_to_token_swap_output(

input_token, output_token, qty, recipient, fee, slippage, route, fees

)

def _eth_to_token_swap_input(

self,

output_token: AddressLike,

qty: Wei,

recipient: AddressLike | None,

fee: int,

slippage: float,

fee_on_transfer: bool = False,

) -> HexBytes:

"""Convert ETH to tokens given an input amount."""

if output_token == ETH_ADDRESS:

raise ValueError

eth_balance = self.get_eth_balance()

if qty > eth_balance:

raise InsufficientBalance(eth_balance, qty)

if self.version == 1:

token_funcs = self._exchange_contract(output_token).functions

tx_params = self._get_tx_params(qty)

func_params: list[Any] = [qty, self._deadline()]

if not recipient:

function = token_funcs.ethToTokenSwapInput(*func_params)

else:

func_params.append(recipient)

function = token_funcs.ethToTokenTransferInput(*func_params)

return self._build_and_send_tx(function, tx_params)

elif self.version == 2:

if recipient is None:

recipient = self.address

amount_out_min = int(

(1 - slippage) * self._get_eth_token_input_price(output_token, qty, fee)

)

if fee_on_transfer:

func = self.router.functions.swapExactETHForTokensSupportingFeeOnTransferTokens

else:

func = self.router.functions.swapExactETHForTokens

return self._build_and_send_tx(

func(

amount_out_min,

[self.get_weth_address(), output_token],

recipient,

self._deadline(),

),

self._get_tx_params(qty),

)

elif self.version == 3:

if recipient is None:

recipient = self.address

if fee_on_transfer:

raise ValueError("fee on transfer not supported by Uniswap v3")

min_tokens_bought = int(

(1 - slippage)

* self._get_eth_token_input_price(output_token, qty, fee=fee)

)

sqrtPriceLimitX96 = 0

return self._build_and_send_tx(

self.router.functions.exactInputSingle(

{

"tokenIn": self.get_weth_address(),

"tokenOut": output_token,

"fee": fee,

"recipient": recipient,

"deadline": self._deadline(),

"amountIn": qty,

"amountOutMinimum": min_tokens_bought,

"sqrtPriceLimitX96": sqrtPriceLimitX96,

}

),

self._get_tx_params(value=qty),

)

else:

raise ValueError # pragma: no cover

def _token_to_eth_swap_input(

self,

input_token: AddressLike,

qty: int,

recipient: AddressLike | None,

fee: int,

slippage: float,

fee_on_transfer: bool = False,

) -> HexBytes:

"""Convert tokens to ETH given an input amount."""

if input_token == ETH_ADDRESS:

raise ValueError

# Balance check

input_balance = self.get_token_balance(input_token)

if qty > input_balance:

raise InsufficientBalance(input_balance, qty)

if self.version == 1:

token_funcs = self._exchange_contract(input_token).functions

func_params: list[Any] = [qty, 1, self._deadline()]

if not recipient:

function = token_funcs.tokenToEthSwapInput(*func_params)

else:

func_params.append(recipient)

function = token_funcs.tokenToEthTransferInput(*func_params)

return self._build_and_send_tx(function)

elif self.version == 2:

if recipient is None:

recipient = self.address

amount_out_min = int(

(1 - slippage) * self._get_token_eth_input_price(input_token, qty, fee)

)

if fee_on_transfer:

func = self.router.functions.swapExactTokensForETHSupportingFeeOnTransferTokens

else:

func = self.router.functions.swapExactTokensForETH

return self._build_and_send_tx(

func(

qty,

amount_out_min,

[input_token, self.get_weth_address()],

recipient,

self._deadline(),

),

)

elif self.version == 3:

if recipient is None:

recipient = self.address

if fee_on_transfer:

raise ValueError("fee on transfer not supported by Uniswap v3")

output_token = self.get_weth_address()

min_tokens_bought = int(

(1 - slippage)

* self._get_token_eth_input_price(input_token, qty, fee=fee)

)

sqrtPriceLimitX96 = 0

swap_data = self.router.encode_abi(

"exactInputSingle",

args=[

(

input_token,

output_token,

fee,

ETH_ADDRESS,

self._deadline(),

qty,

min_tokens_bought,

sqrtPriceLimitX96,

)

],

)

unwrap_data = self.router.encode_abi(

"unwrapWETH9", args=[min_tokens_bought, recipient]

)

# Multicall

return self._build_and_send_tx(

self.router.functions.multicall([swap_data, unwrap_data]),

self._get_tx_params(),

)

else:

raise ValueError # pragma: no cover

def _token_to_token_swap_input(

self,

input_token: AddressLike,

output_token: AddressLike,

qty: int,

recipient: AddressLike | None,

fee: int,

slippage: float,

fee_on_transfer: bool = False,

route: list[AddressLike] | None = None,

fees: list[int] | None = None,

) -> HexBytes:

"""Convert tokens to tokens given an input amount."""

# Balance check

input_balance = self.get_token_balance(input_token)

if qty > input_balance:

raise InsufficientBalance(input_balance, qty)

if recipient is None:

recipient = self.address

if (input_token == ETH_ADDRESS or output_token == ETH_ADDRESS) and (

route is None or fees is None

):

raise ValueError

if self.version == 1:

token_funcs = self._exchange_contract(input_token).functions

# TODO: This might not be correct

min_tokens_bought, min_eth_bought = self._calculate_max_output_token(

input_token, qty, output_token

)

func_params = [

qty,

min_tokens_bought,

min_eth_bought,

self._deadline(),

output_token,

]

if not recipient:

function = token_funcs.tokenToTokenSwapInput(*func_params)

else:

func_params.insert(len(func_params) - 1, recipient)

function = token_funcs.tokenToTokenTransferInput(*func_params)

return self._build_and_send_tx(function)

elif self.version == 2:

if fee_on_transfer:

func = self.router.functions.swapExactTokensForTokensSupportingFeeOnTransferTokens

else:

func = self.router.functions.swapExactTokensForTokens

if route is None:

weth_address = self.get_weth_address()

if is_same_address(input_token, weth_address) or is_same_address(

output_token, weth_address

):

path = [input_token, output_token]

else:

path = [input_token, weth_address, output_token]

else:

path = route

min_tokens_bought = int(

(1 - slippage)

* self._get_token_token_input_price(

input_token, output_token, qty, fee=fee, route=path, fees=fees

)

)

return self._build_and_send_tx(

func(

qty,

min_tokens_bought,

path,

recipient,

self._deadline(),

),

)

elif self.version == 3:

if fee_on_transfer:

raise ValueError("fee on transfer not supported by Uniswap v3")

min_tokens_bought = int(

(1 - slippage)

* self._get_token_token_input_price(

input_token, output_token, qty, fee=fee, route=route, fees=fees

)

)

sqrtPriceLimitX96 = 0

if route and fees:

encoded_v3_path = self.encode_v3_path(route, fees)

return self._build_and_send_tx(

self.router.functions.exactInput(

{

"path": encoded_v3_path,

"recipient": recipient,

"deadline": self._deadline(),

"amountIn": qty,

"amountOutMinimum": min_tokens_bought,

}

),

self._get_tx_params(),

)

else:

return self._build_and_send_tx(

self.router.functions.exactInputSingle(

{

"tokenIn": input_token,

"tokenOut": output_token,

"fee": fee,

"recipient": recipient,

"deadline": self._deadline(),

"amountIn": qty,

"amountOutMinimum": min_tokens_bought,

"sqrtPriceLimitX96": sqrtPriceLimitX96,

}

),

self._get_tx_params(),

)

else:

raise ValueError # pragma: no cover

def _eth_to_token_swap_output(

self,

output_token: AddressLike,

qty: int,

recipient: AddressLike | None,

fee: int,

slippage: float,

) -> HexBytes:

"""Convert ETH to tokens given an output amount."""

if output_token == ETH_ADDRESS:

raise ValueError

# Balance check

eth_balance = self.get_eth_balance()

cost = self._get_eth_token_output_price(output_token, qty, fee)

amount_in_max = Wei(int((1 + slippage) * cost))

# We check balance against amount_in_max rather than cost to be conservative

if amount_in_max > eth_balance:

raise InsufficientBalance(eth_balance, amount_in_max)

if self.version == 1:

token_funcs = self._exchange_contract(output_token).functions

eth_qty = self._get_eth_token_output_price(output_token, qty)

tx_params = self._get_tx_params(eth_qty)

func_params: list[Any] = [qty, self._deadline()]

if not recipient:

function = token_funcs.ethToTokenSwapOutput(*func_params)

else:

func_params.append(recipient)

function = token_funcs.ethToTokenTransferOutput(*func_params)

return self._build_and_send_tx(function, tx_params)

elif self.version == 2:

if recipient is None:

recipient = self.address

eth_qty = int(

(1 + slippage)

* self._get_eth_token_output_price(output_token, qty, fee)

) # type: ignore

return self._build_and_send_tx(

self.router.functions.swapETHForExactTokens(

qty,

[self.get_weth_address(), output_token],

recipient,

self._deadline(),

),

self._get_tx_params(eth_qty),

)

elif self.version == 3:

if recipient is None:

recipient = self.address

sqrtPriceLimitX96 = 0

swap_data = self.router.encode_abi(

"exactOutputSingle",

args=[

(

self.get_weth_address(),

output_token,

fee,

recipient,

self._deadline(),

qty,

amount_in_max,

sqrtPriceLimitX96,

)

],

)

refund_data = self.router.encode_abi("refundETH", args=None)

# Multicall

return self._build_and_send_tx(

self.router.functions.multicall([swap_data, refund_data]),

self._get_tx_params(value=amount_in_max),

)

else:

raise ValueError

def _token_to_eth_swap_output(

self,

input_token: AddressLike,

qty: Wei,

recipient: AddressLike | None,

fee: int,

slippage: float,

) -> HexBytes:

"""Convert tokens to ETH given an output amount."""

if input_token == ETH_ADDRESS:

raise ValueError

# Balance check

input_balance = self.get_token_balance(input_token)

cost = self._get_token_eth_output_price(input_token, qty, fee)

amount_in_max = int((1 + slippage) * cost)

# We check balance against amount_in_max rather than cost to be conservative

if amount_in_max > input_balance:

raise InsufficientBalance(input_balance, amount_in_max)

if self.version == 1:

# From https://uniswap.org/docs/v1/frontend-integration/trade-tokens/

# Is all this really necessary? Can't we just use `cost` for max_tokens?

outputAmount = qty

inputReserve = self.get_ex_token_balance(input_token)

outputReserve = self.get_ex_eth_balance(input_token)

numerator = outputAmount * inputReserve * 1000

denominator = (outputReserve - outputAmount) * 997

inputAmount = numerator / denominator + 1

max_tokens = int((1 + slippage) * inputAmount)

ex = self._exchange_contract(input_token)

func_params: list[Any] = [qty, max_tokens, self._deadline()]

if not recipient:

function = ex.functions.tokenToEthSwapOutput(*func_params)

else:

func_params.append(recipient)

function = ex.functions.tokenToEthTransferOutput(*func_params)

return self._build_and_send_tx(function)

elif self.version == 2:

if recipient is None:

recipient = self.address

max_tokens = int((1 + slippage) * cost)

return self._build_and_send_tx(

self.router.functions.swapTokensForExactETH(

qty,

max_tokens,

[input_token, self.get_weth_address()],

recipient,

self._deadline(),

),

)

elif self.version == 3:

if recipient is None:

recipient = self.address

sqrtPriceLimitX96 = 0

swap_data = self.router.encode_abi(

"exactOutputSingle",

args=[

(

input_token,

self.get_weth_address(),

fee,

ETH_ADDRESS,

self._deadline(),

qty,

amount_in_max,

sqrtPriceLimitX96,

)

],