refactor: Tilgungsplan-Helfer schedule_for, Kredit-Laufzeit 1–1200 Monate (A14-T5)

- Neuer schedule_for() in app/engine/loans.py: vereinheitlicht Tilgungsplan-Erstellung
- Laufzeit-Validierung in LoanIn und LoanPatch: 1–1200 Monate
- Aufrufstellen aktualisiert (routers/planning.py, services/breakdown.py,
  services/projection_service.py, services/planning_chart.py)
- Keine direkten loan_schedule(Decimal(...)) Aufrufe außerhalb engine/loans.py mehr
- Tests für schedule_for und Laufzeit-Grenzen (1, 1200, sowie 0/-5/1201 Ablehnung)

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
2026-09-30 13:10:55 +02:00
co-authored by Claude Haiku 4.5
parent cd7ba6aa6a
commit 80b8b369c7
7 changed files with 60 additions and 19 deletions
+8
View File
@@ -54,3 +54,11 @@ def loan_schedule(principal: Decimal, annual_rate_pct: Decimal, term_months: int
-(interest + principal_part), interest, -(interest + principal_part), interest,
principal_part, remaining.quantize(CENT))) principal_part, remaining.quantize(CENT)))
return plan return plan
def schedule_for(loan) -> list[Installment]:
"""Tilgungsplan eines Kredit-Objekts (ORM-Loan oder gleichartig mit
principal, annual_rate_pct, term_months, payout_date, repayment_type).
Ersetzt die wiederholte Feld-für-Feld-Übergabe an loan_schedule (A14)."""
return loan_schedule(Decimal(loan.principal), Decimal(loan.annual_rate_pct),
loan.term_months, loan.payout_date, loan.repayment_type)
+4 -5
View File
@@ -9,7 +9,7 @@ from sqlalchemy.orm import Session
from app.auth import require_auth from app.auth import require_auth
from app.db import get_session from app.db import get_session
from app.engine.loans import loan_schedule from app.engine.loans import schedule_for
from app.models.tables import (Category, Loan, PlannedItem, RecurringItem, from app.models.tables import (Category, Loan, PlannedItem, RecurringItem,
ScenarioLoan) ScenarioLoan)
from app.services.double_count import (CODE as DOPPELZAEHLUNG_CODE, Konflikt, from app.services.double_count import (CODE as DOPPELZAEHLUNG_CODE, Konflikt,
@@ -227,7 +227,7 @@ class LoanIn(BaseModel):
name: str name: str
principal: Decimal principal: Decimal
annual_rate_pct: Decimal annual_rate_pct: Decimal
term_months: int term_months: int = Field(ge=1, le=1200)
payout_date: date payout_date: date
repayment_type: Literal["annuity", "bullet"] = "annuity" repayment_type: Literal["annuity", "bullet"] = "annuity"
@@ -245,7 +245,7 @@ class LoanPatch(BaseModel):
name: str | None = None name: str | None = None
principal: Decimal | None = None principal: Decimal | None = None
annual_rate_pct: Decimal | None = None annual_rate_pct: Decimal | None = None
term_months: int | None = None term_months: int | None = Field(default=None, ge=1, le=1200)
payout_date: date | None = None payout_date: date | None = None
repayment_type: Literal["annuity", "bullet"] | None = None repayment_type: Literal["annuity", "bullet"] | None = None
doppelzaehlung_bestaetigt: bool = False doppelzaehlung_bestaetigt: bool = False
@@ -318,8 +318,7 @@ def get_loan_schedule(loan_id: int, session: Session = Depends(get_session)):
loan = session.get(Loan, loan_id) loan = session.get(Loan, loan_id)
if loan is None: if loan is None:
raise HTTPException(404, "Kredit nicht gefunden") raise HTTPException(404, "Kredit nicht gefunden")
plan = loan_schedule(Decimal(loan.principal), Decimal(loan.annual_rate_pct), plan = schedule_for(loan)
loan.term_months, loan.payout_date, loan.repayment_type)
return [InstallmentOut(due=i.due, payment=i.payment, interest=i.interest, return [InstallmentOut(due=i.due, payment=i.payment, interest=i.interest,
principal=i.principal, remaining=i.remaining) principal=i.principal, remaining=i.remaining)
for i in plan] for i in plan]
+2 -6
View File
@@ -5,14 +5,13 @@ Modifikatoren) - dieselbe Datenbasis wie load_scenario_inputs(None).
""" """
from dataclasses import dataclass from dataclasses import dataclass
from datetime import date from datetime import date
from decimal import Decimal
from sqlalchemy import select from sqlalchemy import select
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from app.engine.breakdown import (BreakdownRow, Filters, PlainLoan, Summary, from app.engine.breakdown import (BreakdownRow, Filters, PlainLoan, Summary,
breakdown_rows, summarize) breakdown_rows, summarize)
from app.engine.loans import loan_schedule from app.engine.loans import schedule_for
from app.models.tables import Category, Scenario from app.models.tables import Category, Scenario
from app.reserved import SYSTEM_PREFIX from app.reserved import SYSTEM_PREFIX
from app.services.double_count import Konflikt, all_conflicts from app.services.double_count import Konflikt, all_conflicts
@@ -34,10 +33,7 @@ def compute_breakdown(session: Session, month: date, scenario: Scenario | None,
filters: Filters) -> Breakdown: filters: Filters) -> Breakdown:
inputs = load_scenario_inputs(session, scenario) inputs = load_scenario_inputs(session, scenario)
names = {c.id: c.name for c in session.execute(select(Category)).scalars()} names = {c.id: c.name for c in session.execute(select(Category)).scalars()}
loans = [PlainLoan(l.id, l.name, loans = [PlainLoan(l.id, l.name, schedule_for(l)) for l in inputs.loans]
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, rows = breakdown_rows(month, inputs.recurring, inputs.planned, loans,
inputs.modifiers, filters, category_names=names, inputs.modifiers, filters, category_names=names,
loan_category=KREDITRATEN, uncategorized=UNKATEGORISIERT) loan_category=KREDITRATEN, uncategorized=UNKATEGORISIERT)
+2 -4
View File
@@ -15,7 +15,7 @@ from sqlalchemy.orm import Session
from app.config import get_settings from app.config import get_settings
from app.db import advisory_xact_lock from app.db import advisory_xact_lock
from app.engine.loans import loan_schedule from app.engine.loans import schedule_for
from app.engine.recurrence import occurrences from app.engine.recurrence import occurrences
from app.models.tables import (Category, Loan, PlannedItem, PlanningMonthly, from app.models.tables import (Category, Loan, PlannedItem, PlanningMonthly,
RecurringItem) RecurringItem)
@@ -46,9 +46,7 @@ def planning_monthly_totals(session: Session, today: date,
if window_start <= p.due <= window_end: if window_start <= p.due <= window_end:
totals[(p.due.replace(day=1), category(p.category_id))] += Decimal(p.amount) totals[(p.due.replace(day=1), category(p.category_id))] += Decimal(p.amount)
for loan in session.execute(select(Loan)).scalars(): for loan in session.execute(select(Loan)).scalars():
for inst in loan_schedule(Decimal(loan.principal), Decimal(loan.annual_rate_pct), for inst in schedule_for(loan):
loan.term_months, loan.payout_date,
loan.repayment_type):
if window_start <= inst.due <= window_end: if window_start <= inst.due <= window_end:
# payment ist bereits negativ (Abfluss), wie in build_cashflows. # payment ist bereits negativ (Abfluss), wie in build_cashflows.
totals[(inst.due.replace(day=1), KREDITRATEN)] += inst.payment totals[(inst.due.replace(day=1), KREDITRATEN)] += inst.payment
+2 -4
View File
@@ -8,7 +8,7 @@ from sqlalchemy.orm import Session
from app.config import get_settings from app.config import get_settings
from app.db import advisory_xact_lock from app.db import advisory_xact_lock
from app.engine.loans import loan_schedule from app.engine.loans import schedule_for
from app.engine.projection import project from app.engine.projection import project
from app.engine.scenario import (PlainModifier, PlainPlanned, PlainRecurring, from app.engine.scenario import (PlainModifier, PlainPlanned, PlainRecurring,
build_cashflows) build_cashflows)
@@ -76,9 +76,7 @@ def run_projection(session: Session, scenario: Scenario, horizon_days: int,
start_date: date) -> ProjectionResult: start_date: date) -> ProjectionResult:
end = start_date + timedelta(days=horizon_days) end = start_date + timedelta(days=horizon_days)
inputs = load_scenario_inputs(session, scenario) inputs = load_scenario_inputs(session, scenario)
schedules = [loan_schedule(Decimal(l.principal), Decimal(l.annual_rate_pct), schedules = [schedule_for(l) for l in inputs.loans]
l.term_months, l.payout_date, l.repayment_type)
for l in inputs.loans]
payouts = [(l.payout_date, Decimal(l.principal)) for l in inputs.loans] payouts = [(l.payout_date, Decimal(l.principal)) for l in inputs.loans]
flows = build_cashflows(inputs.recurring, inputs.planned, schedules, payouts, flows = build_cashflows(inputs.recurring, inputs.planned, schedules, payouts,
inputs.modifiers, start_date, end) inputs.modifiers, start_date, end)
+18
View File
@@ -27,3 +27,21 @@ def test_bullet_loan():
assert all(i.principal == 0 for i in plan[:-1]) assert all(i.principal == 0 for i in plan[:-1])
assert plan[-1].principal == Decimal("10000") assert plan[-1].principal == Decimal("10000")
assert plan[-1].remaining == Decimal("0.00") assert plan[-1].remaining == Decimal("0.00")
def test_schedule_for_entspricht_loan_schedule():
from types import SimpleNamespace
from app.engine.loans import schedule_for
kredit = SimpleNamespace(principal="1200.00", annual_rate_pct="3.00", term_months=12,
payout_date=date(2030, 1, 31), repayment_type="annuity")
assert schedule_for(kredit) == loan_schedule(Decimal("1200.00"), Decimal("3.00"), 12,
date(2030, 1, 31), "annuity")
def test_keine_manuelle_tilgungsplan_konvertierung_mehr():
from pathlib import Path
app_dir = Path(__file__).resolve().parent.parent / "app"
engine_loans = app_dir / "engine" / "loans.py"
treffer = [str(p.relative_to(app_dir)) for p in app_dir.rglob("*.py")
if p != engine_loans and "loan_schedule(Decimal(" in p.read_text()]
assert treffer == []
+24
View File
@@ -2,6 +2,8 @@ from datetime import date
from decimal import Decimal from decimal import Decimal
from itertools import count from itertools import count
import pytest
from app.models.tables import Account, Category, RecurringItem, Transaction from app.models.tables import Account, Category, RecurringItem, Transaction
from app.services.suggestions import _Series, _try_merge, suggest_recurring from app.services.suggestions import _Series, _try_merge, suggest_recurring
@@ -368,3 +370,25 @@ def test_suggest_volatilitaetshinweis_bei_abgetrennter_neuester_buchung(db):
assert len(out) == 1 assert len(out) == 1
assert out[0]["amount"] == Decimal("-200.00") # Betrag NICHT durch die 600er-Buchung verfaelscht assert out[0]["amount"] == Decimal("-200.00") # Betrag NICHT durch die 600er-Buchung verfaelscht
assert "schwanken" in out[0]["hinweis"].lower() assert "schwanken" in out[0]["hinweis"].lower()
# ----------------------------------------------------------- Ausbaustufe 14: Task 5
KREDIT = {"name": "Laufzeit-Test", "principal": "1000.00", "annual_rate_pct": "2.00",
"payout_date": "2030-01-01", "repayment_type": "annuity"}
@pytest.mark.parametrize("laufzeit", [0, -5, 1201])
def test_kredit_laufzeit_ausserhalb_1_bis_1200(client, laufzeit):
r = client.post("/api/loans", headers=H, json={**KREDIT, "term_months": laufzeit})
assert r.status_code == 422
def test_kredit_laufzeit_grenzen_und_patch(client):
a = client.post("/api/loans", headers=H, json={**KREDIT, "term_months": 1})
b = client.post("/api/loans", headers=H, json={**KREDIT, "name": "L2", "term_months": 1200})
assert (a.status_code, b.status_code) == (201, 201)
url = f"/api/loans/{a.json()['id']}"
assert client.patch(url, headers=H, json={"term_months": 0}).status_code == 422
assert client.patch(url, headers=H, json={"term_months": 1201}).status_code == 422
r = client.patch(url, headers=H, json={"term_months": 1200})
assert r.status_code == 200 and r.json()["term_months"] == 1200