Skip to content

Commit

Permalink
refactor currency converter
Browse files Browse the repository at this point in the history
to allow multiple exchange rate getters
defaulting to user defined csv file
or hard coded exchange rates
  • Loading branch information
gr4viton committed Oct 28, 2024
1 parent 6582e2c commit 23d5b73
Showing 1 changed file with 85 additions and 188 deletions.
273 changes: 85 additions & 188 deletions ihatemoney/currency_convertor.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,17 @@
import csv
import traceback
import warnings
from abc import ABC, abstractmethod
from decimal import Decimal
from typing import Dict, Optional

from cachetools import TTLCache, cached
import requests


NO_CURRENCY = "XXX"
ExchangeRates = Dict[str, float]

class Singleton(type):
_instances = {}

Expand All @@ -14,205 +21,93 @@ def __call__(cls, *args, **kwargs):
return cls._instances[cls]


class CurrencyConverter(object, metaclass=Singleton):
# Get exchange rates
no_currency = "XXX"
class ExchangeRateGetter(ABC):

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"

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):
pass

@cached(cache=TTLCache(maxsize=1, ttl=86400))
def get_rates(self):
try:
rates = requests.get(self.api_url).json()["rates"]
except Exception:
warnings.warn(
f"Call to {self.api_url} failed: {traceback.format_exc(limit=0).strip()}"
)
# In case of any exception, let's have an empty value
"""Try to retrieve the exchange rate from various sources, defaulting to hard coded values."""
for provider in [ApiExchangeRate, UserExchangeRate, HardCodedExchangeRate]:
if rates:= provider.get_rates():
break
else:
rates = {}
rates[self.no_currency] = 1.0
rates[NO_CURRENCY] = 1.0
return rates

def get_currencies(self, with_no_currency=True):
currencies = [
"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",
]
def get_currencies(self, with_no_currency: bool=True) -> list:
currencies = list(self.get_rates.keys())
if with_no_currency:
currencies.append(self.no_currency)
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 (
source_currency == dest_currency
or source_currency == self.no_currency
Expand All @@ -223,6 +118,8 @@ def exchange_currency(self, amount, source_currency, dest_currency):
rates = self.get_rates()
source_rate = rates[source_currency]
dest_rate = rates[dest_currency]
new_amount = (float(amount) / source_rate) * dest_rate
# round to two digits because we are dealing with money
return round(new_amount, 2)
# Using Decimal to not introduce floating-point operation absolute errors
new_amount = (Decimal(amount) / Decimal(source_rate)) * Decimal(dest_rate)
# dealing with money - only round the shown amount before showing it to user
# - think about 10 * 0.0003 == 0?
return float(new_amount)

0 comments on commit 23d5b73

Please sign in to comment.