price_feed.py (3688B)
1 import requests 2 import logging 3 from abc import ABC, abstractmethod 4 5 import config 6 7 8 class PriceFeed(ABC): 9 10 def __init__(self, price_feed_url: str = None) -> None: 11 self._price_data = None 12 self._price_feed_url = price_feed_url 13 14 @abstractmethod 15 def _get_rate(self, base_currency: str) -> float: 16 pass 17 18 def _fetch_price_data(self) -> None: 19 if self._price_feed_url is not None: 20 for i in range(config.connection_attempts): 21 try: 22 r = requests.get(self._price_feed_url) 23 self._price_data = r.json() 24 return 25 except Exception as e: 26 logging.error(e) 27 logging.info( 28 "Attempting again... {}/{}...".format( 29 i + 1, config.connection_attempts) 30 ) 31 32 else: 33 raise RuntimeError("Failed to reach {}.".format( 34 self._price_feed_url)) 35 36 # used by tests 37 def set_price_data(self, price_data: dict) -> None: 38 self._price_data = price_data 39 40 def _get_btc_exchange_rate(self, base_currency: str, bitcoin_rate_multiplier: float) -> float: 41 self._fetch_price_data() 42 try: 43 rate = self._get_rate(base_currency) 44 if bitcoin_rate_multiplier != 1.00: 45 logging.debug( 46 "Adjusting BTC/{} exchange rate from {} to {} " 47 "because of rate multiplier {}.".format( 48 base_currency, rate, rate * bitcoin_rate_multiplier, 49 bitcoin_rate_multiplier)) 50 rate = rate * bitcoin_rate_multiplier 51 return rate 52 except Exception: 53 logging.error( 54 "Failed to find currency {} from {}.".format( 55 base_currency, self._price_feed_url) 56 ) 57 return None 58 59 def get_btc_value(self, base_amount: float, base_currency: str) -> float: 60 if base_currency == "BTC": 61 return float(base_amount) 62 elif base_currency == "sats": 63 return float(base_amount) / 10**8 64 65 exchange_rate = self._get_btc_exchange_rate( 66 base_currency, config.bitcoin_rate_multiplier) 67 68 if exchange_rate is not None: 69 try: 70 float_value = float(base_amount) / exchange_rate 71 except Exception as e: 72 logging.error(e) 73 74 return round(float_value, 8) 75 76 raise RuntimeError("Failed to get base currency value.") 77 78 79 class CoinDeskPriceFeed(PriceFeed): 80 81 def __init__(self, price_feed_url: str = "https://api.coindesk.com/v1/bpi/currentprice.json") -> None: 82 super().__init__(price_feed_url) 83 84 def _get_rate(self, base_currency: str) -> float: 85 return float(self._price_data["bpi"][base_currency.upper()]["rate_float"]) 86 87 88 class CoinGeckoPriceFeed(PriceFeed): 89 90 def __init__(self, price_feed_url: str = "https://api.coingecko.com/api/v3/exchange_rates") -> None: 91 super().__init__(price_feed_url) 92 93 def _get_rate(self, base_currency: str) -> float: 94 return float(self._price_data["rates"][base_currency.lower()]["value"]) 95 96 97 def get_btc_value(base_amount: float, base_currency: str) -> float: 98 if config.currency_provider == "COINDESK": 99 provider = CoinDeskPriceFeed() 100 elif config.currency_provider == "COINGECKO": 101 provider = CoinGeckoPriceFeed() 102 else: 103 raise Exception( 104 "Unsupported exchange rate provider (currency_provider): " + 105 config.currency_provider 106 ) 107 return provider.get_btc_value(base_amount, base_currency)