testing new features

cashflow_model/__init__.py

@@ -1,5 +1,6 @@
 from cashflow_model.account import Account
 from cashflow_model.asset import Asset
+from cashflow_model.currency import CURRENCY_SYMBOLS, CurrencyConverter, CurrencyError, ExchangeRate
 from cashflow_model.liability import Liability
 from cashflow_model.model import FinancialModel
 from cashflow_model.recurring import RecurringCashflow
@@ -14,4 +15,8 @@ __all__ = [
     "Liability",
     "ForecastScenario",
     "FinancialModel",
+    "ExchangeRate",
+    "CurrencyConverter",
+    "CurrencyError",
+    "CURRENCY_SYMBOLS",
 ]

cashflow_model/currency.py

@@ -0,0 +1,79 @@
+from dataclasses import dataclass
+
+CURRENCY_SYMBOLS = {
+    "RUB": "₽",
+    "USD": "$",
+    "EUR": "€",
+    "GBP": "£",
+    "CNY": "¥",
+    "JPY": "¥",
+    "KZT": "₸",
+    "UAH": "₴",
+}
+
+
+@dataclass
+class ExchangeRate:
+    from_currency: str = "USD"
+    to_currency: str = "RUB"
+    rate: float = 80.0
+
+    def to_dict(self) -> dict:
+        return {
+            "from_currency": self.from_currency,
+            "to_currency": self.to_currency,
+            "rate": self.rate,
+        }
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "ExchangeRate":
+        return cls(
+            from_currency=data.get("from_currency", "USD"),
+            to_currency=data.get("to_currency", "RUB"),
+            rate=data.get("rate", 80.0),
+        )
+
+
+DEFAULT_RATES: list[ExchangeRate] = [
+    ExchangeRate(from_currency="USD", to_currency="RUB", rate=80.0),
+]
+
+
+class CurrencyError(Exception):
+    pass
+
+
+class CurrencyConverter:
+    def __init__(self, rates: list[ExchangeRate] | None = None):
+        self._rates: dict[tuple[str, str], float] = {}
+        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: float) -> None:
+        if rate <= 0:
+            raise CurrencyError(f"Rate must be positive: {rate}")
+        self._rates[(from_currency, to_currency)] = rate
+        inverse = 1.0 / rate
+        self._rates[(to_currency, from_currency)] = inverse
+
+    def get_rate(self, from_currency: str, to_currency: str) -> float:
+        if from_currency == to_currency:
+            return 1.0
+        try:
+            return self._rates[(from_currency, to_currency)]
+        except KeyError:
+            raise CurrencyError(f"No exchange rate: {from_currency} → {to_currency}")
+
+    def convert(self, amount: float, from_currency: str, to_currency: str) -> float:
+        if from_currency == to_currency:
+            return amount
+        rate = self.get_rate(from_currency, to_currency)
+        return round(amount * rate, 2)
+
+    def get_symbol(self, currency: str) -> str:
+        return CURRENCY_SYMBOLS.get(currency, currency)
+
+    @classmethod
+    def with_defaults(cls) -> "CurrencyConverter":
+        return cls(DEFAULT_RATES)

cashflow_model/model.py

@@ -4,6 +4,7 @@ from pathlib import Path
 
 from cashflow_model.account import Account
 from cashflow_model.asset import Asset
+from cashflow_model.currency import DEFAULT_RATES, ExchangeRate
 from cashflow_model.liability import Liability
 from cashflow_model.recurring import RecurringCashflow
 from cashflow_model.scenario import ForecastScenario
@@ -12,32 +13,38 @@ from cashflow_model.transaction import Transaction
 
 @dataclass
 class FinancialModel:
+    base_currency: str = "RUB"
     accounts: list[Account] = field(default_factory=list)
     transactions: list[Transaction] = field(default_factory=list)
     recurring: list[RecurringCashflow] = field(default_factory=list)
     assets: list[Asset] = field(default_factory=list)
     liabilities: list[Liability] = field(default_factory=list)
     scenarios: list[ForecastScenario] = field(default_factory=list)
+    exchange_rates: list[ExchangeRate] = field(default_factory=lambda: DEFAULT_RATES.copy())
 
     def to_dict(self) -> dict:
         return {
+            "base_currency": self.base_currency,
             "accounts": [a.to_dict() for a in self.accounts],
             "transactions": [t.to_dict() for t in self.transactions],
             "recurring": [r.to_dict() for r in self.recurring],
             "assets": [a.to_dict() for a in self.assets],
             "liabilities": [li.to_dict() for li in self.liabilities],
             "scenarios": [s.to_dict() for s in self.scenarios],
+            "exchange_rates": [r.to_dict() for r in self.exchange_rates],
         }
 
     @classmethod
     def from_dict(cls, data: dict) -> "FinancialModel":
         return cls(
+            base_currency=data.get("base_currency", "RUB"),
             accounts=[Account.from_dict(a) for a in data.get("accounts", [])],
             transactions=[Transaction.from_dict(t) for t in data.get("transactions", [])],
             recurring=[RecurringCashflow.from_dict(r) for r in data.get("recurring", [])],
             assets=[Asset.from_dict(a) for a in data.get("assets", [])],
             liabilities=[Liability.from_dict(li) for li in data.get("liabilities", [])],
             scenarios=[ForecastScenario.from_dict(s) for s in data.get("scenarios", [])],
+            exchange_rates=[ExchangeRate.from_dict(r) for r in data.get("exchange_rates", [])],
         )
 
     def save(self, path: str | Path) -> None: