Add Cowswap oracle

- Related to safe-global/safe-transaction-service#645

gnosis/eth/oracles/__init__.py

@@ -1,16 +1,19 @@
 # flake8: noqa F401
+from .cowswap import CowswapOracle
+from .exceptions import (
+    CannotGetPriceFromOracle,
+    InvalidPriceFromOracle,
+    OracleException,
+)
 from .kyber import KyberOracle
 from .oracles import (
     AaveOracle,
     BalancerOracle,
-    CannotGetPriceFromOracle,
     ComposedPriceOracle,
     CreamOracle,
     CurveOracle,
     EnzymeOracle,
-    InvalidPriceFromOracle,
     MooniswapOracle,
-    OracleException,
     PoolTogetherOracle,
     PriceOracle,
     PricePoolOracle,
@@ -27,15 +30,13 @@
 __all__ = [
     "AaveOracle",
     "BalancerOracle",
-    "CannotGetPriceFromOracle",
     "ComposedPriceOracle",
+    "CowswapOracle",
     "CreamOracle",
     "CurveOracle",
     "EnzymeOracle",
-    "InvalidPriceFromOracle",
     "KyberOracle",
     "MooniswapOracle",
-    "OracleException",
     "PoolTogetherOracle",
     "PriceOracle",
     "PricePoolOracle",

gnosis/eth/oracles/cowswap.py

@@ -0,0 +1,68 @@
+import logging
+from typing import Optional
+
+from gnosis.protocol import GnosisProtocolAPI, OrderKind
+
+from .. import EthereumClient, EthereumNetworkNotSupported
+from .exceptions import CannotGetPriceFromOracle
+from .oracles import PriceOracle
+from .utils import get_decimals
+
+logger = logging.getLogger(__name__)
+
+
+class CowswapOracle(PriceOracle):
+    """
+    CowSwap Oracle implementation
+
+    https://docs.cow.fi/off-chain-services/api
+    """
+
+    def __init__(
+        self,
+        ethereum_client: EthereumClient,
+    ):
+        """
+        :param ethereum_client:
+        """
+        self.ethereum_client = ethereum_client
+        self.w3 = ethereum_client.w3
+        self.api = GnosisProtocolAPI(ethereum_client.get_network())
+
+    @classmethod
+    def is_available(
+        cls,
+        ethereum_client: EthereumClient,
+    ) -> bool:
+        """
+        :param ethereum_client:
+        :return: `True` if CowSwap is available for the EthereumClient provided, `False` otherwise
+        """
+        try:
+            GnosisProtocolAPI(ethereum_client.get_network())
+            return True
+        except EthereumNetworkNotSupported:
+            return False
+
+    def get_price(
+        self, token_address_1: str, token_address_2: Optional[str] = None
+    ) -> float:
+        token_address_2 = token_address_2 or self.api.weth_address
+        if token_address_1 == token_address_2:
+            return 1.0
+
+        token_1_decimals = get_decimals(token_address_1, self.ethereum_client)
+        result = self.api.get_estimated_amount(
+            token_address_1, token_address_2, OrderKind.SELL, 10**token_1_decimals
+        )
+        if "errorType" in result:
+            error_message = (
+                f"Cannot get price from CowSwap {result} "
+                f"for token-1={token_address_1} to token-2={token_address_2}"
+            )
+            logger.warning(error_message)
+            raise CannotGetPriceFromOracle(error_message)
+        else:
+            # Decimals needs to be adjusted
+            token_2_decimals = get_decimals(token_address_2, self.ethereum_client)
+            return float(result["amount"]) / 10**token_2_decimals

gnosis/eth/oracles/exceptions.py

@@ -0,0 +1,10 @@
+class OracleException(Exception):
+    pass
+
+
+class InvalidPriceFromOracle(OracleException):
+    pass
+
+
+class CannotGetPriceFromOracle(OracleException):
+    pass

gnosis/eth/oracles/kyber.py

@@ -8,12 +8,9 @@
 
 from .. import EthereumClient, EthereumNetwork
 from ..contracts import get_kyber_network_proxy_contract
-from .oracles import (
-    CannotGetPriceFromOracle,
-    InvalidPriceFromOracle,
-    PriceOracle,
-    get_decimals,
-)
+from .exceptions import CannotGetPriceFromOracle, InvalidPriceFromOracle
+from .oracles import PriceOracle
+from .utils import get_decimals
 
 logger = logging.getLogger(__name__)
 

gnosis/eth/oracles/oracles.py

@@ -29,7 +29,9 @@
 from .abis.cream_abis import cream_ctoken_abi
 from .abis.mooniswap_abis import mooniswap_abi
 from .abis.zerion_abis import ZERION_TOKEN_ADAPTER_ABI
+from .exceptions import CannotGetPriceFromOracle, InvalidPriceFromOracle
 from .helpers.curve_gauge_list import CURVE_GAUGE_TO_LP_TOKEN
+from .utils import get_decimals
 
 logger = logging.getLogger(__name__)
 
@@ -40,18 +42,6 @@ class UnderlyingToken:
     quantity: int
 
 
-class OracleException(Exception):
-    pass
-
-
-class InvalidPriceFromOracle(OracleException):
-    pass
-
-
-class CannotGetPriceFromOracle(OracleException):
-    pass
-
-
 class PriceOracle(ABC):
     @abstractmethod
     def get_price(self, *args) -> float:
@@ -76,31 +66,6 @@ def get_underlying_tokens(self, *args) -> List[Tuple[UnderlyingToken]]:
         pass
 
 
-@functools.lru_cache(maxsize=10_000)
-def get_decimals(
-    token_address: ChecksumAddress, ethereum_client: EthereumClient
-) -> int:
-    """
-    Auxiliary function so RPC call to get `token decimals` is cached and
-    can be reused for every Oracle, instead of having a cache per Oracle.
-
-    :param token_address:
-    :param ethereum_client:
-    :return: Decimals for a token
-    :raises CannotGetPriceFromOracle: If there's a problem with the query
-    """
-    try:
-        return ethereum_client.erc20.get_decimals(token_address)
-    except (
-        ValueError,
-        BadFunctionCallOutput,
-        DecodingError,
-    ) as e:
-        error_message = f"Cannot get decimals for token={token_address}"
-        logger.warning(error_message)
-        raise CannotGetPriceFromOracle(error_message) from e
-
-
 class UniswapOracle(PriceOracle):
     ADDRESSES = {
         EthereumNetwork.MAINNET: "0xc0a47dFe034B400B47bDaD5FecDa2621de6c4d95",

gnosis/eth/oracles/uniswap_v3.py

@@ -16,7 +16,9 @@
     uniswap_v3_pool_abi,
     uniswap_v3_router_abi,
 )
-from .oracles import CannotGetPriceFromOracle, PriceOracle, get_decimals
+from .exceptions import CannotGetPriceFromOracle
+from .oracles import PriceOracle
+from .utils import get_decimals
 
 logger = logging.getLogger(__name__)
 
@@ -51,7 +53,7 @@ def is_available(
         """
         :param ethereum_client:
         :param uniswap_v3_router_address:
-        :return: `True` is Uniswap V3 is available for the EthereumClient provided, `False` otherwise
+        :return: `True` if Uniswap V3 is available for the EthereumClient provided, `False` otherwise
         """
         return ethereum_client.is_contract(
             uniswap_v3_router_address or cls.UNISWAP_V3_ROUTER

gnosis/eth/oracles/utils.py

@@ -0,0 +1,37 @@
+import functools
+import logging
+
+from eth_abi.exceptions import DecodingError
+from eth_typing import ChecksumAddress
+from web3.exceptions import BadFunctionCallOutput
+
+from gnosis.eth import EthereumClient
+
+from .exceptions import CannotGetPriceFromOracle
+
+logger = logging.getLogger(__name__)
+
+
+@functools.lru_cache(maxsize=10_000)
+def get_decimals(
+    token_address: ChecksumAddress, ethereum_client: EthereumClient
+) -> int:
+    """
+    Auxiliary function so RPC call to get `token decimals` is cached and
+    can be reused for every Oracle, instead of having a cache per Oracle.
+
+    :param token_address:
+    :param ethereum_client:
+    :return: Decimals for a token
+    :raises CannotGetPriceFromOracle: If there's a problem with the query
+    """
+    try:
+        return ethereum_client.erc20.get_decimals(token_address)
+    except (
+        ValueError,
+        BadFunctionCallOutput,
+        DecodingError,
+    ) as e:
+        error_message = f"Cannot get decimals for token={token_address}"
+        logger.warning(error_message)
+        raise CannotGetPriceFromOracle(error_message) from e

gnosis/eth/tests/oracles/test_cowswap.py

@@ -0,0 +1,73 @@
+from unittest import mock
+
+from django.test import TestCase
+
+from eth_account import Account
+
+from ... import EthereumClient
+from ...oracles import CannotGetPriceFromOracle, CowswapOracle
+from ..ethereum_test_case import EthereumTestCaseMixin
+from ..test_oracles import (
+    dai_token_mainnet_address,
+    usdc_token_mainnet_address,
+    usdt_token_mainnet_address,
+    weth_token_mainnet_address,
+)
+from ..utils import just_test_if_mainnet_node
+
+
+class TestCowswapOracle(EthereumTestCaseMixin, TestCase):
+    def test_get_price(self):
+        mainnet_node = just_test_if_mainnet_node()
+        ethereum_client = EthereumClient(mainnet_node)
+
+        self.assertTrue(CowswapOracle.is_available(ethereum_client))
+
+        cowswap_oracle = CowswapOracle(ethereum_client)
+
+        price = cowswap_oracle.get_price(weth_token_mainnet_address)
+        self.assertEqual(price, 1.0)
+
+        price = cowswap_oracle.get_price(dai_token_mainnet_address)
+        self.assertLess(price, 1)
+        self.assertGreater(price, 0)
+
+        price = cowswap_oracle.get_price(
+            weth_token_mainnet_address, dai_token_mainnet_address
+        )
+        self.assertGreater(price, 1)
+
+        # Test with 2 stablecoins with same decimals
+        price = cowswap_oracle.get_price(
+            usdt_token_mainnet_address, usdc_token_mainnet_address
+        )
+        self.assertAlmostEqual(price, 1.0, delta=0.5)
+
+        # Test with 2 stablecoins with different decimals
+        price = cowswap_oracle.get_price(
+            dai_token_mainnet_address, usdc_token_mainnet_address
+        )
+        self.assertAlmostEqual(price, 1.0, delta=0.5)
+
+        price = cowswap_oracle.get_price(
+            usdc_token_mainnet_address, dai_token_mainnet_address
+        )
+        self.assertAlmostEqual(price, 1.0, delta=0.5)
+
+        random_token = Account.create().address
+        with self.assertRaisesMessage(
+            CannotGetPriceFromOracle,
+            f"Cannot get decimals for token={random_token}",
+        ):
+            cowswap_oracle.get_price(random_token)
+
+        with mock.patch(
+            "gnosis.eth.oracles.cowswap.get_decimals", autospec=True, return_value=18
+        ):
+            with self.assertRaisesMessage(
+                CannotGetPriceFromOracle,
+                f"Cannot get price from CowSwap "
+                f"{{'errorType': 'UnsupportedToken', 'description': 'Token address {random_token.lower()}'}} "
+                f"for token-1={random_token} to token-2={weth_token_mainnet_address}",
+            ):
+                cowswap_oracle.get_price(random_token)

gnosis/eth/tests/oracles/test_sushiswap.py

@@ -2,7 +2,7 @@
 
 from ... import EthereumClient
 from ...oracles import SushiswapOracle
-from ...oracles.oracles import get_decimals as oracles_get_decimals
+from ...oracles.utils import get_decimals as oracles_get_decimals
 from ..ethereum_test_case import EthereumTestCaseMixin
 from ..test_oracles import (
     dai_token_mainnet_address,

gnosis/eth/tests/test_oracles.py

@@ -23,7 +23,7 @@
     YearnOracle,
     ZerionComposedOracle,
 )
-from ..oracles.oracles import get_decimals as oracles_get_decimals
+from ..oracles.utils import get_decimals as oracles_get_decimals
 from .ethereum_test_case import EthereumTestCaseMixin
 from .utils import just_test_if_mainnet_node
 

gnosis/protocol/gnosis_protocol_api.py

@@ -9,6 +9,7 @@
 from web3 import Web3
 
 from gnosis.eth import EthereumNetwork, EthereumNetworkNotSupported
+from gnosis.util import cached_property
 
 from .order import Order, OrderKind
 
@@ -37,7 +38,7 @@ class AmountResponse(TypedDict):
 
 
 class ErrorResponse(TypedDict):
-    error_type: str
+    errorType: str
     description: str
 
 
@@ -67,6 +68,18 @@ def __init__(self, ethereum_network: EthereumNetwork):
         self.domain_separator = self.build_domain_separator(self.network)
         self.base_url = self.api_base_urls[self.network]
 
+    @cached_property
+    def weth_address(self) -> ChecksumAddress:
+        """
+        :return: Wrapped ether checksummed address
+        """
+        if self.network == EthereumNetwork.MAINNET:
+            return ChecksumAddress("0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2")
+        elif self.network == EthereumNetwork.RINKEBY:
+            return ChecksumAddress("0xc778417E063141139Fce010982780140Aa0cD5Ab")
+        else:  # XDAI
+            return ChecksumAddress("0x6A023CCd1ff6F2045C3309768eAd9E68F978f6e1")
+
     @classmethod
     def build_domain_separator(cls, ethereum_network: EthereumNetwork):
         return make_domain(