refactor currency converter

to allow multiple exchange rate getters
defaulting to user defined csv file
or hard coded exchange rates
This commit is contained in:
gr4viton 2024-10-28 22:59:10 +01:00
parent 6582e2c0c3
commit 23d5b73ae1

View file

@ -1,10 +1,17 @@
import csv
import traceback import traceback
import warnings import warnings
from abc import ABC, abstractmethod
from decimal import Decimal
from typing import Dict, Optional
from cachetools import TTLCache, cached from cachetools import TTLCache, cached
import requests import requests
NO_CURRENCY = "XXX"
ExchangeRates = Dict[str, float]
class Singleton(type): class Singleton(type):
_instances = {} _instances = {}
@ -14,205 +21,93 @@ class Singleton(type):
return cls._instances[cls] return cls._instances[cls]
class CurrencyConverter(object, metaclass=Singleton): class ExchangeRateGetter(ABC):
# Get exchange rates
no_currency = "XXX" def get_rates(self) -> Optional[ExchangeRates]:
"""Method to retrieve a list of currency conversion rates.
Returns:
currencies: dict - key is a three-letter currency, value is float of conversion to base currency
"""
try:
return self._get_rates()
except Exception:
warnings.warn(
f"Exchange rate getter failed - {traceback.format_exc(limit=0).strip()}"
)
@abstractmethod
def _get_rates(self) -> Optional[ExchangeRates]:
"""Actual implementation of the exchange rate getter."""
raise NotImplementedError
class ApiExchangeRate(ExchangeRateGetter):
api_url = "https://api.exchangerate.host/latest?base=USD" api_url = "https://api.exchangerate.host/latest?base=USD"
def _get_rates(self) -> Optional[ExchangeRates]:
return requests.get(self.api_url).json()["rates"] # TODO not working currently probably
class UserExchangeRate(ExchangeRateGetter):
user_csv_file = "path/to/file.csv"
def _get_rates(self) -> Optional[ExchangeRates]:
"""Get rates from user defined csv.
The user_csv_file should contain the currency conversions to "USD" without 1 header row
Example:
```
currency_code,fx_rate_to_USD
CZK,25.0
...
```
TODO: make it work bi-directionally
TODO: document for the user where to place the file
"""
reader = csv.reader(self.user_csv_file)
rates = {}
for row in reader:
from_currency = row[0]
rate = float(row[1])
# TODO add validation and exception handling for typos
rates[from_currency] = rate
return rates
class HardCodedExchangeRate(ExchangeRateGetter):
def _get_rates(self) -> Optional[dict]:
return {"USD": 1.0} # TODO fill in more
class CurrencyConverter(object, metaclass=Singleton):
no_currency = NO_CURRENCY
def __init__(self): def __init__(self):
pass pass
@cached(cache=TTLCache(maxsize=1, ttl=86400)) @cached(cache=TTLCache(maxsize=1, ttl=86400))
def get_rates(self): def get_rates(self):
try: """Try to retrieve the exchange rate from various sources, defaulting to hard coded values."""
rates = requests.get(self.api_url).json()["rates"] for provider in [ApiExchangeRate, UserExchangeRate, HardCodedExchangeRate]:
except Exception: if rates:= provider.get_rates():
warnings.warn( break
f"Call to {self.api_url} failed: {traceback.format_exc(limit=0).strip()}" else:
)
# In case of any exception, let's have an empty value
rates = {} rates = {}
rates[self.no_currency] = 1.0 rates[NO_CURRENCY] = 1.0
return rates return rates
def get_currencies(self, with_no_currency=True): def get_currencies(self, with_no_currency: bool=True) -> list:
currencies = [ currencies = list(self.get_rates.keys())
"AED",
"AFN",
"ALL",
"AMD",
"ANG",
"AOA",
"ARS",
"AUD",
"AWG",
"AZN",
"BAM",
"BBD",
"BDT",
"BGN",
"BHD",
"BIF",
"BMD",
"BND",
"BOB",
"BRL",
"BSD",
"BTC",
"BTN",
"BWP",
"BYN",
"BZD",
"CAD",
"CDF",
"CHF",
"CLF",
"CLP",
"CNH",
"CNY",
"COP",
"CRC",
"CUC",
"CUP",
"CVE",
"CZK",
"DJF",
"DKK",
"DOP",
"DZD",
"EGP",
"ERN",
"ETB",
"EUR",
"FJD",
"FKP",
"GBP",
"GEL",
"GGP",
"GHS",
"GIP",
"GMD",
"GNF",
"GTQ",
"GYD",
"HKD",
"HNL",
"HRK",
"HTG",
"HUF",
"IDR",
"ILS",
"IMP",
"INR",
"IQD",
"IRR",
"ISK",
"JEP",
"JMD",
"JOD",
"JPY",
"KES",
"KGS",
"KHR",
"KMF",
"KPW",
"KRW",
"KWD",
"KYD",
"KZT",
"LAK",
"LBP",
"LKR",
"LRD",
"LSL",
"LYD",
"MAD",
"MDL",
"MGA",
"MKD",
"MMK",
"MNT",
"MOP",
"MRU",
"MUR",
"MVR",
"MWK",
"MXN",
"MYR",
"MZN",
"NAD",
"NGN",
"NIO",
"NOK",
"NPR",
"NZD",
"OMR",
"PAB",
"PEN",
"PGK",
"PHP",
"PKR",
"PLN",
"PYG",
"QAR",
"RON",
"RSD",
"RUB",
"RWF",
"SAR",
"SBD",
"SCR",
"SDG",
"SEK",
"SGD",
"SHP",
"SLL",
"SOS",
"SRD",
"SSP",
"STD",
"STN",
"SVC",
"SYP",
"SZL",
"THB",
"TJS",
"TMT",
"TND",
"TOP",
"TRY",
"TTD",
"TWD",
"TZS",
"UAH",
"UGX",
"USD",
"UYU",
"UZS",
"VEF",
"VES",
"VND",
"VUV",
"WST",
"XAF",
"XAG",
"XAU",
"XCD",
"XDR",
"XOF",
"XPD",
"XPF",
"XPT",
"YER",
"ZAR",
"ZMW",
"ZWL",
]
if with_no_currency: if with_no_currency:
currencies.append(self.no_currency) currencies.append(self.no_currency)
return currencies return currencies
def exchange_currency(self, amount, source_currency, dest_currency): def exchange_currency(self, amount: float, source_currency: str, dest_currency: str) -> float:
"""Return the money amount converted from source_currency to dest_currency."""
if ( if (
source_currency == dest_currency source_currency == dest_currency
or source_currency == self.no_currency or source_currency == self.no_currency
@ -223,6 +118,8 @@ class CurrencyConverter(object, metaclass=Singleton):
rates = self.get_rates() rates = self.get_rates()
source_rate = rates[source_currency] source_rate = rates[source_currency]
dest_rate = rates[dest_currency] dest_rate = rates[dest_currency]
new_amount = (float(amount) / source_rate) * dest_rate # Using Decimal to not introduce floating-point operation absolute errors
# round to two digits because we are dealing with money new_amount = (Decimal(amount) / Decimal(source_rate)) * Decimal(dest_rate)
return round(new_amount, 2) # dealing with money - only round the shown amount before showing it to user
# - think about 10 * 0.0003 == 0?
return float(new_amount)