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 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 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, r.end_date): 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)