data-xexe/xexe/exchange_rates.py

import datetime
from decimal import Decimal
from numbers import Number
from typing import Iterable, Set, Union

from money.currency import Currency, CurrencyHelper
from money.money import Money


class ExchangeRate:

    def __init__(
        self,
        from_currency: Currency,
        to_currency: Currency,
        rate: Union[Money, Number, str],
        rate_date: datetime.date,
    ) -> None:
        self.from_currency = from_currency
        self.to_currency = to_currency
        if not isinstance(rate, Money):
            rate = Money(rate, to_currency)
        self.rate = rate
        self.rate_date = rate_date

    @property
    def descriptor(self) -> str:
        return (
            str(self.from_currency.value)
            + str(self.to_currency.value)
            + str(self.rate_date.strftime("%Y-%m-%d"))
        )

    @property
    def amount(self) -> Decimal:
        return self.rate.amount


class ExchangeRates:

    def __init__(self, rates: Union[Iterable[ExchangeRate], None] = None):

        self._rate_index = {}

        if rates is not None:
            for rate in rates:
                if not isinstance(rate, ExchangeRate):
                    raise TypeError("ExchangeRates can only hold Rates.")
                self._rate_index[rate.descriptor] = rate

    @property
    def present_currencies(self) -> Set[Currency]:
        present_currencies = set()

        for rate in self:
            present_currencies.add(rate.from_currency)
            present_currencies.add(rate.to_currency)

        return present_currencies

    @property
    def present_dates(self) -> Set[datetime.date]:
        return {rate.rate_date for rate in self}

    def add_rate(self, new_rate: ExchangeRate) -> None:
        self._rate_index[new_rate.descriptor] = new_rate

    def __iter__(self):
        return iter(list(self._rate_index.values()))

    def __len__(self):
        return len(self._rate_index)

    def __contains__(self, rate) -> bool:
        if not isinstance(rate, ExchangeRate):
            raise TypeError("ExchangeRates can only hold Rates.")

        if rate.descriptor in self._rate_index:
            return True

        return False

    def is_rate_present(self, rate: ExchangeRate) -> bool:
        if rate.descriptor in self._rate_index:
            return True

        return False

    def __getitem__(self, rate_descriptor) -> ExchangeRate:
        return self._rate_index[rate_descriptor]


def add_equal_rates(rates: ExchangeRates, overwrite: bool = False) -> ExchangeRates:

    present_currencies = rates.present_currencies
    present_dates = rates.present_dates

    for date in present_dates:
        for currency in present_currencies:
            new_rate = ExchangeRate(
                from_currency=currency,
                to_currency=currency,
                rate=Money(1, currency),
                rate_date=date,
            )
            if new_rate in rates and not overwrite:
                continue
            rates.add_rate(new_rate)

    return rates


def add_inverse_rates(rates: ExchangeRates) -> ExchangeRates:

    # Hey, I haven't thought properly what happens here if the inverse rate is
    # *already* present in the rates set. It's probably going to be fucky. I
    # would advise only running this against sets where you don't have inverse
    # rates already present.

    for rate in rates:
        inverse_rate = ExchangeRate(
            from_currency=rate.to_currency,
            to_currency=rate.from_currency,
            rate_date=rate.rate_date,
            rate=f"{1 / rate.amount:.{CurrencyHelper.decimal_precision_for_currency(rate.from_currency)}f}",
        )
        rates.add_rate(inverse_rate)

    return rates