T1: restructure to domain/application/infrastructure + schema version 1
- Move cashflow_model/ -> domain/ - Move engine/ -> application/ - Move cli/, sync/, ai/ -> infrastructure/ - Update all imports across codebase and tests - Add version=1 to FinancialModel.to_dict() with migration hook in from_dict - Update pyproject.toml: cf entry point and packages.find Tests: 63/63 pass.
This commit is contained in:
@@ -0,0 +1,103 @@
|
||||
from copy import deepcopy
|
||||
|
||||
from domain 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,
|
||||
}
|
||||
Reference in New Issue
Block a user