from copy import deepcopy from cashflow_model import Account, FinancialModel class ForecastError(Exception): pass 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 / 100 / 12 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 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) # Compound asset values for next month for a in self.model.assets: a.value += a.value * a.growth_rate / 100 / 12 return results def _project_account(self, account: Account, months: int) -> list[dict]: balance = account.balance monthly = [] for m in range(months): income = 0.0 expenses = 0.0 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 += income - expenses monthly.append({ "balance": round(balance, 2), "income": round(income, 2), "expenses": round(expenses, 2), }) return monthly def _liability_cost(self, account: Account) -> float: total = 0.0 for liability in self.model.liabilities: interest_cost = liability.balance * liability.interest / 100 / 12 total += interest_cost liability.balance -= liability.payment - interest_cost if liability.balance < 0: liability.balance = 0 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": round(all_income, 2), "total_expenses": round(all_expenses, 2), "months": months, }