diff --git a/README.md b/README.md index 19c8726..900317a 100644 --- a/README.md +++ b/README.md @@ -32,178 +32,10 @@ Documentation is available at https://uniswap-python.com/ ## Getting Started -This README is documentation on the syntax of the python client presented in this repository. See function docstrings for full syntax details. - -This library attempts to present a clean interface to Uniswap, but in order to use it to its full potential, you must familiarize yourself with the official Uniswap documentation. - -* https://docs.uniswap.io/ - -You may manually install the project or use pip: - -```python -pip install uniswap-python - -# or - -pip install git+git://github.com/shanefontaine/uniswap-python.git -``` - -### Environment Variables -The program expects an environment variables to be set in order to run the program. You can use an Infura node, since the transactions are being signed locally and broadcast as a raw transaction. The environment variable is: - -``` -PROVIDER # HTTP Provider for web3 -``` - -### Gas pricing - -To modify the gas pricing strategy you need to pass a custom `Web3` instance to the `Uniswap` constructor. You can find details for how to configure Web3 gas strategies in their [documentation](https://web3py.readthedocs.io/en/stable/gas_price.html). - -### Examples and API - -**Note:** These examples are in the process of being moved to the [docs](https://shanefontaine.github.io/uniswap-python/). - -The `Uniswap` class takes several optional parameters, read the code to see which ones are available. - -```python -from uniswap import Uniswap -address = "YOUR ADDRESS" # or "0x0000000000000000000000000000000000000000", if you're not making transactions -private_key = "YOUR PRIVATE KEY" # or None, if you're not going to make transactions -uniswap_wrapper = Uniswap(address, private_key, version=2) # pass version=2 to use Uniswap v2 -eth = "0x0000000000000000000000000000000000000000" -bat = "0x0D8775F648430679A709E98d2b0Cb6250d2887EF" -dai = "0x89d24A6b4CcB1B6fAA2625fE562bDD9a23260359" -``` - -#### Price Methods - -> **Note:** These methods assume a certain route for the swap to take, which may not be the optimal route. See https://github.com/shanefontaine/uniswap-python/issues/69 for details. - -These methods return the price as an integer in the smallest unit of the token. You need to ensure that you know how many decimals the token you're trying to trade uses to get prices in the common decimal format. See https://github.com/shanefontaine/uniswap-python/issues/12 for details. - -Decimals for common tokens: - - - ETH, DAI, and BAT uses 18 decimals (as you can see in code below) - - WBTC uses 8 decimals - - USDC and USDT uses 6 decimals - -You can look up the number of decimals used by a particular token by looking up the contract on Etherscan. As an example, here is the one for WBTC: https://etherscan.io/token/0x2260fac5e5542a773aa44fbcfedf7c193bc2c599 - -* [get_eth_token_input_price](https://github.com/Uniswap/contracts-vyper/blob/master/contracts/uniswap_exchange.vy#L416) -```python -# Get the public price for ETH to Token trades with an exact input. -uniswap_wrapper.get_eth_token_input_price(bat, 1*10**18) -uniswap_wrapper.get_eth_token_input_price(dai, 5*10**18) -``` - -* [get_token_eth_input_price](https://github.com/Uniswap/contracts-vyper/blob/master/contracts/uniswap_exchange.vy#L437) -```python -# Get the public price for token to ETH trades with an exact input. -uniswap_wrapper.get_token_eth_input_price(bat, 1*10**18) -uniswap_wrapper.get_token_eth_input_price(dai, 5*10**18) -``` - -* [get_eth_token_output_price](https://github.com/Uniswap/contracts-vyper/blob/master/contracts/uniswap_exchange.vy#L426) -```python -# Get the public price for ETH to Token trades with an exact output -uniswap_wrapper.get_eth_token_output_price(bat, 1*10**18) -uniswap_wrapper.get_eth_token_output_price(dai, 5*10**18) -``` - -* [get_token_eth_output_price](https://github.com/Uniswap/contracts-vyper/blob/master/contracts/uniswap_exchange.vy#L448) -```python -# Get the public price for token to ETH trades with an exact output. -uniswap_wrapper.get_token_eth_output_price(bat, 1*10**18) -uniswap_wrapper.get_token_eth_output_price(dai, 5*10**18) -``` - -#### Trading - -> **Note:** The same route assumptions and need for handling decimals apply here as those mentioned in the previous section. - -* make_trade - * eth_to_token_input - * [ethToTokenSwapInput](https://github.com/Uniswap/contracts-vyper/blob/master/contracts/uniswap_exchange.vy#L127) - * [ethToTokenTransferInput](https://github.com/Uniswap/contracts-vyper/blob/master/contracts/uniswap_exchange.vy#L162) - * token_to_eth_input - * [tokenToEthSwapInput](https://github.com/Uniswap/contracts-vyper/blob/master/contracts/uniswap_exchange.vy#L202) - * [tokenToEthTransferInput](https://github.com/Uniswap/contracts-vyper/blob/master/contracts/uniswap_exchange.vy#L232) - * token_to_token_input - * [tokenToTokenSwapInput](https://github.com/Uniswap/contracts-vyper/blob/master/contracts/uniswap_exchange.vy#L271) - * [tokenToTokenTransferInput](https://github.com/Uniswap/contracts-vyper/blob/master/contracts/uniswap_exchange.vy#L307) - -```python -# Make a trade based on the input parameters -uniswap_wrapper.make_trade(eth, bat, 1*10**18) # calls _eth_to_token_input -uniswap_wrapper.make_trade(bat, eth, 1*10**18) # calls _token_to_eth_input -uniswap_wrapper.make_trade(bat, dai, 1*10**18) # calls _token_to_token_input -uniswap_wrapper.make_trade(eth, bat, 1*10**18, "0x123...") # calls _eth_to_token_input -``` - -* make_trade_output - * eth_to_token_swap_output - * [ethToTokenSwapOutput](https://github.com/Uniswap/contracts-vyper/blob/master/contracts/uniswap_exchange.vy#L167) - * [ethToTokenTransferOutput](https://github.com/Uniswap/contracts-vyper/blob/master/contracts/uniswap_exchange.vy#L197) - * token_to_eth_swap_output - * [tokenToEthSwapOutput](https://github.com/Uniswap/contracts-vyper/blob/master/contracts/uniswap_exchange.vy#L237) - * [tokenToEthTransferOutput](https://github.com/Uniswap/contracts-vyper/blob/master/contracts/uniswap_exchange.vy#L266) - * token_to_token_swap_output - * [tokenToTokenSwapOutput](https://github.com/Uniswap/contracts-vyper/blob/master/contracts/uniswap_exchange.vy#L312) - * [tokenToTokenTransferOutput](https://github.com/Uniswap/contracts-vyper/blob/master/contracts/uniswap_exchange.vy#L349)) - -```python -# Make a trade where the output qty is known based on the input parameters -uniswap_wrapper.make_trade_output(eth, bat, 1*10**18) # calls _eth_to_token_swap_output -uniswap_wrapper.make_trade_output(bat, eth, 1*10**18) # calls _token_to_eth_swap_output -uniswap_wrapper.make_trade_output(bat, dai, 1*10**18, "0x123...") # calls _token_to_token_swap_output -``` - -#### Fee Methods -* [get_fee_maker](https://docs.uniswap.io/) -```python -uniswap_wrapper.get_fee_maker() -``` - -* [get_fee_taker](https://docs.uniswap.io/) -```python -uniswap_wrapper.get_fee_taker() -``` - - -#### ERC20 Pool Methods (v1 only) -* [get_ex_eth_balance](https://docs.uniswap.io/smart-contract-integration/vyper) -```python -# Get the balance of ETH in an exchange contract. -uniswap_wrapper.get_ex_eth_balance(bat) -``` - -* [get_ex_token_balance](https://github.com/Uniswap/contracts-vyper/blob/master/contracts/uniswap_exchange.vy#L469) -```python -# Get the balance of a token in an exchange contract. -uniswap_wrapper.get_ex_token_balance(bat) -``` - -* [get_exchange_rate](https://github.com/Uniswap/uniswap-frontend/blob/master/src/pages/Pool/AddLiquidity.js#L351) -```python -# Get the exchange rate of token/ETH -uniswap_wrapper.get_exchange_rate(bat) -``` - -#### Liquidity Methods (v1 only) - -* [add_liquidity](https://github.com/Uniswap/contracts-vyper/blob/master/contracts/uniswap_exchange.vy#L48) -```python -# Add liquidity to the pool. -uniswap_wrapper.add_liquidity(bat, 1*10**18) -``` - -* [remove_liquidity](https://github.com/Uniswap/contracts-vyper/blob/master/contracts/uniswap_exchange.vy#L83) -```python -# Remove liquidity from the pool. -uniswap_wrapper.remove_liquidity(bat, 1*10**18) -``` +See our [Getting started guide](https://uniswap-python.com/getting-started.html) in the documentation. ## Testing + Unit tests are under development using the pytest framework. Contributions are welcome! Test are run on a fork of the main net using ganache-cli. You need to install it with `npm install -g ganache-cli` before running tests. diff --git a/docs/api.rst b/docs/api.rst index 6da6b49..5a0114f 100644 --- a/docs/api.rst +++ b/docs/api.rst @@ -9,9 +9,23 @@ Uniswap class .. autoclass:: Uniswap :members: +Token class +----------- + +.. automodule:: uniswap.token + +.. autoclass:: BaseToken + :members: + +.. autoclass:: ERC20Token + :inherited-members: + :members: + Exceptions ---------- +.. automodule:: uniswap.exceptions + .. autoexception:: InvalidToken .. autoexception:: InsufficientBalance diff --git a/docs/conf.py b/docs/conf.py index cf0df88..cd0d5d6 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -28,7 +28,7 @@ # Add any Sphinx extension module names here, as strings. They can be # extensions coming with Sphinx (named 'sphinx.ext.*') or your custom # ones. -extensions = ["sphinx.ext.autodoc", "sphinx_click"] +extensions = ["sphinx.ext.autodoc", "sphinx.ext.extlinks", "sphinx_click"] # Add any paths that contain templates here, relative to this directory. templates_path = ["_templates"] @@ -38,6 +38,10 @@ # This pattern also affects html_static_path and html_extra_path. exclude_patterns = ["_build", "Thumbs.db", ".DS_Store"] +extlinks = { + "issue": ("https://github.com/shanefontaine/uniswap-python/issues/%s", "issue #"), +} + # -- Options for HTML output ------------------------------------------------- @@ -60,6 +64,10 @@ "path_to_docs": "docs", "use_repository_button": True, "use_edit_page_button": True, + "extra_navbar": """ +

+ Back to GitHub +

""", } show_navbar_depth = 2 diff --git a/docs/getting-started.rst b/docs/getting-started.rst new file mode 100644 index 0000000..6d558af --- /dev/null +++ b/docs/getting-started.rst @@ -0,0 +1,162 @@ +Getting started +=============== + +This library attempts to present a clean interface to Uniswap, but in order to use it to its full potential, you must familiarize yourself with the official Uniswap documentation: + +- V1: https://uniswap.org/docs/v1/ +- V2: https://uniswap.org/docs/v2/ +- V3: https://docs.uniswap.org/ + +.. contents:: Table of contents + :local: + :depth: 3 + +Installation +------------ + +You can install the latest release from PyPI, or install the latest commit directly from git: + +.. code:: sh + + pip install uniswap-python + + # or + + pip install git+git://github.com/shanefontaine/uniswap-python.git + + +Initializing the Uniswap class +------------------------------ + +If you want to trade you need to provide your address and private key. If not, you can set them to ``None``. + +In addition, the :class:`~uniswap.Uniswap` class takes several optional parameters, as documented in the `API Reference`. + +.. code:: python + + from uniswap import Uniswap + + address = "YOUR ADDRESS" # or None if you're not going to make transactions + private_key = "YOUR PRIVATE KEY" # or None if you're not going to make transactions + version = 2 # specify which version of Uniswap to use + provider = "WEB3 PROVIDER URL" # can also be set through the environment variable `PROVIDER` + uniswap = Uniswap(address=address, private_key=private_key, version=version, provider=provider) + + # Some token addresses we'll be using later in this guide + eth = "0x0000000000000000000000000000000000000000" + bat = "0x0D8775F648430679A709E98d2b0Cb6250d2887EF" + dai = "0x89d24A6b4CcB1B6fAA2625fE562bDD9a23260359" + + +Environment Variables +````````````````````` + +The program expects an environment variables to be set in order to run the program. You can use an Infura node, since the transactions are being signed locally and broadcast as a raw transaction. The environment variable is: + +.. code:: sh + + PROVIDER # HTTP Provider for web3 + +Gas pricing +``````````` + +To modify the gas pricing strategy you need to pass a custom `Web3` instance to the :class:`~uniswap.Uniswap` constructor. You can find details for how to configure Web3 gas strategies in their `documentation `_. + + +Quoting prices +-------------- + +.. note:: + + These methods assume a certain route for the swap to take, which may not be the optimal route. See :issue:`69` for details. + +There are two functions to retrieve the price for a given pair, one for specifying how much you get given a certain amount of the input token, and another for specifying how much you need to pay to receive a certain amount of the output token. + +:func:`~uniswap.Uniswap.get_price_input` +```````````````````````````````````````` + +Returns the cost of the given number of input tokens, priced in the output token. + +.. code:: python + + # Returns the amount of DAI you get for 1 ETH (10^18 wei) + uniswap.get_price_input(eth, dai, 10**18) + +:func:`~uniswap.Uniswap.get_price_output` +````````````````````````````````````````` + +Returns the amount of input token you need for the given amount of output tokens. + +.. code:: python + + # Returns the amount of ETH you need to pay (in wei) to get 1000 DAI + uniswap.get_price_output(eth, dai, 1_000 * 10**18) + + +.. note:: + + These methods return the price as an integer in the smallest unit of the token. You need to ensure that you know how many decimals the token you're trying to trade uses to get prices in the common decimal format. See :issue:`12` for details. + + Decimals for common tokens: + + - ETH, DAI, and BAT uses 18 decimals (as you can see in code below) + - WBTC uses 8 decimals + - USDC and USDT uses 6 decimals + + You can look up the number of decimals used by a particular token by looking up the contract on Etherscan. + +Making trades +------------- + +.. note:: + + The same route assumptions and need for handling decimals apply here as those mentioned in the previous section. + +:func:`~uniswap.Uniswap.make_trade` +``````````````````````````````````` + +.. code:: python + + # Make a trade based on the input parameters + uniswap.make_trade(eth, bat, 1*10**18) + uniswap.make_trade(bat, eth, 1*10**18) + uniswap.make_trade(bat, dai, 1*10**18) + uniswap.make_trade(eth, bat, 1*10**18, "0x123...") + +:func:`~uniswap.Uniswap.make_trade_output` +`````````````````````````````````````````` + +.. code:: python + + # Make a trade where the output qty isearch)Lknown based on the input parameters + uniswap.make_trade_output(eth, bat, 1*10**18) # calls _eth_to_token_swap_output + uniswap.make_trade_output(bat, eth, 1*10**18) # calls _token_to_eth_swap_output + uniswap.make_trade_output(bat, dai, 1*10**18, "0x123...") # calls _token_to_token_swap_output + + +Pool Methods (v1 only) +--------------------------- + +.. code:: python + + # Get the balance of ETH in an exchange contract. + uniswap.get_ex_eth_balance(bat) + + # Get the balance of a token in an exchange contract. + uniswap.get_ex_token_balance(bat) + + # Get the exchange rate of token/ETH + uniswap.get_exchange_rate(bat) + + +Liquidity Methods (v1 only) +--------------------------- + +.. code:: python + + # Add liquidity to the pool. + uniswap.add_liquidity(bat, 1*10**18) + + # Remove liquidity from the pool. + uniswap.remove_liquidity(bat, 1*10**18) + diff --git a/docs/index.rst b/docs/index.rst index 11bc946..19697e9 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -15,6 +15,7 @@ Welcome to uniswap-python's documentation! :maxdepth: 2 :caption: Contents: + getting-started api cli examples diff --git a/tests/test_cli.py b/tests/test_cli.py index 2490e54..e62adf2 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -39,10 +39,6 @@ def test_get_token(): print_result(result) assert result.exit_code == 0 - out = json.loads(result.stdout.replace("'", '"')) - assert out["symbol"] == "WETH" - assert out["decimals"] == 18 - def test_get_tokendb(): runner = CliRunner(mix_stderr=False) diff --git a/tests/test_uniswap.py b/tests/test_uniswap.py index ee70f60..4220439 100644 --- a/tests/test_uniswap.py +++ b/tests/test_uniswap.py @@ -10,7 +10,9 @@ from web3 import Web3 -from uniswap import Uniswap, InvalidToken, InsufficientBalance +from uniswap import Uniswap +from uniswap.constants import ETH_ADDRESS +from uniswap.exceptions import InvalidToken, InsufficientBalance logger = logging.getLogger(__name__) @@ -45,7 +47,7 @@ def test_assets(client: Uniswap): for token_name, amount in [("DAI", 100 * 10 ** 18), ("USDC", 100 * 10 ** 6)]: token_addr = tokens[token_name] - price = client.get_eth_token_output_price(token_addr, amount) + price = client.get_price_output(ETH_ADDRESS, token_addr, amount) logger.info(f"Cost of {amount} {token_name}: {price}") logger.info("Buying...") @@ -92,8 +94,8 @@ def does_not_raise(): # TODO: Change pytest.param(..., mark=pytest.mark.xfail) to the expectation/raises method @pytest.mark.usefixtures("client", "web3") class TestUniswap(object): - ONE_WEI = 1 - ONE_ETH = 10 ** 18 * ONE_WEI + ONE_ETH = 10 ** 18 + ONE_USDC = 10 ** 6 ZERO_ADDRESS = "0x0000000000000000000000000000000000000000" @@ -124,90 +126,48 @@ def test_get_fee_taker(self, client: Uniswap): assert r == 0.003 # ------ Market -------------------------------------------------------------------- - @pytest.mark.parametrize( - "token, qty", - [ - (bat, ONE_ETH), - (dai, ONE_ETH), - (bat, 2 * ONE_ETH), - pytest.param("btc", ONE_ETH, marks=pytest.mark.xfail), - ], - ) - def test_get_eth_token_input_price(self, client, token, qty): - r = client.get_eth_token_input_price(token, qty) - assert r - - @pytest.mark.parametrize( - "token, qty", - [ - (bat, ONE_ETH), - (dai, ONE_ETH), - (bat, 2 * ONE_ETH), - pytest.param("btc", ONE_ETH, marks=pytest.mark.xfail), - ], - ) - def test_get_token_eth_input_price(self, client, token, qty): - r = client.get_token_eth_input_price(token, qty) - assert r - @pytest.mark.parametrize( "token0, token1, qty, kwargs", [ - # BAT/DAI has no liquidity in V3 - # (bat, dai, ONE_ETH), - # (dai, bat, ONE_ETH), - # (bat, dai, 2 * ONE_ETH), - (dai, usdc, ONE_ETH, {"fee": 500}), + (eth, bat, ONE_ETH, {}), + (bat, eth, ONE_ETH, {}), + (eth, dai, ONE_ETH, {}), + (dai, eth, ONE_ETH, {}), + (eth, bat, 2 * ONE_ETH, {}), + (bat, eth, 2 * ONE_ETH, {}), (weth, dai, ONE_ETH, {}), (dai, weth, ONE_ETH, {}), + (dai, usdc, ONE_ETH, {"fee": 500}), + pytest.param(eth, "btc", ONE_ETH, {}, marks=pytest.mark.xfail), + pytest.param("btc", eth, ONE_ETH, {}, marks=pytest.mark.xfail), ], ) - def test_get_token_token_input_price(self, client, token0, token1, qty, kwargs): - if client.version not in [2, 3]: - pytest.skip("Tested method not supported in this Uniswap version") - r = client.get_token_token_input_price(token0, token1, qty, **kwargs) - assert r - - @pytest.mark.parametrize( - "token, qty", - [ - (bat, ONE_ETH), - (dai, ONE_ETH), - (bat, 2 * ONE_ETH), - pytest.param("btc", ONE_ETH, marks=pytest.mark.xfail), - ], - ) - def test_get_eth_token_output_price(self, client, token, qty): - r = client.get_eth_token_output_price(token, qty) - assert r - - @pytest.mark.parametrize( - "token, qty", - [ - (bat, ONE_ETH), - (dai, ONE_ETH), - (bat, 2 * ONE_ETH), - pytest.param("btc", ONE_ETH, marks=pytest.mark.xfail), - ], - ) - def test_get_token_eth_output_price(self, client, token, qty): - r = client.get_token_eth_output_price(token, qty) + def test_get_price_input(self, client, token0, token1, qty, kwargs): + if client.version == 1 and ETH_ADDRESS not in [token0, token1]: + pytest.skip("Not supported in this version of Uniswap") + r = client.get_price_input(token0, token1, qty, **kwargs) assert r @pytest.mark.parametrize( "token0, token1, qty, kwargs", [ - # (bat, dai, ONE_ETH), - (dai, usdc, ONE_ETH, {"fee": 500}), - # (bat, dai, 2 * ONE_ETH), + (eth, bat, ONE_ETH, {}), + (bat, eth, ONE_ETH, {}), + (eth, dai, ONE_ETH, {}), + (dai, eth, ONE_ETH, {}), + (eth, bat, 2 * ONE_ETH, {}), + (bat, eth, 2 * ONE_ETH, {}), (weth, dai, ONE_ETH, {}), (dai, weth, ONE_ETH, {}), + (dai, usdc, ONE_USDC, {"fee": 500}), + pytest.param(eth, "btc", ONE_ETH, {}, marks=pytest.mark.xfail), + pytest.param("btc", eth, ONE_ETH, {}, marks=pytest.mark.xfail), ], ) - def test_get_token_token_output_price(self, client, token0, token1, qty, kwargs): - if client.version not in [2, 3]: - pytest.skip("Tested method not supported in this Uniswap version") - r = client.get_token_token_output_price(token0, token1, qty, **kwargs) + def test_get_price_output(self, client, token0, token1, qty, kwargs): + if client.version == 1 and ETH_ADDRESS not in [token0, token1]: + pytest.skip("Not supported in this version of Uniswap") + r = client.get_price_output(token0, token1, qty, **kwargs) assert r # ------ ERC20 Pool ---------------------------------------------------------------- @@ -273,11 +233,11 @@ def test_remove_liquidity( "input_token, output_token, qty, recipient, expectation", [ # ETH -> Token - (eth, dai, 1_000_000_000 * ONE_WEI, None, does_not_raise), + (eth, dai, ONE_ETH, None, does_not_raise), # Token -> Token - (dai, usdc, 1_000_000_000 * ONE_WEI, None, does_not_raise), + (dai, usdc, ONE_ETH, None, does_not_raise), # Token -> ETH - (usdc, eth, 1_000_000 * ONE_WEI, None, does_not_raise), + (usdc, eth, 100 * ONE_USDC, None, does_not_raise), # (eth, bat, 0.00001 * ONE_ETH, ZERO_ADDRESS, does_not_raise), # (bat, eth, 0.00001 * ONE_ETH, ZERO_ADDRESS, does_not_raise), # (dai, bat, 0.00001 * ONE_ETH, ZERO_ADDRESS, does_not_raise), @@ -295,6 +255,10 @@ def test_make_trade( recipient, expectation, ): + if client.version == 1 and ETH_ADDRESS not in [input_token, output_token]: + pytest.skip( + "Not supported in this version of Uniswap, or at least no liquidity" + ) with expectation(): bal_in_before = client.get_token_balance(input_token) @@ -311,11 +275,11 @@ def test_make_trade( "input_token, output_token, qty, recipient, expectation", [ # ETH -> Token - (eth, dai, 1_000_000 * ONE_WEI, None, does_not_raise), + (eth, dai, 10 ** 18, None, does_not_raise), # Token -> Token - (dai, usdc, 1_000_000 * ONE_WEI, None, does_not_raise), + (dai, usdc, ONE_USDC, None, does_not_raise), # Token -> ETH - (dai, eth, 1_000_000 * ONE_WEI, None, does_not_raise), + (dai, eth, 10 ** 16, None, does_not_raise), # FIXME: These should probably be uncommented eventually # (eth, bat, int(0.000001 * ONE_ETH), ZERO_ADDRESS), # (bat, eth, int(0.000001 * ONE_ETH), ZERO_ADDRESS), @@ -323,7 +287,7 @@ def test_make_trade( ( dai, eth, - 10 ** 18 * ONE_WEI, + 1000 * 10 ** 18, None, lambda: pytest.raises(InsufficientBalance), ), diff --git a/uniswap/__init__.py b/uniswap/__init__.py index 7dc75c5..53d0a5b 100644 --- a/uniswap/__init__.py +++ b/uniswap/__init__.py @@ -1,2 +1,3 @@ -from .uniswap import Uniswap, InvalidToken, InsufficientBalance, _str_to_addr +from . import exceptions +from .uniswap import Uniswap, _str_to_addr from .cli import main diff --git a/uniswap/cli.py b/uniswap/cli.py index 47d13c4..f68b9b0 100644 --- a/uniswap/cli.py +++ b/uniswap/cli.py @@ -6,6 +6,7 @@ from web3 import Web3 from .uniswap import Uniswap, AddressLike, _str_to_addr +from .token import BaseToken from .tokens import tokens @@ -65,14 +66,14 @@ def price( quantity: int = None, ) -> None: """Returns the price of ``quantity`` tokens of ``token_in`` quoted in ``token_out``.""" - uni = ctx.obj["UNISWAP"] + uni: Uniswap = ctx.obj["UNISWAP"] if quantity is None: - quantity = 10 ** uni.get_token(token_in)["decimals"] - price = uni.get_token_token_input_price(token_in, token_out, qty=quantity) + quantity = 10 ** uni.get_token(token_in).decimals + price = uni.get_price_input(token_in, token_out, qty=quantity) if raw: print(price) else: - decimals = uni.get_token(token_out)["decimals"] + decimals = uni.get_token(token_out).decimals print(price / 10 ** decimals) @@ -81,7 +82,7 @@ def price( @click.pass_context def token(ctx: click.Context, token: AddressLike) -> None: """Show metadata for token""" - uni = ctx.obj["UNISWAP"] + uni: Uniswap = ctx.obj["UNISWAP"] t1 = uni.get_token(token) print(t1) @@ -91,12 +92,11 @@ def token(ctx: click.Context, token: AddressLike) -> None: @click.pass_context def tokendb(ctx: click.Context, metadata: bool) -> None: """List known token addresses""" - uni = ctx.obj["UNISWAP"] + uni: Uniswap = ctx.obj["UNISWAP"] for symbol, addr in tokens.items(): if metadata and addr != "0x0000000000000000000000000000000000000000": data = uni.get_token(_str_to_addr(addr)) - data["address"] = addr - assert data["symbol"].lower() == symbol.lower() + assert data.symbol.lower() == symbol.lower() print(data) else: - print({"symbol": symbol, "address": addr}) + print(BaseToken(symbol, addr)) diff --git a/uniswap/constants.py b/uniswap/constants.py new file mode 100644 index 0000000..7092685 --- /dev/null +++ b/uniswap/constants.py @@ -0,0 +1,39 @@ +ETH_ADDRESS = "0x0000000000000000000000000000000000000000" +WETH9_ADDRESS = "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2" + +# see: https://chainid.network/chains/ +_netid_to_name = { + 1: "mainnet", + 3: "ropsten", + 4: "rinkeby", + 56: "binance", + 97: "binance_testnet", + 100: "xdai", +} + +_factory_contract_addresses_v1 = { + "mainnet": "0xc0a47dFe034B400B47bDaD5FecDa2621de6c4d95", + "ropsten": "0x9c83dCE8CA20E9aAF9D3efc003b2ea62aBC08351", + "rinkeby": "0xf5D915570BC477f9B8D6C0E980aA81757A3AaC36", + "kovan": "0xD3E51Ef092B2845f10401a0159B2B96e8B6c3D30", + "görli": "0x6Ce570d02D73d4c384b46135E87f8C592A8c86dA", +} + + +# For v2 the address is the same on mainnet, Ropsten, Rinkeby, Görli, and Kovan +# https://uniswap.org/docs/v2/smart-contracts/factory +_factory_contract_addresses_v2 = { + "mainnet": "0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f", + "ropsten": "0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f", + "rinkeby": "0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f", + "görli": "0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f", + "xdai": "0xA818b4F111Ccac7AA31D0BCc0806d64F2E0737D7", +} + +_router_contract_addresses_v2 = { + "mainnet": "0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D", + "ropsten": "0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D", + "rinkeby": "0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D", + "görli": "0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D", + "xdai": "0x1C232F01118CB8B424793ae03F870aa7D0ac7f77", +} diff --git a/uniswap/decorators.py b/uniswap/decorators.py new file mode 100644 index 0000000..d0f0ed3 --- /dev/null +++ b/uniswap/decorators.py @@ -0,0 +1,60 @@ +import functools +from typing import Callable, Any, List, Dict, TYPE_CHECKING + +from .constants import ETH_ADDRESS + +if TYPE_CHECKING: + from .uniswap import Uniswap + + +def check_approval(method: Callable) -> Callable: + """Decorator to check if user is approved for a token. It approves them if they + need to be approved.""" + + @functools.wraps(method) + def approved(self: Any, *args: Any, **kwargs: Any) -> Any: + # Check to see if the first token is actually ETH + token = args[0] if args[0] != ETH_ADDRESS else None + token_two = None + + # Check second token, if needed + if method.__name__ == "make_trade" or method.__name__ == "make_trade_output": + token_two = args[1] if args[1] != ETH_ADDRESS else None + + # Approve both tokens, if needed + if token: + is_approved = self._is_approved(token) + # logger.warning(f"Approved? {token}: {is_approved}") + if not is_approved: + self.approve(token) + if token_two: + is_approved = self._is_approved(token_two) + # logger.warning(f"Approved? {token_two}: {is_approved}") + if not is_approved: + self.approve(token_two) + return method(self, *args, **kwargs) + + return approved + + +def supports(versions: List[int]) -> Callable: + def g(f: Callable) -> Callable: + if f.__doc__ is None: + f.__doc__ = "" + f.__doc__ += """\n\n + Supports Uniswap + """ + ", ".join( + "v" + str(ver) for ver in versions + ) + + @functools.wraps(f) + def check_version(self: "Uniswap", *args: List, **kwargs: Dict) -> Any: + if self.version not in versions: + raise Exception( + f"Function {f.__name__} does not support version {self.version} of Uniswap passed to constructor" + ) + return f(self, *args, **kwargs) + + return check_version + + return g diff --git a/uniswap/exceptions.py b/uniswap/exceptions.py new file mode 100644 index 0000000..8080b77 --- /dev/null +++ b/uniswap/exceptions.py @@ -0,0 +1,15 @@ +from typing import Any + + +class InvalidToken(Exception): + """Raised when an invalid token address is used.""" + + def __init__(self, address: Any) -> None: + Exception.__init__(self, f"Invalid token address: {address}") + + +class InsufficientBalance(Exception): + """Raised when the account has insufficient balance for a transaction.""" + + def __init__(self, had: int, needed: int) -> None: + Exception.__init__(self, f"Insufficient balance. Had {had}, needed {needed}") diff --git a/uniswap/token.py b/uniswap/token.py new file mode 100644 index 0000000..bcbeaf5 --- /dev/null +++ b/uniswap/token.py @@ -0,0 +1,30 @@ +from dataclasses import dataclass +from .types import AddressLike + + +@dataclass +class BaseToken: + """Base for tokens of all kinds""" + + symbol: str + """Symbol such as ETH, DAI, etc.""" + + address: AddressLike + """Address of the token contract.""" + + def __repr__(self) -> str: + return f"BaseToken({self.symbol}, {self.address!r})" + + +@dataclass +class ERC20Token(BaseToken): + """Represents an ERC20 token""" + + name: str + """Name of the token, as specified in the contract.""" + + decimals: int + """Decimals used to denominate the token.""" + + def __repr__(self) -> str: + return f"Token({self.symbol}, {self.address!r}, {self.decimals})" diff --git a/uniswap/types.py b/uniswap/types.py new file mode 100644 index 0000000..62f2b4d --- /dev/null +++ b/uniswap/types.py @@ -0,0 +1,6 @@ +from typing import Union +from web3.eth import Contract # noqa: F401 +from web3.types import Address, ChecksumAddress + + +AddressLike = Union[Address, ChecksumAddress] diff --git a/uniswap/uniswap.py b/uniswap/uniswap.py index abd3bc8..b940cf3 100644 --- a/uniswap/uniswap.py +++ b/uniswap/uniswap.py @@ -1,9 +1,8 @@ import os -import json import time import logging import functools -from typing import List, Any, Optional, Callable, Union, Tuple, Dict +from typing import List, Any, Optional, Union, Tuple, Dict from web3 import Web3 from web3.eth import Contract @@ -13,183 +12,47 @@ Wei, Address, ChecksumAddress, - ENS, Nonce, HexBytes, ) from eth_utils import is_same_address -from eth_typing import AnyAddress +from .types import AddressLike +from .token import ERC20Token from .tokens import tokens, tokens_rinkeby - -ETH_ADDRESS = "0x0000000000000000000000000000000000000000" -WETH9_ADDRESS = Web3.toChecksumAddress("0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2") +from .exceptions import InvalidToken, InsufficientBalance +from .util import ( + _str_to_addr, + _addr_to_str, + _validate_address, + _load_contract, + _load_contract_erc20, +) +from .decorators import supports, check_approval +from .constants import ( + _netid_to_name, + _factory_contract_addresses_v1, + _factory_contract_addresses_v2, + _router_contract_addresses_v2, + ETH_ADDRESS, +) logger = logging.getLogger(__name__) -# TODO: Consider dropping support for ENS altogether and instead use AnyAddress -AddressLike = Union[Address, ChecksumAddress, ENS] - - -class InvalidToken(Exception): - """Raised when an invalid token address is used.""" - - def __init__(self, address: Any) -> None: - Exception.__init__(self, f"Invalid token address: {address}") - - -class InsufficientBalance(Exception): - """Raised when the account has insufficient balance for a transaction.""" - - def __init__(self, had: int, needed: int) -> None: - Exception.__init__(self, f"Insufficient balance. Had {had}, needed {needed}") - - -def _load_abi(name: str) -> str: - path = f"{os.path.dirname(os.path.abspath(__file__))}/assets/" - with open(os.path.abspath(path + f"{name}.abi")) as f: - abi: str = json.load(f) - return abi - - -def check_approval(method: Callable) -> Callable: - """Decorator to check if user is approved for a token. It approves them if they - need to be approved.""" - - @functools.wraps(method) - def approved(self: Any, *args: Any, **kwargs: Any) -> Any: - # Check to see if the first token is actually ETH - token = args[0] if args[0] != ETH_ADDRESS else None - token_two = None - - # Check second token, if needed - if method.__name__ == "make_trade" or method.__name__ == "make_trade_output": - token_two = args[1] if args[1] != ETH_ADDRESS else None - - # Approve both tokens, if needed - if token: - is_approved = self._is_approved(token) - # logger.warning(f"Approved? {token}: {is_approved}") - if not is_approved: - self.approve(token) - if token_two: - is_approved = self._is_approved(token_two) - # logger.warning(f"Approved? {token_two}: {is_approved}") - if not is_approved: - self.approve(token_two) - return method(self, *args, **kwargs) - - return approved - - -def supports(versions: List[int]) -> Callable: - def g(f: Callable) -> Callable: - if f.__doc__ is None: - f.__doc__ = "" - f.__doc__ += """\n\n - Supports Uniswap - """ + ", ".join( - "v" + str(ver) for ver in versions - ) - - @functools.wraps(f) - def check_version(self: "Uniswap", *args: List, **kwargs: Dict) -> Any: - if self.version not in versions: - raise Exception( - f"Function {f.__name__} does not support version {self.version} of Uniswap passed to constructor" - ) - return f(self, *args, **kwargs) - - return check_version - - return g - - -def _str_to_addr(s: Union[str, Address]) -> AddressLike: - """Idempotent""" - if isinstance(s, str): - if s.startswith("0x"): - return Address(bytes.fromhex(s[2:])) - elif s.endswith(".eth"): - return ENS(s) - else: - raise Exception(f"Couldn't convert string '{s}' to AddressLike") - else: - return s - - -def _addr_to_str(a: AddressLike) -> str: - if isinstance(a, bytes): - # Address or ChecksumAddress - addr: str = Web3.toChecksumAddress("0x" + bytes(a).hex()) - return addr - elif isinstance(a, str): - if a.endswith(".eth"): - # Address is ENS - raise Exception("ENS not supported for this operation") - elif a.startswith("0x"): - addr = Web3.toChecksumAddress(a) - return addr - - raise InvalidToken(a) - - -def _validate_address(a: AddressLike) -> None: - assert _addr_to_str(a) - - -# see: https://chainid.network/chains/ -_netid_to_name = { - 1: "mainnet", - 3: "ropsten", - 4: "rinkeby", - 56: "binance", - 97: "binance_testnet", - 100: "xdai", -} - -_factory_contract_addresses_v1 = { - "mainnet": "0xc0a47dFe034B400B47bDaD5FecDa2621de6c4d95", - "ropsten": "0x9c83dCE8CA20E9aAF9D3efc003b2ea62aBC08351", - "rinkeby": "0xf5D915570BC477f9B8D6C0E980aA81757A3AaC36", - "kovan": "0xD3E51Ef092B2845f10401a0159B2B96e8B6c3D30", - "görli": "0x6Ce570d02D73d4c384b46135E87f8C592A8c86dA", -} - - -# For v2 the address is the same on mainnet, Ropsten, Rinkeby, Görli, and Kovan -# https://uniswap.org/docs/v2/smart-contracts/factory -_factory_contract_addresses_v2 = { - "mainnet": "0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f", - "ropsten": "0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f", - "rinkeby": "0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f", - "görli": "0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f", - "xdai": "0xA818b4F111Ccac7AA31D0BCc0806d64F2E0737D7", -} - -_router_contract_addresses_v2 = { - "mainnet": "0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D", - "ropsten": "0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D", - "rinkeby": "0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D", - "görli": "0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D", - "xdai": "0x1C232F01118CB8B424793ae03F870aa7D0ac7f77", -} - - class Uniswap: """ - Wrapper around Uniswap v1 and v2 contracts. + Wrapper around Uniswap contracts. """ def __init__( self, - address: Union[str, AddressLike, None], + address: Union[AddressLike, str, None], private_key: Optional[str], provider: str = None, web3: Web3 = None, version: int = 1, - max_slippage: float = 0.1, + default_slippage: float = 0.01, factory_contract_addr: str = None, router_contract_addr: str = None, ) -> None: @@ -199,7 +62,7 @@ def __init__( :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 max_slippage: Max allowed slippage for a trade. + :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). """ @@ -214,7 +77,7 @@ def __init__( self.version = version # TODO: Write tests for slippage - self.max_slippage = max_slippage + self.default_slippage = default_slippage if web3: self.w3 = web3 @@ -248,7 +111,8 @@ def __init__( if factory_contract_addr is None: factory_contract_addr = _factory_contract_addresses_v1[self.network] - self.factory_contract = self._load_contract( + self.factory_contract = _load_contract( + self.w3, abi_name="uniswap-v1/factory", address=_str_to_addr(factory_contract_addr), ) @@ -259,13 +123,14 @@ def __init__( if factory_contract_addr is None: factory_contract_addr = _factory_contract_addresses_v2[self.network] - self.factory_contract = self._load_contract( + 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 = self._load_contract( - abi_name="uniswap-v2/router02", address=self.router_address, + 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 @@ -273,11 +138,11 @@ def __init__( self.router_address = _str_to_addr( "0xE592427A0AEce92De3Edee1F18E0157C05861564" ) - self.quoter = self._load_contract( - abi_name="uniswap-v3/quoter", address=quoter_addr + self.quoter = _load_contract( + self.w3, abi_name="uniswap-v3/quoter", address=quoter_addr ) - self.router = self._load_contract( - abi_name="uniswap-v3/router", address=self.router_address + self.router = _load_contract( + self.w3, abi_name="uniswap-v3/router", address=self.router_address ) else: raise Exception(f"Invalid version '{self.version}', only 1 or 2 supported") @@ -285,108 +150,53 @@ def __init__( if hasattr(self, "factory_contract"): logger.info(f"Using factory contract: {self.factory_contract}") - @supports([1]) - def get_all_tokens(self) -> List[dict]: - """ - Retrieves all token pairs. - - Note: This is a *very* expensive operation and might therefore not work properly. - """ - # FIXME: This is a very expensive operation, would benefit greatly from caching. - tokenCount = self.factory_contract.functions.tokenCount().call() - tokens = [] - for i in range(tokenCount): - address = self.factory_contract.functions.getTokenWithId(i).call() - if address == "0x0000000000000000000000000000000000000000": - # Token is ETH - continue - token = self.get_token(address) - tokens.append(token) - return tokens - - def get_token(self, address: AddressLike) -> dict: - """ - Retrieves metadata from the ERC20 contract of a given token, like its name, symbol, and decimals. - """ - # FIXME: This function should always return the same output for the same input - # and would therefore benefit from caching - token_contract = self._load_contract(abi_name="erc20", address=address) - try: - name = token_contract.functions.name().call() - symbol = token_contract.functions.symbol().call() - decimals = token_contract.functions.decimals().call() - except Exception as e: - logger.warning( - f"Exception occurred while trying to get token {_addr_to_str(address)}: {e}" - ) - raise InvalidToken(address) - return {"name": name, "symbol": symbol, "decimals": decimals} - - @supports([1]) - def exchange_address_from_token(self, token_addr: AddressLike) -> AddressLike: - ex_addr: AddressLike = self.factory_contract.functions.getExchange( - token_addr - ).call() - # TODO: What happens if the token doesn't have an exchange/doesn't exist? - # Should probably raise an Exception (and test it) - return ex_addr - - @supports([1]) - def token_address_from_exchange(self, exchange_addr: AddressLike) -> Address: - token_addr: Address = ( - self.exchange_contract(ex_addr=exchange_addr) - .functions.tokenAddress(exchange_addr) - .call() - ) - return token_addr - - @functools.lru_cache() - @supports([1]) - def exchange_contract( - self, token_addr: AddressLike = None, ex_addr: AddressLike = None - ) -> Contract: - if not ex_addr and token_addr: - ex_addr = self.exchange_address_from_token(token_addr) - if ex_addr is None: - raise InvalidToken(token_addr) - abi_name = "uniswap-v1/exchange" - contract = self._load_contract(abi_name=abi_name, address=ex_addr) - logger.info(f"Loaded exchange contract {contract} at {contract.address}") - return contract - - @functools.lru_cache() - def erc20_contract(self, token_addr: AddressLike) -> Contract: - return self._load_contract(abi_name="erc20", address=token_addr) - - @functools.lru_cache() - @supports([2, 3]) - def get_weth_address(self) -> ChecksumAddress: - if self.version == 2: - # Contract calls should always return checksummed addresses - address: ChecksumAddress = self.router.functions.WETH().call() - elif self.version == 3: - address = self.router.functions.WETH9().call() - return address - - def _load_contract(self, abi_name: str, address: AddressLike) -> Contract: - return self.w3.eth.contract(address=address, abi=_load_abi(abi_name)) - - # ------ Exchange ------------------------------------------------------------------ - @supports([1, 2]) - def get_fee_maker(self) -> float: - """Get the maker fee.""" - return 0 + # ------ Market -------------------------------------------------------------------- - @supports([1, 2]) - def get_fee_taker(self) -> float: - """Get the taker fee.""" - return 0.003 + @supports([1, 2, 3]) + def get_price_input( + self, + token0: AddressLike, + token1: AddressLike, + qty: int, + fee: int = None, + route: Optional[List[AddressLike]] = None, + ) -> int: + """Returns the amount of the input token you get for `qty` of the output token""" + if fee is None: + fee = 3000 + if self.version == 3: + logger.warning("No fee set, assuming 0.3%") + + if token0 == ETH_ADDRESS: + return self._get_eth_token_input_price(token1, Wei(qty), fee) + elif token1 == ETH_ADDRESS: + return self._get_token_eth_input_price(token0, qty, fee) + else: + return self._get_token_token_input_price(token0, token1, qty, fee, route) - # ------ Market -------------------------------------------------------------------- @supports([1, 2, 3]) - def get_eth_token_input_price( - self, token: AddressLike, qty: Wei, fee: int = 3000 - ) -> Wei: + def get_price_output( + self, + token0: AddressLike, + token1: AddressLike, + qty: int, + fee: int = None, + route: Optional[List[AddressLike]] = None, + ) -> int: + """Returns the amount of input token you need to get `qty` of the output token""" + if fee is None: + fee = 3000 + if self.version == 3: + logger.warning("No fee set, assuming 0.3%") + + if token0 == ETH_ADDRESS: + return self._get_eth_token_output_price(token1, qty, fee) + elif token1 == ETH_ADDRESS: + return self._get_token_eth_output_price(token0, Wei(qty), fee) + else: + return self._get_token_token_output_price(token0, token1, qty, fee, route) + + def _get_eth_token_input_price(self, token: AddressLike, qty: Wei, fee: int) -> Wei: """Public price for ETH to Token trades with an exact input.""" if self.version == 1: ex = self.exchange_contract(token) @@ -396,15 +206,12 @@ def get_eth_token_input_price( qty, [self.get_weth_address(), token] ).call()[-1] elif self.version == 3: - price = self.get_token_token_input_price( + price = self._get_token_token_input_price( self.get_weth_address(), token, qty, fee=fee - ) + ) # type: ignore return price - @supports([1, 2, 3]) - def get_token_eth_input_price( - self, token: AddressLike, qty: int, fee: int = 3000 - ) -> int: + def _get_token_eth_input_price(self, token: AddressLike, qty: int, fee: int) -> int: """Public price for token to ETH trades with an exact input.""" if self.version == 1: ex = self.exchange_contract(token) @@ -414,19 +221,18 @@ def get_token_eth_input_price( qty, [token, self.get_weth_address()] ).call()[-1] elif self.version == 3: - price = self.get_token_token_input_price( + price = self._get_token_token_input_price( token, self.get_weth_address(), qty, fee=fee ) return price - @supports([2, 3]) - def get_token_token_input_price( + def _get_token_token_input_price( self, - token0: AnyAddress, - token1: AnyAddress, + token0: AddressLike, + token1: AddressLike, qty: int, - route: Optional[List[AnyAddress]] = None, - fee: int = 3000, + fee: int, + route: Optional[List[AddressLike]] = None, ) -> int: """ Public price for token to token trades with an exact input. @@ -437,10 +243,10 @@ def get_token_token_input_price( if 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, qty)) - elif is_same_address(token1, self.get_weth_address()): - return int(self.get_token_eth_input_price(token0, qty)) + if is_same_address(token0, self.get_weth_address()): # type: ignore + return int(self._get_eth_token_input_price(token1, Wei(qty), fee)) + elif is_same_address(token1, self.get_weth_address()): # type: ignore + 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}") @@ -453,19 +259,17 @@ def get_token_token_input_price( # result: tuple = self.quoter.functions.quoteExactInput(route, qty).call() raise Exception("custom route not yet supported for v3") - # FIXME: This comment is no longer valid, but some indication is needed if fee is unset in V3 - logger.warning("Assuming 0.3% fee") - # 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 - @supports([1, 2, 3]) - def get_eth_token_output_price( - self, token: AddressLike, qty: int, fee: int = 3000 + def _get_eth_token_output_price( + self, token: AddressLike, qty: int, fee: int = None ) -> Wei: """Public price for ETH to Token trades with an exact output.""" if self.version == 1: @@ -475,37 +279,42 @@ def get_eth_token_output_price( route = [self.get_weth_address(), token] price = self.router.functions.getAmountsIn(qty, route).call()[0] elif self.version == 3: - price = self.get_token_token_output_price( - self.get_weth_address(), token, qty, fee=fee + if not fee: + logger.warning("No fee set, assuming 0.3%") + fee = 3000 + price = Wei( + self._get_token_token_output_price( + self.get_weth_address(), token, qty, fee=fee + ) ) return price - @supports([1, 2, 3]) - def get_token_eth_output_price( - self, token: AddressLike, qty: Wei, fee: int = 3000 + def _get_token_eth_output_price( + self, token: AddressLike, qty: Wei, fee: int = None ) -> int: """Public price for token to ETH trades with an exact output.""" if self.version == 1: ex = self.exchange_contract(token) price: int = ex.functions.getTokenToEthOutputPrice(qty).call() elif self.version == 2: - price = self.router.functions.getAmountsIn( - qty, [token, self.get_weth_address()] - ).call()[0] + 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( + if not fee: + logger.warning("No fee set, assuming 0.3%") + fee = 3000 + price = self._get_token_token_output_price( token, self.get_weth_address(), qty, fee=fee ) return price - @supports([2, 3]) - def get_token_token_output_price( + def _get_token_token_output_price( self, - token0: AnyAddress, - token1: AnyAddress, + token0: AddressLike, + token1: AddressLike, qty: int, - route: Optional[List[AnyAddress]] = None, - fee: int = 3000, + fee: int = None, + route: Optional[List[AddressLike]] = None, ) -> int: """ Public price for token to token trades with an exact output. @@ -517,10 +326,10 @@ def get_token_token_output_price( # If one of the tokens are WETH, delegate to appropriate call. # See: https://github.com/shanefontaine/uniswap-python/issues/22 # TODO: Will these equality checks always work? (Address vs ChecksumAddress vs str) - if is_same_address(token0, self.get_weth_address()): - return int(self.get_eth_token_output_price(token1, qty)) - elif is_same_address(token1, self.get_weth_address()): - return int(self.get_token_eth_output_price(token0, qty)) + if is_same_address(token0, self.get_weth_address()): # type: ignore + return int(self._get_eth_token_output_price(token1, qty, fee)) + elif is_same_address(token1, self.get_weth_address()): # type: ignore + 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}") @@ -528,12 +337,14 @@ def get_token_token_output_price( if self.version == 2: price: int = self.router.functions.getAmountsIn(qty, route).call()[0] elif self.version == 3: + if not fee: + logger.warning("No fee set, assuming 0.3%") + fee = 3000 if route: # 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() raise Exception("custom route not yet supported for v3") - logger.warning("Assuming 0.3% fee") # 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 @@ -541,69 +352,10 @@ def get_token_token_output_price( price = self.quoter.functions.quoteExactOutputSingle( token0, token1, fee, qty, sqrtPriceLimitX96 ).call() + else: + raise ValueError("function not supported for this version of Uniswap") return price - # ------ Wallet balance ------------------------------------------------------------ - def get_eth_balance(self) -> Wei: - """Get the balance of ETH in a wallet.""" - return self.w3.eth.getBalance(self.address) - - def get_token_balance(self, token: AddressLike) -> int: - """Get the balance of a token in a wallet.""" - _validate_address(token) - if _addr_to_str(token) == ETH_ADDRESS: - return self.get_eth_balance() - erc20 = self.erc20_contract(token) - balance: int = erc20.functions.balanceOf(self.address).call() - return balance - - # ------ ERC20 Pool ---------------------------------------------------------------- - @supports([1]) - def get_ex_eth_balance(self, token: AddressLike) -> int: - """Get the balance of ETH in an exchange contract.""" - ex_addr: AddressLike = self.exchange_address_from_token(token) - return self.w3.eth.getBalance(ex_addr) - - @supports([1]) - def get_ex_token_balance(self, token: AddressLike) -> int: - """Get the balance of a token in an exchange contract.""" - erc20 = self.erc20_contract(token) - balance: int = erc20.functions.balanceOf( - self.exchange_address_from_token(token) - ).call() - return balance - - # TODO: ADD TOTAL SUPPLY - @supports([1]) - def get_exchange_rate(self, token: AddressLike) -> float: - """Get the current ETH/token exchange rate of the token.""" - eth_reserve = self.get_ex_eth_balance(token) - token_reserve = self.get_ex_token_balance(token) - return float(token_reserve / eth_reserve) - - # ------ Liquidity ----------------------------------------------------------------- - @supports([1]) - @check_approval - def add_liquidity( - self, token: AddressLike, max_eth: Wei, min_liquidity: int = 1 - ) -> HexBytes: - """Add liquidity to the pool.""" - tx_params = self._get_tx_params(max_eth) - # Add 1 to avoid rounding errors, per - # https://hackmd.io/hthz9hXKQmSyXfMbPsut1g#Add-Liquidity-Calculations - max_token = int(max_eth * self.get_exchange_rate(token)) + 10 - func_params = [min_liquidity, max_token, self._deadline()] - function = self.exchange_contract(token).functions.addLiquidity(*func_params) - return self._build_and_send_tx(function, tx_params) - - @supports([1]) - @check_approval - def remove_liquidity(self, token: str, max_token: int) -> HexBytes: - """Remove liquidity from the pool.""" - func_params = [int(max_token), 1, 1, self._deadline()] - function = self.exchange_contract(token).functions.removeLiquidity(*func_params) - return self._build_and_send_tx(function) - # ------ Make Trade ---------------------------------------------------------------- @check_approval def make_trade( @@ -612,19 +364,33 @@ def make_trade( output_token: AddressLike, qty: Union[int, Wei], recipient: AddressLike = None, + fee: int = None, + slippage: float = None, ) -> HexBytes: """Make a trade by defining the qty of the input token.""" + if fee is None: + fee = 3000 + if self.version == 3: + logger.warning("No fee set, assuming 0.3%") + + if slippage is None: + slippage = self.default_slippage + if input_token == ETH_ADDRESS: - return self._eth_to_token_swap_input(output_token, Wei(qty), recipient) + return self._eth_to_token_swap_input( + output_token, Wei(qty), recipient, fee, slippage + ) else: balance = self.get_token_balance(input_token) if balance < qty: raise InsufficientBalance(balance, qty) if output_token == ETH_ADDRESS: - return self._token_to_eth_swap_input(input_token, qty, recipient) + return self._token_to_eth_swap_input( + input_token, qty, recipient, fee, slippage + ) else: return self._token_to_token_swap_input( - input_token, output_token, qty, recipient + input_token, output_token, qty, recipient, fee, slippage ) @check_approval @@ -634,24 +400,43 @@ def make_trade_output( output_token: AddressLike, qty: Union[int, Wei], recipient: AddressLike = None, + fee: int = None, + slippage: float = None, ) -> HexBytes: """Make a trade by defining the qty of the output token.""" + if fee is None: + fee = 3000 + if self.version == 3: + logger.warning("No fee set, assuming 0.3%") + + if slippage is None: + slippage = self.default_slippage + if input_token == ETH_ADDRESS: balance = self.get_eth_balance() - need = self.get_eth_token_output_price(output_token, qty) + need = self._get_eth_token_output_price(output_token, qty) if balance < need: raise InsufficientBalance(balance, need) - return self._eth_to_token_swap_output(output_token, qty, recipient) + return self._eth_to_token_swap_output( + output_token, qty, recipient, fee, slippage + ) elif output_token == ETH_ADDRESS: qty = Wei(qty) - return self._token_to_eth_swap_output(input_token, qty, recipient) + return self._token_to_eth_swap_output( + input_token, qty, recipient, fee, slippage + ) else: return self._token_to_token_swap_output( - input_token, output_token, qty, recipient + input_token, output_token, qty, recipient, fee, slippage ) def _eth_to_token_swap_input( - self, output_token: AddressLike, qty: Wei, recipient: Optional[AddressLike] + self, + output_token: AddressLike, + qty: Wei, + recipient: Optional[AddressLike], + fee: int, + slippage: float, ) -> HexBytes: """Convert ETH to tokens given an input amount.""" eth_balance = self.get_eth_balance() @@ -673,8 +458,7 @@ def _eth_to_token_swap_input( if recipient is None: recipient = self.address amount_out_min = int( - (1 - self.max_slippage) - * self.get_eth_token_input_price(output_token, qty) + (1 - slippage) * self._get_eth_token_input_price(output_token, qty, fee) ) return self._build_and_send_tx( self.router.functions.swapExactETHForTokens( @@ -687,13 +471,18 @@ def _eth_to_token_swap_input( ) elif self.version == 3: return self._token_to_token_swap_input( - self.get_weth_address(), output_token, qty, recipient + self.get_weth_address(), output_token, qty, recipient, fee, slippage ) else: raise ValueError def _token_to_eth_swap_input( - self, input_token: AddressLike, qty: int, recipient: Optional[AddressLike] + self, + input_token: AddressLike, + qty: int, + recipient: Optional[AddressLike], + fee: int, + slippage: float, ) -> HexBytes: """Convert tokens to ETH given an input amount.""" # Balance check @@ -714,8 +503,7 @@ def _token_to_eth_swap_input( if recipient is None: recipient = self.address amount_out_min = int( - (1 - self.max_slippage) - * self.get_token_eth_input_price(input_token, qty) + (1 - slippage) * self._get_token_eth_input_price(input_token, qty, fee) ) return self._build_and_send_tx( self.router.functions.swapExactTokensForETH( @@ -728,7 +516,7 @@ def _token_to_eth_swap_input( ) elif self.version == 3: return self._token_to_token_swap_input( - input_token, self.get_weth_address(), qty, recipient + input_token, self.get_weth_address(), qty, recipient, fee, slippage ) else: raise ValueError @@ -739,7 +527,8 @@ def _token_to_token_swap_input( output_token: AddressLike, qty: int, recipient: Optional[AddressLike], - fee: int = 3000, + fee: int, + slippage: float, ) -> HexBytes: """Convert tokens to tokens given an input amount.""" if recipient is None: @@ -747,7 +536,7 @@ def _token_to_token_swap_input( 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_input_token( + min_tokens_bought, min_eth_bought = self._calculate_max_output_token( input_token, qty, output_token ) func_params = [ @@ -765,8 +554,10 @@ def _token_to_token_swap_input( return self._build_and_send_tx(function) elif self.version == 2: min_tokens_bought = int( - (1 - self.max_slippage) - * self.get_token_token_input_price(input_token, output_token, qty) + (1 - slippage) + * self._get_token_token_input_price( + input_token, output_token, qty, fee=fee + ) ) return self._build_and_send_tx( self.router.functions.swapExactTokensForTokens( @@ -779,8 +570,8 @@ def _token_to_token_swap_input( ) elif self.version == 3: min_tokens_bought = int( - (1 - self.max_slippage) - * self.get_token_token_input_price( + (1 - slippage) + * self._get_token_token_input_price( input_token, output_token, qty, fee=fee ) ) @@ -806,12 +597,17 @@ def _token_to_token_swap_input( raise ValueError def _eth_to_token_swap_output( - self, output_token: AddressLike, qty: int, recipient: Optional[AddressLike] + self, + output_token: AddressLike, + qty: int, + recipient: Optional[AddressLike], + fee: int, + slippage: float, ) -> HexBytes: """Convert ETH to tokens given an output amount.""" if self.version == 1: token_funcs = self.exchange_contract(output_token).functions - eth_qty = self.get_eth_token_output_price(output_token, qty) + 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: @@ -824,9 +620,9 @@ def _eth_to_token_swap_output( if recipient is None: recipient = self.address eth_qty = int( - (1 + self.max_slippage) - * self.get_eth_token_output_price(output_token, qty) - ) + (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, @@ -838,24 +634,27 @@ def _eth_to_token_swap_output( ) elif self.version == 3: return self._token_to_token_swap_output( - self.get_weth_address(), output_token, qty, recipient + self.get_weth_address(), output_token, qty, recipient, fee, slippage ) else: raise ValueError def _token_to_eth_swap_output( - self, input_token: AddressLike, qty: Wei, recipient: Optional[AddressLike] + self, + input_token: AddressLike, + qty: Wei, + recipient: Optional[AddressLike], + fee: int, + slippage: float, ) -> HexBytes: """Convert tokens to ETH given an output amount.""" # Balance check input_balance = self.get_token_balance(input_token) - cost = self.get_token_eth_output_price(input_token, qty) + cost = self._get_token_eth_output_price(input_token, qty, fee) if cost > input_balance: raise InsufficientBalance(input_balance, cost) if self.version == 1: - token_funcs = self.exchange_contract(input_token).functions - # 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 @@ -866,17 +665,18 @@ def _token_to_eth_swap_output( denominator = (outputReserve - outputAmount) * 997 inputAmount = numerator / denominator + 1 - max_tokens = int((1 + self.max_slippage) * inputAmount) + 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 = token_funcs.tokenToEthSwapOutput(*func_params) + function = ex.functions.tokenToEthSwapOutput(*func_params) else: func_params.append(recipient) - function = token_funcs.tokenToEthTransferOutput(*func_params) + function = ex.functions.tokenToEthTransferOutput(*func_params) return self._build_and_send_tx(function) elif self.version == 2: - max_tokens = int((1 + self.max_slippage) * cost) + max_tokens = int((1 + slippage) * cost) return self._build_and_send_tx( self.router.functions.swapTokensForExactETH( qty, @@ -888,7 +688,7 @@ def _token_to_eth_swap_output( ) elif self.version == 3: return self._token_to_token_swap_output( - input_token, self.get_weth_address(), qty, recipient + input_token, self.get_weth_address(), qty, recipient, fee, slippage ) else: raise ValueError @@ -899,7 +699,8 @@ def _token_to_token_swap_output( output_token: AddressLike, qty: int, recipient: Optional[AddressLike], - fee: int = 3000, + fee: int, + slippage: float, ) -> HexBytes: """ Convert tokens to tokens given an output amount. @@ -928,8 +729,10 @@ def _token_to_token_swap_output( elif self.version == 2: if recipient is None: recipient = self.address - cost = self.get_token_token_output_price(input_token, output_token, qty) - amount_in_max = int((1 + self.max_slippage) * cost) + cost = self._get_token_token_output_price( + input_token, output_token, qty, fee=fee + ) + amount_in_max = int((1 + slippage) * cost) return self._build_and_send_tx( self.router.functions.swapTokensForExactTokens( qty, @@ -943,10 +746,10 @@ def _token_to_token_swap_output( if recipient is None: recipient = self.address - cost = self.get_token_token_output_price( + cost = self._get_token_token_output_price( input_token, output_token, qty, fee=fee ) - amount_in_max = int((1 + self.max_slippage) * cost) + amount_in_max = int((1 + slippage) * cost) sqrtPriceLimitX96 = 0 return self._build_and_send_tx( @@ -971,6 +774,67 @@ def _token_to_token_swap_output( else: raise ValueError + # ------ Wallet balance ------------------------------------------------------------ + def get_eth_balance(self) -> Wei: + """Get the balance of ETH for your address.""" + return self.w3.eth.getBalance(self.address) + + def get_token_balance(self, token: AddressLike) -> int: + """Get the balance of a token for your address.""" + _validate_address(token) + if _addr_to_str(token) == ETH_ADDRESS: + return self.get_eth_balance() + erc20 = _load_contract_erc20(self.w3, token) + balance: int = erc20.functions.balanceOf(self.address).call() + return balance + + # ------ ERC20 Pool ---------------------------------------------------------------- + @supports([1]) + def get_ex_eth_balance(self, token: AddressLike) -> int: + """Get the balance of ETH in an exchange contract.""" + ex_addr: AddressLike = self.exchange_address_from_token(token) + return self.w3.eth.getBalance(ex_addr) + + @supports([1]) + def get_ex_token_balance(self, token: AddressLike) -> int: + """Get the balance of a token in an exchange contract.""" + erc20 = _load_contract_erc20(self.w3, token) + balance: int = erc20.functions.balanceOf( + self.exchange_address_from_token(token) + ).call() + return balance + + # TODO: ADD TOTAL SUPPLY + @supports([1]) + def get_exchange_rate(self, token: AddressLike) -> float: + """Get the current ETH/token exchange rate of the token.""" + eth_reserve = self.get_ex_eth_balance(token) + token_reserve = self.get_ex_token_balance(token) + return float(token_reserve / eth_reserve) + + # ------ Liquidity ----------------------------------------------------------------- + @supports([1]) + @check_approval + def add_liquidity( + self, token: AddressLike, max_eth: Wei, min_liquidity: int = 1 + ) -> HexBytes: + """Add liquidity to the pool.""" + tx_params = self._get_tx_params(max_eth) + # Add 1 to avoid rounding errors, per + # https://hackmd.io/hthz9hXKQmSyXfMbPsut1g#Add-Liquidity-Calculations + max_token = int(max_eth * self.get_exchange_rate(token)) + 10 + func_params = [min_liquidity, max_token, self._deadline()] + function = self.exchange_contract(token).functions.addLiquidity(*func_params) + return self._build_and_send_tx(function, tx_params) + + @supports([1]) + @check_approval + def remove_liquidity(self, token: str, max_token: int) -> HexBytes: + """Remove liquidity from the pool.""" + func_params = [int(max_token), 1, 1, self._deadline()] + function = self.exchange_contract(token).functions.removeLiquidity(*func_params) + return self._build_and_send_tx(function) + # ------ Approval Utils ------------------------------------------------------------ def approve(self, token: AddressLike, max_approval: Optional[int] = None) -> None: """Give an exchange/router max approval of a token.""" @@ -980,7 +844,7 @@ def approve(self, token: AddressLike, max_approval: Optional[int] = None) -> Non if self.version == 1 else self.router_address ) - function = self.erc20_contract(token).functions.approve( + function = _load_contract_erc20(self.w3, token).functions.approve( contract_addr, max_approval ) logger.warning(f"Approving {_addr_to_str(token)}...") @@ -998,7 +862,7 @@ def _is_approved(self, token: AddressLike) -> bool: elif self.version in [2, 3]: contract_addr = self.router_address amount = ( - self.erc20_contract(token) + _load_contract_erc20(self.w3, token) .functions.allowance(self.address, contract_addr) .call() ) @@ -1103,6 +967,48 @@ def _calculate_max_output_token( return int(outputAmountB), int(1.2 * outputAmountA) + # ------ Helpers ------------------------------------------------------------ + + def get_token(self, address: AddressLike) -> ERC20Token: + """ + Retrieves metadata from the ERC20 contract of a given token, like its name, symbol, and decimals. + """ + # FIXME: This function should always return the same output for the same input + # and would therefore benefit from caching + token_contract = _load_contract(self.w3, abi_name="erc20", address=address) + try: + name = token_contract.functions.name().call() + symbol = token_contract.functions.symbol().call() + decimals = token_contract.functions.decimals().call() + except Exception as e: + logger.warning( + f"Exception occurred while trying to get token {_addr_to_str(address)}: {e}" + ) + raise InvalidToken(address) + return ERC20Token(symbol, address, name, decimals) + + @functools.lru_cache() + @supports([2, 3]) + def get_weth_address(self) -> ChecksumAddress: + """Retrieves the WETH address from the contracts (which may vary between chains).""" + if self.version == 2: + # Contract calls should always return checksummed addresses + address: ChecksumAddress = self.router.functions.WETH().call() + elif self.version == 3: + address = self.router.functions.WETH9().call() + return address + + # ------ Exchange ------------------------------------------------------------------ + @supports([1, 2]) + def get_fee_maker(self) -> float: + """Get the maker fee.""" + return 0 + + @supports([1, 2]) + def get_fee_taker(self) -> float: + """Get the taker fee.""" + return 0.003 + # ------ Test utilities ------------------------------------------------------------ def _get_token_addresses(self) -> Dict[str, ChecksumAddress]: @@ -1118,3 +1024,56 @@ def _get_token_addresses(self) -> Dict[str, ChecksumAddress]: return tokens_rinkeby else: raise Exception(f"Unknown net '{netname}'") + + # ---- Old v1 utils ---- + + @supports([1]) + def exchange_address_from_token(self, token_addr: AddressLike) -> AddressLike: + ex_addr: AddressLike = self.factory_contract.functions.getExchange( + token_addr + ).call() + # TODO: What happens if the token doesn't have an exchange/doesn't exist? + # Should probably raise an Exception (and test it) + return ex_addr + + @supports([1]) + def token_address_from_exchange(self, exchange_addr: AddressLike) -> Address: + token_addr: Address = ( + self.exchange_contract(ex_addr=exchange_addr) + .functions.tokenAddress(exchange_addr) + .call() + ) + return token_addr + + @functools.lru_cache() + @supports([1]) + def exchange_contract( + self, token_addr: AddressLike = None, ex_addr: AddressLike = None + ) -> Contract: + if not ex_addr and token_addr: + ex_addr = self.exchange_address_from_token(token_addr) + if ex_addr is None: + raise InvalidToken(token_addr) + abi_name = "uniswap-v1/exchange" + contract = _load_contract(self.w3, abi_name=abi_name, address=ex_addr) + logger.info(f"Loaded exchange contract {contract} at {contract.address}") + return contract + + @supports([1]) + def get_all_tokens(self) -> List[ERC20Token]: + """ + Retrieves all token pairs. + + Note: This is a *very* expensive operation and might therefore not work properly. + """ + # FIXME: This is a very expensive operation, would benefit greatly from caching. + tokenCount = self.factory_contract.functions.tokenCount().call() + tokens = [] + for i in range(tokenCount): + address = self.factory_contract.functions.getTokenWithId(i).call() + if address == "0x0000000000000000000000000000000000000000": + # Token is ETH + continue + token = self.get_token(address) + tokens.append(token) + return tokens diff --git a/uniswap/util.py b/uniswap/util.py new file mode 100644 index 0000000..cdf0017 --- /dev/null +++ b/uniswap/util.py @@ -0,0 +1,62 @@ +import os +import json +import functools +from typing import Union, List, Tuple + +from web3 import Web3 + +from .types import AddressLike, Address, Contract +from .exceptions import InvalidToken + + +def _str_to_addr(s: Union[AddressLike, str]) -> Address: + """Idempotent""" + if isinstance(s, str): + if s.startswith("0x"): + return Address(bytes.fromhex(s[2:])) + else: + raise Exception(f"Couldn't convert string '{s}' to AddressLike") + else: + return s + + +def _addr_to_str(a: AddressLike) -> str: + if isinstance(a, bytes): + # Address or ChecksumAddress + addr: str = Web3.toChecksumAddress("0x" + bytes(a).hex()) + return addr + elif isinstance(a, str) and a.startswith("0x"): + addr = Web3.toChecksumAddress(a) + return addr + + raise InvalidToken(a) + + +def _validate_address(a: AddressLike) -> None: + assert _addr_to_str(a) + + +def _load_abi(name: str) -> str: + path = f"{os.path.dirname(os.path.abspath(__file__))}/assets/" + with open(os.path.abspath(path + f"{name}.abi")) as f: + abi: str = json.load(f) + return abi + + +@functools.lru_cache() +def _load_contract(w3: Web3, abi_name: str, address: AddressLike) -> Contract: + address = Web3.toChecksumAddress(address) + return w3.eth.contract(address=address, abi=_load_abi(abi_name)) + + +def _load_contract_erc20(w3: Web3, address: AddressLike) -> Contract: + return _load_contract(w3, "erc20", address) + + +def _encode_path(token_in: AddressLike, route: List[Tuple[int, AddressLike]]) -> bytes: + """ + Needed for multi-hop swaps in V3. + + https://github.com/Uniswap/uniswap-v3-sdk/blob/1a74d5f0a31040fec4aeb1f83bba01d7c03f4870/src/utils/encodeRouteToPath.ts + """ + raise NotImplementedError