97 lines
3.2 KiB
Python
97 lines
3.2 KiB
Python
from dataclasses import dataclass
|
|
from datetime import date
|
|
from decimal import ROUND_HALF_UP, Decimal
|
|
|
|
from app.engine.loans import Installment
|
|
from app.engine.recurrence import occurrences
|
|
|
|
CENT = Decimal("0.01")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PlainRecurring:
|
|
id: int
|
|
name: str
|
|
amount: Decimal
|
|
rhythm: str
|
|
due_day: int
|
|
start_date: date | None
|
|
end_date: date | None
|
|
category_id: int | None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PlainPlanned:
|
|
name: str
|
|
amount: Decimal
|
|
due: date
|
|
category_id: int | None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PlainModifier:
|
|
target_type: str
|
|
target_id: int
|
|
kind: str
|
|
value: Decimal
|
|
end_date: date | None = None
|
|
|
|
|
|
def _modified(amount: Decimal, category_id: int | None, recurring_id: int | None,
|
|
modifiers: list[PlainModifier]) -> Decimal | None:
|
|
for m in modifiers:
|
|
hit = ((m.target_type == "category" and category_id == m.target_id)
|
|
or (m.target_type == "recurring" and recurring_id == m.target_id))
|
|
if not hit:
|
|
continue
|
|
if m.kind == "remove":
|
|
return None
|
|
if m.kind == "percent":
|
|
amount = (amount * (Decimal(100) - m.value) / Decimal(100)).quantize(CENT, ROUND_HALF_UP)
|
|
elif m.kind == "absolute":
|
|
if amount < 0:
|
|
amount = min(Decimal("0"), amount + m.value)
|
|
else:
|
|
amount = max(Decimal("0"), amount - m.value)
|
|
return amount
|
|
|
|
|
|
def _effective_end(r: PlainRecurring, modifiers: list[PlainModifier]) -> date | None:
|
|
"""Fruehestes Ende aus eigenem end_date und allen treffenden ende-Modifikatoren."""
|
|
end = r.end_date
|
|
for m in modifiers:
|
|
if m.kind != "ende" or m.end_date is None:
|
|
continue
|
|
hit = ((m.target_type == "category" and r.category_id == m.target_id)
|
|
or (m.target_type == "recurring" and r.id == m.target_id))
|
|
if hit:
|
|
end = m.end_date if end is None else min(end, m.end_date)
|
|
return end
|
|
|
|
|
|
def build_cashflows(recurring: list[PlainRecurring], planned: list[PlainPlanned],
|
|
loan_schedules: list[list[Installment]],
|
|
loan_payouts: list[tuple[date, Decimal]],
|
|
modifiers: list[PlainModifier],
|
|
window_start: date, window_end: date) -> list[tuple[date, Decimal]]:
|
|
flows: list[tuple[date, Decimal]] = []
|
|
for r in recurring:
|
|
amount = _modified(r.amount, r.category_id, r.id, modifiers)
|
|
if amount is None:
|
|
continue
|
|
for d in occurrences(r.rhythm, r.due_day, window_start, window_end,
|
|
r.start_date, _effective_end(r, modifiers)):
|
|
flows.append((d, amount))
|
|
for p in planned:
|
|
amount = _modified(p.amount, p.category_id, None, modifiers)
|
|
if amount is not None and window_start <= p.due <= window_end:
|
|
flows.append((p.due, amount))
|
|
for d, payout in loan_payouts:
|
|
if window_start <= d <= window_end:
|
|
flows.append((d, payout))
|
|
for sched in loan_schedules:
|
|
for inst in sched:
|
|
if window_start <= inst.due <= window_end:
|
|
flows.append((inst.due, inst.payment))
|
|
return sorted(flows)
|