from copy import deepcopy from decimal import Decimal, ROUND_HALF_UP from domain import Account, FinancialModel 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) def forecast_cashflow(self, months: int = 12) -> list[dict]: if months < 1: raise ForecastError("months must be >= 1") results = [] for account in self.model.accounts: monthly = self._project_account(account, months) for m in range(months): results.append({ "account": account.name, "month": m + 1, "balance": monthly[m]["balance"], "income": monthly[m]["income"], "expenses": monthly[m]["expenses"], }) # Asset growth — once per month, distributed across accounts proportionally for m in range(months): total_growth = sum( 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 _ONE for r in month_rows: share = r["balance"] / total_bal 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.value * a.growth_rate / _HUNDRED / _TWELVE return results def _project_account(self, account: Account, months: int) -> list[dict]: balance = account.balance monthly = [] for m in range(months): income = _ZERO expenses = _ZERO for t in self.model.transactions: if t.account == str(account.id): if t.amount > 0: income += t.amount else: expenses += abs(t.amount) for r in self.model.recurring: if r.amount > 0: income += r.amount else: expenses += abs(r.amount) expenses += self._liability_cost(account) balance = balance + income - expenses monthly.append({ "balance": _q(balance), "income": _q(income), "expenses": _q(expenses), }) return monthly def _liability_cost(self, account: Account) -> Decimal: total = _ZERO for liability in self.model.liabilities: interest_cost = liability.balance * liability.interest / _HUNDRED / _TWELVE total += interest_cost liability.balance = liability.balance - (liability.payment - interest_cost) if liability.balance < 0: liability.balance = _ZERO return total def summary(self, months: int = 12) -> dict: results = self.forecast_cashflow(months) if not results: return {"total_balance": 0, "total_income": 0, "total_expenses": 0, "months": months} final = results[-1] all_income = sum(r["income"] for r in results) all_expenses = sum(r["expenses"] for r in results) return { "total_balance": final["balance"], "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)