oqyude/nifodea
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

T3: Decimal for money — all monetary fields migrated from float

Browse Source 
- 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.
This commit is contained in:
oqyude
2026-10-08 16:44:59 +03:00
parent c765f3672a
commit 363d440d53
10 changed files with 118 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
application/forecast.py
+30 -18
View File
@@ -1,4 +1,5 @@
from copy import deepcopy
from decimal import Decimal, ROUND_HALF_UP
from domain import Account, FinancialModel
@@ -7,6 +8,12 @@ class ForecastError(Exception):
pass
_ZERO = Decimal("0")
_ONE = Decimal("1")
_TWELVE = Decimal("12")
_HUNDRED = Decimal("100")
class ForecastService:
def __init__(self, model: FinancialModel):
self.model = deepcopy(model)
@@ -30,19 +37,19 @@ class ForecastService:
# Asset growth — once per month, distributed across accounts proportionally
for m in range(months):
total_growth = sum(
a.value * a.growth_rate / 100 / 12
a.value * a.growth_rate / _HUNDRED / _TWELVE
for a in self.model.assets
)
month_rows = [r for r in results if r["month"] == m + 1]
total_bal = sum(r["balance"] for r in month_rows) or 1
total_bal = sum(r["balance"] for r in month_rows) or _ONE
for r in month_rows:
share = r["balance"] / total_bal
r["income"] = round(r["income"] + total_growth * share, 2)
r["balance"] = round(r["balance"] + total_growth * share, 2)
r["income"] = _q(r["income"] + total_growth * share)
r["balance"] = _q(r["balance"] + total_growth * share)
# Compound asset values for next month
for a in self.model.assets:
a.value += a.value * a.growth_rate / 100 / 12
a.value = a.value + a.value * a.growth_rate / _HUNDRED / _TWELVE
return results
@@ -50,8 +57,8 @@ class ForecastService:
balance = account.balance
monthly = []
for m in range(months):
income = 0.0
expenses = 0.0
income = _ZERO
expenses = _ZERO
for t in self.model.transactions:
if t.account == str(account.id):
@@ -68,23 +75,23 @@ class ForecastService:
expenses += self._liability_cost(account)
balance += income - expenses
balance = balance + income - expenses
monthly.append({
"balance": round(balance, 2),
"income": round(income, 2),
"expenses": round(expenses, 2),
"balance": _q(balance),
"income": _q(income),
"expenses": _q(expenses),
})
return monthly
def _liability_cost(self, account: Account) -> float:
total = 0.0
def _liability_cost(self, account: Account) -> Decimal:
total = _ZERO
for liability in self.model.liabilities:
interest_cost = liability.balance * liability.interest / 100 / 12
interest_cost = liability.balance * liability.interest / _HUNDRED / _TWELVE
total += interest_cost
liability.balance -= liability.payment - interest_cost
liability.balance = liability.balance - (liability.payment - interest_cost)
if liability.balance < 0:
liability.balance = 0
liability.balance = _ZERO
return total
def summary(self, months: int = 12) -> dict:
@@ -97,7 +104,12 @@ class ForecastService:
all_expenses = sum(r["expenses"] for r in results)
return {
"total_balance": final["balance"],
"total_income": round(all_income, 2),
"total_expenses": round(all_expenses, 2),
"total_income": _q(all_income),
"total_expenses": _q(all_expenses),
"months": months,
}
def _q(v: Decimal) -> Decimal:
"""Quantize to 2 decimal places, ROUND_HALF_UP."""
return v.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
application/scenarios.py
+24 -20
View File
@@ -1,4 +1,5 @@
from copy import deepcopy
from decimal import Decimal
from domain import FinancialModel, ForecastScenario
from application.forecast import ForecastService
@@ -11,23 +12,23 @@ class ScenarioError(Exception):
DEFAULT_SCENARIOS = {
"baseline": ForecastScenario(
name="baseline",
income_multiplier=1.0,
expense_multiplier=1.0,
growth_multiplier=1.0,
income_multiplier=Decimal("1.0"),
expense_multiplier=Decimal("1.0"),
growth_multiplier=Decimal("1.0"),
description="Базовый сценарий без изменений",
),
"optimistic": ForecastScenario(
name="optimistic",
income_multiplier=1.15,
expense_multiplier=0.95,
growth_multiplier=1.2,
income_multiplier=Decimal("1.15"),
expense_multiplier=Decimal("0.95"),
growth_multiplier=Decimal("1.2"),
description="Оптимистичный: доход +15%, расход -5%, рост активов +20%",
),
"pessimistic": ForecastScenario(
name="pessimistic",
income_multiplier=0.85,
expense_multiplier=1.1,
growth_multiplier=0.8,
income_multiplier=Decimal("0.85"),
expense_multiplier=Decimal("1.1"),
growth_multiplier=Decimal("0.8"),
description="Пессимистичный: доход -15%, расход +10%, рост активов -20%",
),
}
@@ -42,18 +43,18 @@ class ScenarioService:
for t in model.transactions:
if t.amount > 0:
t.amount *= scenario.income_multiplier
t.amount = t.amount * scenario.income_multiplier
else:
t.amount *= scenario.expense_multiplier
t.amount = t.amount * scenario.expense_multiplier
for r in model.recurring:
if r.category == "income":
r.amount *= scenario.income_multiplier
r.amount = r.amount * scenario.income_multiplier
else:
r.amount *= scenario.expense_multiplier
r.amount = r.amount * scenario.expense_multiplier
for a in model.assets:
a.growth_rate *= scenario.growth_multiplier
a.growth_rate = a.growth_rate * scenario.growth_multiplier
service = ForecastService(model)
result = service.summary(months)
@@ -69,16 +70,19 @@ class ScenarioService:
def what_if(
self,
income_mult: float = 1.0,
expense_mult: float = 1.0,
growth_mult: float = 1.0,
income_mult: Decimal | float = Decimal("1.0"),
expense_mult: Decimal | float = Decimal("1.0"),
growth_mult: Decimal | float = Decimal("1.0"),
months: int = 12,
) -> dict:
def to_d(v: Decimal | float) -> Decimal:
return v if isinstance(v, Decimal) else Decimal(str(v))
scenario = ForecastScenario(
name="what-if",
income_multiplier=income_mult,
expense_multiplier=expense_mult,
growth_multiplier=growth_mult,
income_multiplier=to_d(income_mult),
expense_multiplier=to_d(expense_mult),
growth_multiplier=to_d(growth_mult),
description=(
f"What-if: income x{income_mult}, "
f"expense x{expense_mult}, growth x{growth_mult}"