9394559170

Committed 26 Apr 2024 06:36AM UTC coverage: 94.656% (-0.02%) from 94.674%

Build # 9394559170

Build Type

push

github

Committed by

xmatthias

Commit Message

Loader should be passed as kwarg for clarity

Run Details

20280 of 21425 relevant lines covered (94.66%)

0.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.77

/freqtrade/rpc/fiat_convert.py

"""
Module that define classes to convert Crypto-currency to FIAT
e.g BTC to USD
"""

import logging
from datetime import datetime
from typing import Dict, List

from cachetools import TTLCache
from pycoingecko import CoinGeckoAPI
from requests.exceptions import RequestException

from freqtrade.constants import SUPPORTED_FIAT
from freqtrade.mixins.logging_mixin import LoggingMixin


logger = logging.getLogger(__name__)


# Manually map symbol to ID for some common coins
# with duplicate coingecko entries
coingecko_mapping = {
    'eth': 'ethereum',
    'bnb': 'binancecoin',
    'sol': 'solana',
    'usdt': 'tether',
    'busd': 'binance-usd',
    'tusd': 'true-usd',
    'usdc': 'usd-coin',
    'btc': 'bitcoin'
}


class CryptoToFiatConverter(LoggingMixin):
    """
    Main class to initiate Crypto to FIAT.
    This object contains a list of pair Crypto, FIAT
    This object is also a Singleton
    """
    __instance = None
    _coingecko: CoinGeckoAPI = None
    _coinlistings: List[Dict] = []
    _backoff: float = 0.0

    def __new__(cls):
        """
        This class is a singleton - cannot be instantiated twice.
        """
        if CryptoToFiatConverter.__instance is None:
            CryptoToFiatConverter.__instance = object.__new__(cls)
            try:
                # Limit retires to 1 (0 and 1)
                # otherwise we risk bot impact if coingecko is down.
                CryptoToFiatConverter._coingecko = CoinGeckoAPI(retries=1)
            except BaseException:
                CryptoToFiatConverter._coingecko = None
        return CryptoToFiatConverter.__instance

    def __init__(self) -> None:
        # Timeout: 6h
        self._pair_price: TTLCache = TTLCache(maxsize=500, ttl=6 * 60 * 60)

        LoggingMixin.__init__(self, logger, 3600)
        self._load_cryptomap()

    def _load_cryptomap(self) -> None:
        try:
            # Use list-comprehension to ensure we get a list.
            self._coinlistings = [x for x in self._coingecko.get_coins_list()]
        except RequestException as request_exception:
            if "429" in str(request_exception):
                logger.warning(
                    "Too many requests for CoinGecko API, backing off and trying again later.")
                # Set backoff timestamp to 60 seconds in the future
                self._backoff = datetime.now().timestamp() + 60
                return
            # If the request is not a 429 error we want to raise the normal error
            logger.error(
                "Could not load FIAT Cryptocurrency map for the following problem: "
                f"{request_exception}"
            )
        except (Exception) as exception:
            logger.error(
                f"Could not load FIAT Cryptocurrency map for the following problem: {exception}")

    def _get_gecko_id(self, crypto_symbol):
        if not self._coinlistings:
            if self._backoff <= datetime.now().timestamp():
                self._load_cryptomap()
                # Still not loaded.
                if not self._coinlistings:
                    return None
            else:
                return None
        found = [x for x in self._coinlistings if x['symbol'].lower() == crypto_symbol]

        if crypto_symbol in coingecko_mapping.keys():
            found = [x for x in self._coinlistings if x['id'] == coingecko_mapping[crypto_symbol]]

        if len(found) == 1:
            return found[0]['id']

        if len(found) > 0:
            # Wrong!
            logger.warning(f"Found multiple mappings in CoinGecko for {crypto_symbol}.")
            return None

    def convert_amount(self, crypto_amount: float, crypto_symbol: str, fiat_symbol: str) -> float:
        """
        Convert an amount of crypto-currency to fiat
        :param crypto_amount: amount of crypto-currency to convert
        :param crypto_symbol: crypto-currency used
        :param fiat_symbol: fiat to convert to
        :return: float, value in fiat of the crypto-currency amount
        """
        if crypto_symbol == fiat_symbol:
            return float(crypto_amount)
        price = self.get_price(crypto_symbol=crypto_symbol, fiat_symbol=fiat_symbol)
        return float(crypto_amount) * float(price)

    def get_price(self, crypto_symbol: str, fiat_symbol: str) -> float:
        """
        Return the price of the Crypto-currency in Fiat
        :param crypto_symbol: Crypto-currency you want to convert (e.g BTC)
        :param fiat_symbol: FIAT currency you want to convert to (e.g USD)
        :return: Price in FIAT
        """
        crypto_symbol = crypto_symbol.lower()
        fiat_symbol = fiat_symbol.lower()
        inverse = False

        if crypto_symbol == 'usd':
            # usd corresponds to "uniswap-state-dollar" for coingecko.
            # We'll therefore need to "swap" the currencies
            logger.info(f"reversing Rates {crypto_symbol}, {fiat_symbol}")
            crypto_symbol = fiat_symbol
            fiat_symbol = 'usd'
            inverse = True

        symbol = f"{crypto_symbol}/{fiat_symbol}"
        # Check if the fiat conversion you want is supported
        if not self._is_supported_fiat(fiat=fiat_symbol):
            raise ValueError(f'The fiat {fiat_symbol} is not supported.')

        price = self._pair_price.get(symbol, None)

        if not price:
            price = self._find_price(
                crypto_symbol=crypto_symbol,
                fiat_symbol=fiat_symbol
            )
            if inverse and price != 0.0:
                price = 1 / price
            self._pair_price[symbol] = price

        return price

    def _is_supported_fiat(self, fiat: str) -> bool:
        """
        Check if the FIAT your want to convert to is supported
        :param fiat: FIAT to check (e.g USD)
        :return: bool, True supported, False not supported
        """

        return fiat.upper() in SUPPORTED_FIAT

    def _find_price(self, crypto_symbol: str, fiat_symbol: str) -> float:
        """
        Call CoinGecko API to retrieve the price in the FIAT
        :param crypto_symbol: Crypto-currency you want to convert (e.g btc)
        :param fiat_symbol: FIAT currency you want to convert to (e.g usd)
        :return: float, price of the crypto-currency in Fiat
        """
        # Check if the fiat conversion you want is supported
        if not self._is_supported_fiat(fiat=fiat_symbol):
            raise ValueError(f'The fiat {fiat_symbol} is not supported.')

        # No need to convert if both crypto and fiat are the same
        if crypto_symbol == fiat_symbol:
            return 1.0

        _gecko_id = self._get_gecko_id(crypto_symbol)

        if not _gecko_id:
            # return 0 for unsupported stake currencies (fiat-convert should not break the bot)
            self.log_once(
                f"unsupported crypto-symbol {crypto_symbol.upper()} - returning 0.0",
                logger.warning)
            return 0.0

        try:
            return float(
                self._coingecko.get_price(
                    ids=_gecko_id,
                    vs_currencies=fiat_symbol
                )[_gecko_id][fiat_symbol]
            )
        except Exception as exception:
            logger.error("Error in _find_price: %s", exception)
            return 0.0

1	"""
2	Module that define classes to convert Crypto-currency to FIAT
3	e.g BTC to USD
4	"""
5
6	import logging	1✔
7	from datetime import datetime	1✔
8	from typing import Dict, List	1✔
9
10	from cachetools import TTLCache	1✔
11	from pycoingecko import CoinGeckoAPI	1✔
12	from requests.exceptions import RequestException	1✔
13
14	from freqtrade.constants import SUPPORTED_FIAT	1✔
15	from freqtrade.mixins.logging_mixin import LoggingMixin	1✔
16
17
18	logger = logging.getLogger(__name__)	1✔
19
20
21	# Manually map symbol to ID for some common coins
22	# with duplicate coingecko entries
23	coingecko_mapping = {	1✔
24	'eth': 'ethereum',
25	'bnb': 'binancecoin',
26	'sol': 'solana',
27	'usdt': 'tether',
28	'busd': 'binance-usd',
29	'tusd': 'true-usd',
30	'usdc': 'usd-coin',
31	'btc': 'bitcoin'
32	}
33
34
35	class CryptoToFiatConverter(LoggingMixin):	1✔
36	"""
37	Main class to initiate Crypto to FIAT.
38	This object contains a list of pair Crypto, FIAT
39	This object is also a Singleton
40	"""
41	__instance = None	1✔
42	_coingecko: CoinGeckoAPI = None	1✔
43	_coinlistings: List[Dict] = []	1✔
44	_backoff: float = 0.0	1✔
45
46	def __new__(cls):	1✔
47	"""
48	This class is a singleton - cannot be instantiated twice.
49	"""
50	if CryptoToFiatConverter.__instance is None:	1✔
51	CryptoToFiatConverter.__instance = object.__new__(cls)	1✔
52	try:	1✔
53	# Limit retires to 1 (0 and 1)
54	# otherwise we risk bot impact if coingecko is down.
55	CryptoToFiatConverter._coingecko = CoinGeckoAPI(retries=1)	1✔
56	except BaseException:	×
57	CryptoToFiatConverter._coingecko = None	×
58	return CryptoToFiatConverter.__instance	1✔
59
60	def __init__(self) -> None:	1✔
61	# Timeout: 6h
62	self._pair_price: TTLCache = TTLCache(maxsize=500, ttl=6 * 60 * 60)	1✔
63
64	LoggingMixin.__init__(self, logger, 3600)	1✔
65	self._load_cryptomap()	1✔
66
67	def _load_cryptomap(self) -> None:	1✔
68	try:	1✔
69	# Use list-comprehension to ensure we get a list.
70	self._coinlistings = [x for x in self._coingecko.get_coins_list()]	1✔
71	except RequestException as request_exception:	1✔
72	if "429" in str(request_exception):	1✔
73	logger.warning(	1✔
74	"Too many requests for CoinGecko API, backing off and trying again later.")
75	# Set backoff timestamp to 60 seconds in the future
76	self._backoff = datetime.now().timestamp() + 60	1✔
77	return	1✔
78	# If the request is not a 429 error we want to raise the normal error
79	logger.error(	1✔
80	"Could not load FIAT Cryptocurrency map for the following problem: "
81	f"{request_exception}"
82	)
83	except (Exception) as exception:	1✔
84	logger.error(	1✔
85	f"Could not load FIAT Cryptocurrency map for the following problem: {exception}")
86
87	def _get_gecko_id(self, crypto_symbol):	1✔
88	if not self._coinlistings:	1✔
89	if self._backoff <= datetime.now().timestamp():	1✔
90	self._load_cryptomap()	1✔
91	# Still not loaded.
92	if not self._coinlistings:	1✔
93	return None	1✔
94	else:
95	return None	×
96	found = [x for x in self._coinlistings if x['symbol'].lower() == crypto_symbol]	1✔
97
98	if crypto_symbol in coingecko_mapping.keys():	1✔
99	found = [x for x in self._coinlistings if x['id'] == coingecko_mapping[crypto_symbol]]	1✔
100
101	if len(found) == 1:	1✔
102	return found[0]['id']	1✔
103
104	if len(found) > 0:	1✔
105	# Wrong!
106	logger.warning(f"Found multiple mappings in CoinGecko for {crypto_symbol}.")	1✔
107	return None	1✔
108
109	def convert_amount(self, crypto_amount: float, crypto_symbol: str, fiat_symbol: str) -> float:	1✔
110	"""
111	Convert an amount of crypto-currency to fiat
112	:param crypto_amount: amount of crypto-currency to convert
113	:param crypto_symbol: crypto-currency used
114	:param fiat_symbol: fiat to convert to
115	:return: float, value in fiat of the crypto-currency amount
116	"""
117	if crypto_symbol == fiat_symbol:	1✔
118	return float(crypto_amount)	1✔
119	price = self.get_price(crypto_symbol=crypto_symbol, fiat_symbol=fiat_symbol)	1✔
120	return float(crypto_amount) * float(price)	1✔
121
122	def get_price(self, crypto_symbol: str, fiat_symbol: str) -> float:	1✔
123	"""
124	Return the price of the Crypto-currency in Fiat
125	:param crypto_symbol: Crypto-currency you want to convert (e.g BTC)
126	:param fiat_symbol: FIAT currency you want to convert to (e.g USD)
127	:return: Price in FIAT
128	"""
129	crypto_symbol = crypto_symbol.lower()	1✔
130	fiat_symbol = fiat_symbol.lower()	1✔
131	inverse = False	1✔
132
133	if crypto_symbol == 'usd':	1✔
134	# usd corresponds to "uniswap-state-dollar" for coingecko.
135	# We'll therefore need to "swap" the currencies
136	logger.info(f"reversing Rates {crypto_symbol}, {fiat_symbol}")	1✔
137	crypto_symbol = fiat_symbol	1✔
138	fiat_symbol = 'usd'	1✔
139	inverse = True	1✔
140
141	symbol = f"{crypto_symbol}/{fiat_symbol}"	1✔
142	# Check if the fiat conversion you want is supported
143	if not self._is_supported_fiat(fiat=fiat_symbol):	1✔
144	raise ValueError(f'The fiat {fiat_symbol} is not supported.')	1✔
145
146	price = self._pair_price.get(symbol, None)	1✔
147
148	if not price:	1✔
149	price = self._find_price(	1✔
150	crypto_symbol=crypto_symbol,
151	fiat_symbol=fiat_symbol
152	)
153	if inverse and price != 0.0:	1✔
154	price = 1 / price	1✔
155	self._pair_price[symbol] = price	1✔
156
157	return price	1✔
158
159	def _is_supported_fiat(self, fiat: str) -> bool:	1✔
160	"""
161	Check if the FIAT your want to convert to is supported
162	:param fiat: FIAT to check (e.g USD)
163	:return: bool, True supported, False not supported
164	"""
165
166	return fiat.upper() in SUPPORTED_FIAT	1✔
167
168	def _find_price(self, crypto_symbol: str, fiat_symbol: str) -> float:	1✔
169	"""
170	Call CoinGecko API to retrieve the price in the FIAT
171	:param crypto_symbol: Crypto-currency you want to convert (e.g btc)
172	:param fiat_symbol: FIAT currency you want to convert to (e.g usd)
173	:return: float, price of the crypto-currency in Fiat
174	"""
175	# Check if the fiat conversion you want is supported
176	if not self._is_supported_fiat(fiat=fiat_symbol):	1✔
177	raise ValueError(f'The fiat {fiat_symbol} is not supported.')	1✔
178
179	# No need to convert if both crypto and fiat are the same
180	if crypto_symbol == fiat_symbol:	1✔
181	return 1.0	1✔
182
183	_gecko_id = self._get_gecko_id(crypto_symbol)	1✔
184
185	if not _gecko_id:	1✔
186	# return 0 for unsupported stake currencies (fiat-convert should not break the bot)
187	self.log_once(	1✔
188	f"unsupported crypto-symbol {crypto_symbol.upper()} - returning 0.0",
189	logger.warning)
190	return 0.0	1✔
191
192	try:	1✔
193	return float(	1✔
194	self._coingecko.get_price(
195	ids=_gecko_id,
196	vs_currencies=fiat_symbol
197	)[_gecko_id][fiat_symbol]
198	)
199	except Exception as exception:	1✔
200	logger.error("Error in _find_price: %s", exception)	1✔
201	return 0.0	1✔

freqtrade / freqtrade / 9394559170

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous