104 lines
3.5 KiB
Python
104 lines
3.5 KiB
Python
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,
|
|
}
|