freqtrade_origin/freqtrade/rpc/fiat_convert.py

203 lines
7.2 KiB
Python
Raw Normal View History

2018-01-28 01:33:04 +00:00
"""
Module that define classes to convert Crypto-currency to FIAT
e.g BTC to USD
"""
2021-05-22 15:15:35 +00:00
import logging
2022-10-07 12:00:04 +00:00
from datetime import datetime
from typing import Dict, List
2021-05-22 15:15:35 +00:00
from cachetools import TTLCache
2020-03-07 10:52:26 +00:00
from pycoingecko import CoinGeckoAPI
from requests.exceptions import RequestException
2018-07-04 07:31:35 +00:00
2018-06-03 11:47:36 +00:00
from freqtrade.constants import SUPPORTED_FIAT
2022-08-06 11:18:40 +00:00
from freqtrade.mixins.logging_mixin import LoggingMixin
2017-12-25 07:51:41 +00:00
2018-07-04 07:31:35 +00:00
2017-12-25 07:51:41 +00:00
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',
2023-07-21 20:49:37 +00:00
'busd': 'binance-usd',
2023-08-16 05:44:19 +00:00
'tusd': 'true-usd',
2023-10-22 16:43:35 +00:00
'usdc': 'usd-coin',
2023-12-15 08:09:31 +00:00
'btc': 'bitcoin'
}
2022-08-06 11:18:40 +00:00
class CryptoToFiatConverter(LoggingMixin):
2018-01-28 01:33:04 +00:00
"""
Main class to initiate Crypto to FIAT.
This object contains a list of pair Crypto, FIAT
This object is also a Singleton
"""
__instance = None
2020-03-07 10:52:26 +00:00
_coingekko: CoinGeckoAPI = None
_coinlistings: List[Dict] = []
2021-05-22 15:15:35 +00:00
_backoff: float = 0.0
def __new__(cls):
2020-03-07 10:52:26 +00:00
"""
2020-03-07 12:05:46 +00:00
This class is a singleton - cannot be instantiated twice.
2020-03-07 10:52:26 +00:00
"""
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:
2020-03-07 10:52:26 +00:00
CryptoToFiatConverter._coingekko = None
return CryptoToFiatConverter.__instance
def __init__(self) -> None:
# Timeout: 6h
self._pair_price: TTLCache = TTLCache(maxsize=500, ttl=6 * 60 * 60)
2022-08-06 11:18:40 +00:00
LoggingMixin.__init__(self, logger, 3600)
2018-05-13 17:46:08 +00:00
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):
2021-05-17 10:05:25 +00:00
logger.warning(
"Too many requests for CoinGecko API, backing off and trying again later.")
# Set backoff timestamp to 60 seconds in the future
2022-10-07 12:00:04 +00:00
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(
2021-08-06 22:19:36 +00:00
request_exception
)
)
2020-03-07 10:52:26 +00:00
except (Exception) as exception:
logger.error(
2020-03-07 10:52:26 +00:00
f"Could not load FIAT Cryptocurrency map for the following problem: {exception}")
2017-12-25 07:51:41 +00:00
def _get_gekko_id(self, crypto_symbol):
if not self._coinlistings:
2022-10-07 12:00:04 +00:00
if self._backoff <= datetime.now().timestamp():
self._load_cryptomap()
# Still not loaded.
if not self._coinlistings:
return None
else:
return None
2022-04-07 04:58:55 +00:00
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]]
2022-01-25 10:20:49 +00:00
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
2017-12-25 07:51:41 +00:00
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)
2017-12-25 07:51:41 +00:00
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
"""
2020-03-07 10:52:26 +00:00
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
2017-12-25 07:51:41 +00:00
symbol = f"{crypto_symbol}/{fiat_symbol}"
2021-06-25 13:45:49 +00:00
# Check if the fiat conversion you want is supported
2017-12-25 07:51:41 +00:00
if not self._is_supported_fiat(fiat=fiat_symbol):
2018-06-05 05:33:08 +00:00
raise ValueError(f'The fiat {fiat_symbol} is not supported.')
2017-12-25 07:51:41 +00:00
price = self._pair_price.get(symbol, None)
2017-12-25 07:51:41 +00:00
if not price:
price = self._find_price(
2017-12-25 07:51:41 +00:00
crypto_symbol=crypto_symbol,
fiat_symbol=fiat_symbol
2017-12-25 07:51:41 +00:00
)
if inverse and price != 0.0:
price = 1 / price
self._pair_price[symbol] = price
2017-12-25 07:51:41 +00:00
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
"""
2020-03-07 10:52:26 +00:00
return fiat.upper() in SUPPORTED_FIAT
2017-12-25 07:51:41 +00:00
def _find_price(self, crypto_symbol: str, fiat_symbol: str) -> float:
"""
Call CoinGecko API to retrieve the price in the FIAT
2020-03-07 10:52:26 +00:00
: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)
2017-12-25 07:51:41 +00:00
:return: float, price of the crypto-currency in Fiat
"""
2021-06-25 13:45:49 +00:00
# Check if the fiat conversion you want is supported
2017-12-25 07:51:41 +00:00
if not self._is_supported_fiat(fiat=fiat_symbol):
2018-06-05 05:33:08 +00:00
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)
2022-08-06 11:18:40 +00:00
self.log_once(
f"unsupported crypto-symbol {crypto_symbol.upper()} - returning 0.0",
logger.warning)
return 0.0
try:
return float(
2020-03-07 10:52:26 +00:00
self._coingekko.get_price(
ids=_gekko_id,
vs_currencies=fiat_symbol
)[_gekko_id][fiat_symbol]
)
2020-03-07 10:52:26 +00:00
except Exception as exception:
logger.error("Error in _find_price: %s", exception)
return 0.0