101 lines
3.2 KiB
Python
101 lines
3.2 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:
|
|
balance = account.balance
|
|
monthly = self._project_account(account, months)
|
|
for m in range(months):
|
|
balance = monthly[m]["balance"]
|
|
results.append({
|
|
"account": account.name,
|
|
"month": m + 1,
|
|
"balance": round(balance, 2),
|
|
"income": round(monthly[m]["income"], 2),
|
|
"expenses": round(monthly[m]["expenses"], 2),
|
|
})
|
|
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)
|
|
|
|
income += self._asset_income(account)
|
|
expenses += self._liability_cost(account)
|
|
|
|
balance += income - expenses
|
|
|
|
asset_growth = sum(
|
|
a.value * a.growth_rate / 12
|
|
for a in self.model.assets
|
|
)
|
|
balance += asset_growth
|
|
|
|
monthly.append({
|
|
"balance": balance,
|
|
"income": income,
|
|
"expenses": expenses,
|
|
})
|
|
return monthly
|
|
|
|
def _asset_income(self, account: Account) -> float:
|
|
return sum(
|
|
a.value * a.growth_rate / 12
|
|
for a in self.model.assets
|
|
)
|
|
|
|
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,
|
|
}
|