feat: Engine breakdown – Monatsbeträge je Posten, Summen, Kategorie-Filter (A13-T2)
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -0,0 +1,128 @@
|
||||
"""Aufteilung (Ausbaustufe 13): Monatsbeträge je Posten für einen Monat.
|
||||
|
||||
Rein (kein DB-Zugriff). Fixposten werden auf den Monat normalisiert
|
||||
(quartalsweise /3, jährlich /12), Einmalposten und Kreditraten zählen voll
|
||||
im Monat ihrer Fälligkeit; die Kredit-Auszahlung zählt nie. Modifikatoren
|
||||
wirken exakt wie in build_cashflows (gleiche Helfer).
|
||||
"""
|
||||
import calendar
|
||||
from collections import defaultdict
|
||||
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 STEP
|
||||
from app.engine.scenario import (CENT, PlainModifier, PlainPlanned,
|
||||
PlainRecurring, effective_end, modified_amount)
|
||||
|
||||
FREI = "Frei verfügbar"
|
||||
PCT = Decimal("0.1")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Filters:
|
||||
recurring: bool = True
|
||||
planned: bool = True
|
||||
loans: bool = True
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PlainLoan:
|
||||
id: int
|
||||
name: str
|
||||
schedule: list[Installment]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class BreakdownRow:
|
||||
name: str
|
||||
category: str
|
||||
kind: str # "recurring" | "planned" | "loan"
|
||||
rhythm: str | None
|
||||
amount: Decimal # Monatsbetrag, Kosten negativ
|
||||
recurring_id: int | None = None
|
||||
loan_id: int | None = None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Summary:
|
||||
income: Decimal
|
||||
costs: Decimal # positiv
|
||||
free: Decimal # income - costs
|
||||
deficit: bool # costs > income
|
||||
pie_on_income: bool # Tortenbasis = Einkünfte
|
||||
category_costs: list[tuple[str, Decimal]] # positiv, absteigend
|
||||
pie_rest: Decimal | None # „Frei verfügbar"-Segment
|
||||
|
||||
|
||||
def breakdown_rows(month: date, recurring: list[PlainRecurring],
|
||||
planned: list[PlainPlanned], loans: list[PlainLoan],
|
||||
modifiers: list[PlainModifier], filters: Filters, *,
|
||||
category_names: dict[int, str], loan_category: str,
|
||||
uncategorized: str) -> list[BreakdownRow]:
|
||||
first = month.replace(day=1)
|
||||
last = first.replace(day=calendar.monthrange(first.year, first.month)[1])
|
||||
|
||||
def cat(category_id: int | None) -> str:
|
||||
if category_id is None:
|
||||
return uncategorized
|
||||
return category_names.get(category_id, uncategorized)
|
||||
|
||||
out: list[BreakdownRow] = []
|
||||
if filters.recurring:
|
||||
for r in recurring:
|
||||
if r.start_date is not None and r.start_date > last:
|
||||
continue
|
||||
end = effective_end(r, modifiers)
|
||||
if end is not None and end < first:
|
||||
continue
|
||||
amount = modified_amount(r.amount, r.category_id, r.id, modifiers)
|
||||
if amount is None:
|
||||
continue
|
||||
monthly = (amount / STEP[r.rhythm]).quantize(CENT, ROUND_HALF_UP)
|
||||
out.append(BreakdownRow(r.name, cat(r.category_id), "recurring",
|
||||
r.rhythm, monthly, recurring_id=r.id))
|
||||
if filters.planned:
|
||||
for p in planned:
|
||||
if not first <= p.due <= last:
|
||||
continue
|
||||
amount = modified_amount(p.amount, p.category_id, None, modifiers)
|
||||
if amount is None:
|
||||
continue
|
||||
out.append(BreakdownRow(p.name, cat(p.category_id), "planned", None,
|
||||
amount.quantize(CENT, ROUND_HALF_UP)))
|
||||
if filters.loans:
|
||||
for loan in loans:
|
||||
total = sum((i.payment for i in loan.schedule if first <= i.due <= last),
|
||||
Decimal("0"))
|
||||
out.append(BreakdownRow(loan.name, loan_category, "loan", None,
|
||||
total, loan_id=loan.id))
|
||||
return sorted((r for r in out if r.amount != 0), key=lambda r: (r.amount, r.name))
|
||||
|
||||
|
||||
def share(amount: Decimal, basis: Decimal) -> Decimal | None:
|
||||
"""|amount| in Prozent von basis, eine Nachkommastelle; basis 0 -> None."""
|
||||
if basis == 0:
|
||||
return None
|
||||
return (abs(amount) * 100 / basis).quantize(PCT, ROUND_HALF_UP)
|
||||
|
||||
|
||||
def summarize(rows: list[BreakdownRow]) -> Summary:
|
||||
income = sum((r.amount for r in rows if r.amount > 0), Decimal("0"))
|
||||
costs = sum((-r.amount for r in rows if r.amount < 0), Decimal("0"))
|
||||
free = income - costs
|
||||
per_category: dict[str, Decimal] = defaultdict(Decimal)
|
||||
for r in rows:
|
||||
if r.amount < 0:
|
||||
per_category[r.category] += -r.amount
|
||||
category_costs = sorted(per_category.items(), key=lambda kv: (-kv[1], kv[0]))
|
||||
deficit = costs > income
|
||||
pie_on_income = income > 0 and not deficit
|
||||
pie_rest = free if pie_on_income and free > 0 else None
|
||||
return Summary(income, costs, free, deficit, pie_on_income, category_costs, pie_rest)
|
||||
|
||||
|
||||
def category_rows(rows: list[BreakdownRow], categories: set[str]) -> list[BreakdownRow]:
|
||||
"""Kostenzeilen der genannten Kategorien (Einnahmen erscheinen in keiner Torte)."""
|
||||
return [r for r in rows if r.amount < 0 and r.category in categories]
|
||||
@@ -0,0 +1,174 @@
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
|
||||
from app.engine.breakdown import (FREI, BreakdownRow, Filters, PlainLoan,
|
||||
breakdown_rows, category_rows, share,
|
||||
summarize)
|
||||
from app.engine.loans import loan_schedule
|
||||
from app.engine.scenario import PlainModifier, PlainPlanned, PlainRecurring
|
||||
|
||||
M = date(2030, 3, 1)
|
||||
NAMES = {1: "Wohnen", 2: "Einkommen", 3: "Versicherung"}
|
||||
D = Decimal
|
||||
|
||||
|
||||
def rows(month=M, recurring=(), planned=(), loans=(), modifiers=(), filters=Filters()):
|
||||
return breakdown_rows(month, list(recurring), list(planned), list(loans),
|
||||
list(modifiers), filters, category_names=NAMES,
|
||||
loan_category="System-Kreditraten",
|
||||
uncategorized="unkategorisiert")
|
||||
|
||||
|
||||
def rec(id, amount, rhythm="monthly", start=None, end=None, cat=None):
|
||||
return PlainRecurring(id, f"R{id}", D(amount), rhythm, 1, start, end, cat)
|
||||
|
||||
|
||||
def by_name(rs):
|
||||
return {r.name: r.amount for r in rs}
|
||||
|
||||
|
||||
def test_fixposten_normalisiert_auf_monat():
|
||||
rs = rows(recurring=[rec(1, "-1200.00", "yearly"), rec(2, "-100.00", "quarterly"),
|
||||
rec(3, "-100.00", "yearly"), rec(4, "-50.00")])
|
||||
assert by_name(rs) == {"R1": D("-100.00"), "R2": D("-33.33"),
|
||||
"R3": D("-8.33"), "R4": D("-50.00")}
|
||||
r2 = next(r for r in rs if r.name == "R2")
|
||||
assert (r2.kind, r2.rhythm, r2.recurring_id, r2.loan_id) == ("recurring", "quarterly", 2, None)
|
||||
|
||||
|
||||
def test_fixposten_laufzeitgrenzen():
|
||||
rs = rows(recurring=[
|
||||
rec(1, "-1.00", start=date(2030, 3, 31)), # Start am Monatsletzten: zählt
|
||||
rec(2, "-1.00", start=date(2030, 4, 1)), # Start im Folgemonat: fehlt
|
||||
rec(3, "-1.00", end=date(2030, 3, 1)), # Ende am Monatsersten: zählt
|
||||
rec(4, "-1.00", end=date(2030, 2, 28)), # Ende im Vormonat: fehlt
|
||||
])
|
||||
assert sorted(by_name(rs)) == ["R1", "R3"]
|
||||
|
||||
|
||||
def test_ende_modifikator():
|
||||
mods_vor = [PlainModifier("recurring", 1, "ende", D("0"), date(2030, 2, 28))]
|
||||
mods_in = [PlainModifier("recurring", 1, "ende", D("0"), date(2030, 3, 15))]
|
||||
assert rows(recurring=[rec(1, "-10.00")], modifiers=mods_vor) == []
|
||||
assert by_name(rows(recurring=[rec(1, "-10.00")], modifiers=mods_in)) == {"R1": D("-10.00")}
|
||||
|
||||
|
||||
def test_modifikatoren_remove_percent_absolute():
|
||||
mods = [PlainModifier("recurring", 1, "remove", D("0")),
|
||||
PlainModifier("category", 1, "percent", D("10")),
|
||||
PlainModifier("recurring", 3, "absolute", D("30"))]
|
||||
rs = rows(recurring=[rec(1, "-100.00"), rec(2, "-100.00", cat=1), rec(3, "-100.00")],
|
||||
modifiers=mods)
|
||||
assert by_name(rs) == {"R2": D("-90.00"), "R3": D("-70.00")}
|
||||
|
||||
|
||||
def test_null_zeilen_entfallen():
|
||||
mods = [PlainModifier("recurring", 1, "absolute", D("100"))]
|
||||
assert rows(recurring=[rec(1, "-50.00")], modifiers=mods) == []
|
||||
|
||||
|
||||
def test_einmalposten_voll_nur_im_monat():
|
||||
planned = [PlainPlanned("E", D("-500.00"), date(2030, 3, 31), 1),
|
||||
PlainPlanned("F", D("-500.00"), date(2030, 4, 1), 1),
|
||||
PlainPlanned("G", D("-80.00"), date(2030, 3, 1), None)]
|
||||
rs = rows(planned=planned, modifiers=[PlainModifier("category", 1, "percent", D("10"))])
|
||||
assert by_name(rs) == {"E": D("-450.00"), "G": D("-80.00")}
|
||||
e = next(r for r in rs if r.name == "E")
|
||||
assert (e.kind, e.category, e.rhythm) == ("planned", "Wohnen", None)
|
||||
assert next(r for r in rs if r.name == "G").category == "unkategorisiert"
|
||||
|
||||
|
||||
def test_kredit_annuitaet():
|
||||
k = PlainLoan(7, "Auto", loan_schedule(D("1200.00"), D("0"), 12, date(2030, 1, 15), "annuity"))
|
||||
rs = rows(loans=[k])
|
||||
assert len(rs) == 1
|
||||
assert (rs[0].name, rs[0].category, rs[0].kind, rs[0].amount, rs[0].loan_id) == \
|
||||
("Auto", "System-Kreditraten", "loan", D("-100.00"), 7)
|
||||
|
||||
|
||||
def test_kredit_endfaellig_schlussrate_voll():
|
||||
sched = loan_schedule(D("1000.00"), D("12"), 3, date(2030, 1, 10), "bullet")
|
||||
k = PlainLoan(1, "Endf", sched)
|
||||
assert by_name(rows(month=date(2030, 3, 1), loans=[k])) == {"Endf": D("-10.00")}
|
||||
assert by_name(rows(month=date(2030, 4, 1), loans=[k])) == {"Endf": D("-1010.00")}
|
||||
|
||||
|
||||
def test_auszahlung_zaehlt_nicht():
|
||||
# Auszahlung im März, erste Rate im April: März hat keine Kreditzeile.
|
||||
k = PlainLoan(1, "K", loan_schedule(D("1200.00"), D("0"), 12, date(2030, 3, 5), "annuity"))
|
||||
assert rows(loans=[k]) == []
|
||||
|
||||
|
||||
def test_kredit_rate_am_monatsende_geclampt():
|
||||
k = PlainLoan(1, "K", loan_schedule(D("1200.00"), D("0"), 12, date(2030, 1, 31), "annuity"))
|
||||
assert by_name(rows(month=date(2030, 2, 1), loans=[k])) == {"K": D("-100.00")}
|
||||
|
||||
|
||||
def test_filter_wirken_auf_einnahmen_und_ausgaben():
|
||||
recurring = [rec(1, "3000.00", cat=2), rec(2, "-100.00")]
|
||||
planned = [PlainPlanned("Bonus", D("500.00"), date(2030, 3, 10), 2)]
|
||||
k = PlainLoan(1, "K", loan_schedule(D("1200.00"), D("0"), 12, date(2030, 1, 15), "annuity"))
|
||||
assert sorted(by_name(rows(recurring=recurring, planned=planned, loans=[k],
|
||||
filters=Filters(recurring=False)))) == ["Bonus", "K"]
|
||||
assert sorted(by_name(rows(recurring=recurring, planned=planned, loans=[k],
|
||||
filters=Filters(planned=False)))) == ["K", "R1", "R2"]
|
||||
assert sorted(by_name(rows(recurring=recurring, planned=planned, loans=[k],
|
||||
filters=Filters(loans=False)))) == ["Bonus", "R1", "R2"]
|
||||
assert rows(recurring=recurring, planned=planned, loans=[k],
|
||||
filters=Filters(False, False, False)) == []
|
||||
|
||||
|
||||
def test_sortierung_groesste_kosten_zuerst():
|
||||
rs = rows(recurring=[rec(1, "3000.00"), rec(2, "-50.00"), rec(3, "-100.00")])
|
||||
assert [r.name for r in rs] == ["R3", "R2", "R1"]
|
||||
|
||||
|
||||
def test_unbekannte_kategorie_ist_unkategorisiert():
|
||||
rs = rows(recurring=[rec(1, "-1.00", cat=99), rec(2, "-1.00", cat=None)])
|
||||
assert {r.category for r in rs} == {"unkategorisiert"}
|
||||
|
||||
|
||||
def _r(name, amount, cat):
|
||||
return BreakdownRow(name, cat, "recurring", "monthly", D(amount))
|
||||
|
||||
|
||||
def test_summarize_normalfall():
|
||||
s = summarize([_r("Gehalt", "3000.00", "Einkommen"), _r("Miete", "-1000.00", "Wohnen"),
|
||||
_r("Strom", "-100.00", "Wohnen"), _r("Police", "-200.00", "Versicherung")])
|
||||
assert (s.income, s.costs, s.free) == (D("3000.00"), D("1300.00"), D("1700.00"))
|
||||
assert (s.deficit, s.pie_on_income, s.pie_rest) == (False, True, D("1700.00"))
|
||||
assert s.category_costs == [("Wohnen", D("1100.00")), ("Versicherung", D("200.00"))]
|
||||
|
||||
|
||||
def test_summarize_unterdeckung():
|
||||
s = summarize([_r("Gehalt", "1000.00", "Einkommen"), _r("Miete", "-1300.00", "Wohnen")])
|
||||
assert (s.free, s.deficit, s.pie_on_income, s.pie_rest) == (D("-300.00"), True, False, None)
|
||||
|
||||
|
||||
def test_summarize_ohne_einkuenfte():
|
||||
s = summarize([_r("Miete", "-100.00", "Wohnen")])
|
||||
assert (s.income, s.deficit, s.pie_on_income, s.pie_rest) == (D("0"), True, False, None)
|
||||
assert share(D("-100.00"), s.income) is None
|
||||
|
||||
|
||||
def test_summarize_nur_einkuenfte_und_ausgeglichen():
|
||||
s = summarize([_r("Gehalt", "500.00", "Einkommen")])
|
||||
assert (s.costs, s.category_costs, s.pie_rest) == (D("0"), [], D("500.00"))
|
||||
s = summarize([_r("Gehalt", "500.00", "Einkommen"), _r("Miete", "-500.00", "Wohnen")])
|
||||
assert (s.free, s.deficit, s.pie_on_income, s.pie_rest) == (D("0.00"), False, True, None)
|
||||
|
||||
|
||||
def test_share_rundung():
|
||||
assert share(D("-100.00"), D("3000.00")) == D("3.3")
|
||||
assert share(D("-1"), D("3")) == D("33.3")
|
||||
assert share(D("-1"), D("16")) == D("6.3") # 6.25 -> HALF_UP
|
||||
assert share(D("250"), D("1000")) == D("25.0")
|
||||
|
||||
|
||||
def test_category_rows_nur_kosten_der_kategorien():
|
||||
rs = [_r("Miete", "-1000.00", "Wohnen"), _r("Erstattung", "50.00", "Wohnen"),
|
||||
_r("Police", "-200.00", "Versicherung"), _r("Strom", "-100.00", "Wohnen")]
|
||||
assert [r.name for r in category_rows(rs, {"Wohnen"})] == ["Miete", "Strom"]
|
||||
assert [r.name for r in category_rows(rs, {"Wohnen", "Versicherung"})] == ["Miete", "Police", "Strom"]
|
||||
assert category_rows(rs, set()) == []
|
||||
assert category_rows(rs, {FREI}) == []
|
||||
Reference in New Issue
Block a user