- All domain monetary fields now use Decimal: - Account.balance, Asset.value, Asset.growth_rate - Liability.balance, Liability.interest, Liability.payment - Transaction.amount, RecurringCashflow.amount - ExchangeRate.rate - ForecastScenario.income_multiplier, expense_multiplier, growth_multiplier - CurrencyConverter: all arithmetic in Decimal, quantize to 0.01 with ROUND_HALF_UP - ForecastService: Decimal arithmetic throughout (income, expenses, balance, growth, liability cost) - ScenarioService: Decimal multipliers, deepcopy safe with Decimal fields - assistant.py: _DecimalEncoder for json.dumps (Decimal -> str in JSON) Pydantic v2 + Decimal: - model_dump(mode='json') converts Decimal to str (JSON-safe) - model_validate() parses str back to Decimal - Round-trip preserves precision (100.50 stays 100.50) Tests: 63/63 pass.
84 lines
2.4 KiB
Python
84 lines
2.4 KiB
Python
from decimal import Decimal, ROUND_HALF_UP
|
|
|
|
from pydantic import BaseModel, Field, field_validator
|
|
|
|
CURRENCY_SYMBOLS = {
|
|
"RUB": "₽",
|
|
"USD": "$",
|
|
"EUR": "€",
|
|
"GBP": "£",
|
|
"CNY": "¥",
|
|
"JPY": "¥",
|
|
"KZT": "₸",
|
|
"UAH": "₴",
|
|
}
|
|
|
|
|
|
class ExchangeRate(BaseModel):
|
|
from_currency: str = "USD"
|
|
to_currency: str = "RUB"
|
|
rate: Decimal = Field(default=Decimal("80"))
|
|
|
|
@field_validator("rate")
|
|
@classmethod
|
|
def _rate_positive(cls, v: Decimal) -> Decimal:
|
|
if v <= 0:
|
|
raise ValueError("rate must be positive")
|
|
return v
|
|
|
|
|
|
DEFAULT_RATES: list[ExchangeRate] = [
|
|
ExchangeRate(from_currency="USD", to_currency="RUB", rate=Decimal("80")),
|
|
]
|
|
|
|
|
|
class CurrencyError(Exception):
|
|
pass
|
|
|
|
|
|
class CurrencyConverter:
|
|
def __init__(self, rates: list[ExchangeRate] | None = None):
|
|
self._rates: dict[tuple[str, str], Decimal] = {}
|
|
if rates:
|
|
for r in rates:
|
|
self.set_rate(r.from_currency, r.to_currency, r.rate)
|
|
|
|
def set_rate(
|
|
self,
|
|
from_currency: str,
|
|
to_currency: str,
|
|
rate: Decimal | float | int | str,
|
|
) -> None:
|
|
d = Decimal(str(rate)) if not isinstance(rate, Decimal) else rate
|
|
if d <= 0:
|
|
raise CurrencyError(f"Rate must be positive: {d}")
|
|
self._rates[(from_currency, to_currency)] = d
|
|
self._rates[(to_currency, from_currency)] = Decimal("1") / d
|
|
|
|
def get_rate(self, from_currency: str, to_currency: str) -> Decimal:
|
|
if from_currency == to_currency:
|
|
return Decimal("1")
|
|
try:
|
|
return self._rates[(from_currency, to_currency)]
|
|
except KeyError:
|
|
raise CurrencyError(f"No exchange rate: {from_currency} → {to_currency}")
|
|
|
|
def convert(
|
|
self,
|
|
amount: Decimal | float | int | str,
|
|
from_currency: str,
|
|
to_currency: str,
|
|
) -> Decimal:
|
|
a = Decimal(str(amount)) if not isinstance(amount, Decimal) else amount
|
|
if from_currency == to_currency:
|
|
return a
|
|
rate = self.get_rate(from_currency, to_currency)
|
|
return (a * rate).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
|
|
|
|
def get_symbol(self, currency: str) -> str:
|
|
return CURRENCY_SYMBOLS.get(currency, currency)
|
|
|
|
@classmethod
|
|
def with_defaults(cls) -> "CurrencyConverter":
|
|
return cls(DEFAULT_RATES)
|