feat: Service compute_breakdown – System-Plan/Szenario, Doppelzählungs-Hinweis (A13-T4)

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

finance/app/services/breakdown.py

@@ -0,0 +1,49 @@
+"""Aufteilung (Ausbaustufe 13): Brücke DB -> engine/breakdown.py.
+
+scenario=None ist „System-Plan" (reine Planung, alle Kredite, keine
+Modifikatoren) - dieselbe Datenbasis wie load_scenario_inputs(None).
+"""
+from dataclasses import dataclass
+from datetime import date
+from decimal import Decimal
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.engine.breakdown import (BreakdownRow, Filters, PlainLoan, Summary,
+                                  breakdown_rows, summarize)
+from app.engine.loans import loan_schedule
+from app.models.tables import Category, Scenario
+from app.reserved import SYSTEM_PREFIX
+from app.services.double_count import Konflikt, all_conflicts
+from app.services.planning_chart import KREDITRATEN, UNKATEGORISIERT
+from app.services.projection_service import load_scenario_inputs
+
+PLAN_NAME = f"{SYSTEM_PREFIX}Plan"
+
+
+@dataclass(frozen=True)
+class Breakdown:
+    month: date
+    rows: list[BreakdownRow]
+    summary: Summary
+    conflicts: list[Konflikt]
+
+
+def compute_breakdown(session: Session, month: date, scenario: Scenario | None,
+                      filters: Filters) -> Breakdown:
+    inputs = load_scenario_inputs(session, scenario)
+    names = {c.id: c.name for c in session.execute(select(Category)).scalars()}
+    loans = [PlainLoan(l.id, l.name,
+                       loan_schedule(Decimal(l.principal), Decimal(l.annual_rate_pct),
+                                     l.term_months, l.payout_date, l.repayment_type))
+             for l in inputs.loans]
+    rows = breakdown_rows(month, inputs.recurring, inputs.planned, loans,
+                          inputs.modifiers, filters, category_names=names,
+                          loan_category=KREDITRATEN, uncategorized=UNKATEGORISIERT)
+    # Doppelzählungs-Hinweis nur, wenn Fixposten UND Kredit hier als Zeile stehen.
+    posten = {r.recurring_id for r in rows if r.recurring_id is not None}
+    kredite = {r.loan_id for r in rows if r.loan_id is not None}
+    conflicts = [k for k in all_conflicts(session)
+                 if k.posten_id in posten and k.kredit_id in kredite]
+    return Breakdown(month.replace(day=1), rows, summarize(rows), conflicts)

finance/tests/test_breakdown_service.py

@@ -0,0 +1,88 @@
+from datetime import date
+from decimal import Decimal
+
+from app.engine.breakdown import Filters
+from app.models.tables import (Category, Loan, PlannedItem, RecurringItem,
+                               Scenario, ScenarioLoan, ScenarioModifier,
+                               ScenarioPlannedItem)
+from app.services.breakdown import PLAN_NAME, compute_breakdown
+
+M = date(2030, 3, 1)
+D = Decimal
+
+
+def _basis(db):
+    wohnen = Category(name="Wohnen")
+    db.add(wohnen)
+    db.flush()
+    miete = RecurringItem(name="Miete", amount=D("-1000.00"), rhythm="monthly",
+                          due_day=1, category_id=wohnen.id)
+    gehalt = RecurringItem(name="Gehalt", amount=D("3000.00"), rhythm="monthly", due_day=1)
+    reise = PlannedItem(name="Reise", amount=D("-600.00"), due=date(2030, 3, 20))
+    k1 = Loan(name="K1", principal=D("1200.00"), annual_rate_pct=D("0"), term_months=12,
+              payout_date=date(2030, 1, 15), repayment_type="annuity")
+    k2 = Loan(name="K2", principal=D("2400.00"), annual_rate_pct=D("0"), term_months=12,
+              payout_date=date(2030, 1, 15), repayment_type="annuity")
+    db.add_all([miete, gehalt, reise, k1, k2])
+    db.commit()
+    return wohnen, miete, k1, k2
+
+
+def test_plan_name():
+    assert PLAN_NAME == "System-Plan"
+
+
+def test_system_plan_alle_kredite_ohne_modifikatoren(db):
+    wohnen, miete, k1, k2 = _basis(db)
+    b = compute_breakdown(db, date(2030, 3, 17), None, Filters())
+    assert b.month == M
+    assert {r.name: r.amount for r in b.rows} == {
+        "Miete": D("-1000.00"), "Gehalt": D("3000.00"), "Reise": D("-600.00"),
+        "K1": D("-100.00"), "K2": D("-200.00")}
+    assert {r.name: r.category for r in b.rows}["Miete"] == "Wohnen"
+    assert {r.name: r.category for r in b.rows}["Reise"] == "unkategorisiert"
+    assert {r.name: r.category for r in b.rows}["K1"] == "System-Kreditraten"
+    assert b.summary.income == D("3000.00") and b.summary.costs == D("1900.00")
+
+
+def test_szenario_nur_zugeordnete_kredite_und_flags(db):
+    wohnen, miete, k1, k2 = _basis(db)
+    sc = Scenario(name="S", description="", include_recurring=True, include_planned=False)
+    db.add(sc)
+    db.flush()
+    db.add_all([
+        ScenarioLoan(scenario_id=sc.id, loan_id=k1.id),
+        ScenarioModifier(scenario_id=sc.id, target_type="category", target_id=wohnen.id,
+                         kind="percent", value=D("50")),
+        ScenarioPlannedItem(scenario_id=sc.id, name="Szenario-Kauf", amount=D("-90.00"),
+                            due=date(2030, 3, 5)),
+    ])
+    db.commit()
+    b = compute_breakdown(db, M, sc, Filters())
+    got = {r.name: (r.amount, r.category) for r in b.rows}
+    assert got == {"Miete": (D("-500.00"), "Wohnen"), "Gehalt": (D("3000.00"), "unkategorisiert"),
+                   "Szenario-Kauf": (D("-90.00"), "unkategorisiert"),
+                   "K1": (D("-100.00"), "System-Kreditraten")}
+
+
+def test_szenario_ohne_fixposten(db):
+    _basis(db)
+    sc = Scenario(name="S", description="", include_recurring=False, include_planned=True)
+    db.add(sc)
+    db.commit()
+    assert [r.name for r in compute_breakdown(db, M, sc, Filters()).rows] == ["Reise"]
+
+
+def test_konflikt_nur_wenn_beide_zeilen_enthalten(db):
+    _basis(db)
+    rate = RecurringItem(name="Autorate", amount=D("-100.00"), rhythm="monthly", due_day=15)
+    db.add(rate)
+    db.commit()
+    b = compute_breakdown(db, M, None, Filters())
+    assert [(k.posten_name, k.kredit_name) for k in b.conflicts] == [("Autorate", "K1")]
+    assert compute_breakdown(db, M, None, Filters(loans=False)).conflicts == []
+    assert compute_breakdown(db, M, None, Filters(recurring=False)).conflicts == []
+    sc = Scenario(name="Ohne K1", description="")
+    db.add(sc)
+    db.commit()
+    assert compute_breakdown(db, M, sc, Filters()).conflicts == []