diff --git a/finance/app/engine/projection.py b/finance/app/engine/projection.py new file mode 100644 index 0000000..e023be9 --- /dev/null +++ b/finance/app/engine/projection.py @@ -0,0 +1,39 @@ +from collections import defaultdict +from dataclasses import dataclass +from datetime import date, timedelta +from decimal import Decimal + + +@dataclass(frozen=True) +class Projection: + series: list[tuple[date, Decimal]] + low_point: tuple[date, Decimal] + first_below_zero: date | None + first_below_threshold: date | None + + +def project(start_balance: Decimal, start_date: date, + cashflows: list[tuple[date, Decimal]], horizon_days: int, + threshold: Decimal = Decimal("0")) -> Projection: + end = start_date + timedelta(days=horizon_days) + by_day: dict[date, Decimal] = defaultdict(lambda: Decimal("0")) + for d, amount in cashflows: + if start_date < d <= end: + by_day[d] += amount + series: list[tuple[date, Decimal]] = [] + balance = start_balance + low = (start_date, start_balance) + below0: date | None = None + below_t: date | None = None + d = start_date + timedelta(days=1) + while d <= end: + balance += by_day.get(d, Decimal("0")) + series.append((d, balance)) + if balance < low[1]: + low = (d, balance) + if below0 is None and balance < 0: + below0 = d + if below_t is None and balance < threshold: + below_t = d + d += timedelta(days=1) + return Projection(series, low, below0, below_t) diff --git a/finance/app/engine/scenario.py b/finance/app/engine/scenario.py new file mode 100644 index 0000000..cc7c00d --- /dev/null +++ b/finance/app/engine/scenario.py @@ -0,0 +1,82 @@ +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) diff --git a/finance/tests/test_projection.py b/finance/tests/test_projection.py new file mode 100644 index 0000000..db31b35 --- /dev/null +++ b/finance/tests/test_projection.py @@ -0,0 +1,20 @@ +from datetime import date +from decimal import Decimal + +from app.engine.projection import project + + +def test_low_point_and_threshold(): + flows = [(date(2026, 8, 1), Decimal("-150")), (date(2026, 8, 10), Decimal("200"))] + p = project(Decimal("100"), date(2026, 7, 31), flows, horizon_days=15, + threshold=Decimal("60")) + assert p.low_point == (date(2026, 8, 1), Decimal("-50")) + assert p.first_below_zero == date(2026, 8, 1) + assert p.first_below_threshold == date(2026, 8, 1) + assert p.series[-1][1] == Decimal("150") + + +def test_no_crossing(): + p = project(Decimal("100"), date(2026, 7, 31), [], horizon_days=5) + assert p.first_below_zero is None + assert p.low_point[1] == Decimal("100") diff --git a/finance/tests/test_scenario_engine.py b/finance/tests/test_scenario_engine.py new file mode 100644 index 0000000..b9e0303 --- /dev/null +++ b/finance/tests/test_scenario_engine.py @@ -0,0 +1,37 @@ +from datetime import date +from decimal import Decimal + +from app.engine.loans import loan_schedule +from app.engine.scenario import (PlainModifier, PlainPlanned, PlainRecurring, + build_cashflows) + +W = (date(2026, 8, 1), date(2026, 10, 31)) + + +def test_recurring_with_percent_cut(): + rec = [PlainRecurring(id=1, name="Marketing", amount=Decimal("-1000"), + rhythm="monthly", due_day=5, start_date=None, + end_date=None, category_id=7)] + mods = [PlainModifier(target_type="category", target_id=7, + kind="percent", value=Decimal("50"))] + flows = build_cashflows(rec, [], [], [], mods, *W) + assert flows == [(date(2026, 8, 5), Decimal("-500.00")), + (date(2026, 9, 5), Decimal("-500.00")), + (date(2026, 10, 5), Decimal("-500.00"))] + + +def test_remove_modifier_and_planned_and_loan(): + rec = [PlainRecurring(id=2, name="Abo", amount=Decimal("-50"), + rhythm="monthly", due_day=1, start_date=None, + end_date=None, category_id=None)] + mods = [PlainModifier(target_type="recurring", target_id=2, + kind="remove", value=Decimal("0"))] + planned = [PlainPlanned(name="Steuer", amount=Decimal("-2000"), + due=date(2026, 9, 15), category_id=None)] + sched = loan_schedule(Decimal("10000"), Decimal("6"), 48, date(2026, 8, 1), "annuity") + flows = build_cashflows(rec, planned, [sched], + [(date(2026, 8, 1), Decimal("10000"))], mods, *W) + days = [d for d, _ in flows] + assert date(2026, 8, 1) in days # Kredit-Auszahlung + assert date(2026, 9, 15) in days # Einmalposten + assert all(a != Decimal("-50") for _, a in flows) # Abo entfernt