6181253459

Committed 08 Sep 2023 06:04AM UTC coverage: 94.614% (+0.06%) from 94.556%

Build # 6181253459

Build Type

push

github-actions

Committed by

web-flow

Commit Message

Merge pull request #9159 from stash86/fix-adjust

remove old codes when we only can do partial entries

Run Details

2 of 2 new or added lines in 1 file covered. (100.0%)

19114 of 20202 relevant lines covered (94.61%)

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',
}


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
    _coingekko: 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._coingekko = CoinGeckoAPI(retries=1)
            except BaseException:
                CryptoToFiatConverter._coingekko = 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._coingekko.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: {}".format(
                    request_exception
                )
            )
        except (Exception) as exception:
            logger.error(
                f"Could not load FIAT Cryptocurrency map for the following problem: {exception}")

    def _get_gekko_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

        _gekko_id = self._get_gekko_id(crypto_symbol)

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

freqtrade / freqtrade / 6181253459

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