rdmsandClaude Haiku 4.5
80b8b369c7
- 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>
66 lines
3.1 KiB
Python
66 lines
3.1 KiB
Python
"""Planungs-Diagramm (Grafana-Panel „Planung nach Kategorie", Ausbaustufe 11).
|
|
|
|
Monatssummen je Kategorie aus der reinen Planung (ohne Szenario-Modifikatoren):
|
|
wiederkehrende Posten, Einmalposten und die Raten ALLER Kredite (Kategorie
|
|
„System-Kreditraten", Auszahlung zählt nicht). Einnahmen und Ausgaben mit
|
|
Vorzeichen, Semantik wie v_monthly_by_category. Kreditraten brauchen die
|
|
Annuitätenformel (engine/loans.py) - daher Python + Tabelle statt SQL-View.
|
|
"""
|
|
from collections import defaultdict
|
|
from datetime import date, timedelta
|
|
from decimal import Decimal
|
|
|
|
from sqlalchemy import delete, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.config import get_settings
|
|
from app.db import advisory_xact_lock
|
|
from app.engine.loans import schedule_for
|
|
from app.engine.recurrence import occurrences
|
|
from app.models.tables import (Category, Loan, PlannedItem, PlanningMonthly,
|
|
RecurringItem)
|
|
from app.reserved import SYSTEM_PREFIX
|
|
|
|
UNKATEGORISIERT = "unkategorisiert"
|
|
KREDITRATEN = f"{SYSTEM_PREFIX}Kreditraten"
|
|
PLANNING_CHART_LOCK = 110_002
|
|
|
|
|
|
def planning_monthly_totals(session: Session, today: date,
|
|
horizon_days: int) -> dict[tuple[date, str], Decimal]:
|
|
"""Fenster: 1. des laufenden Monats bis today + horizon_days. Schlüssel
|
|
(Monatserster, Kategoriename); Summen 0 entfallen."""
|
|
window_start = today.replace(day=1)
|
|
window_end = today + timedelta(days=horizon_days)
|
|
names = {c.id: c.name for c in session.execute(select(Category)).scalars()}
|
|
totals: dict[tuple[date, str], Decimal] = defaultdict(lambda: Decimal("0"))
|
|
|
|
def category(category_id: int | None) -> str:
|
|
return names.get(category_id, UNKATEGORISIERT) if category_id else UNKATEGORISIERT
|
|
|
|
for r in session.execute(select(RecurringItem)).scalars():
|
|
for d in occurrences(r.rhythm, r.due_day, window_start, window_end,
|
|
r.start_date, r.end_date):
|
|
totals[(d.replace(day=1), category(r.category_id))] += Decimal(r.amount)
|
|
for p in session.execute(select(PlannedItem)).scalars():
|
|
if window_start <= p.due <= window_end:
|
|
totals[(p.due.replace(day=1), category(p.category_id))] += Decimal(p.amount)
|
|
for loan in session.execute(select(Loan)).scalars():
|
|
for inst in schedule_for(loan):
|
|
if window_start <= inst.due <= window_end:
|
|
# payment ist bereits negativ (Abfluss), wie in build_cashflows.
|
|
totals[(inst.due.replace(day=1), KREDITRATEN)] += inst.payment
|
|
return {k: v for k, v in totals.items() if v != 0}
|
|
|
|
|
|
def refresh_planning_chart(session: Session, today: date) -> int:
|
|
"""Schreibt planning_monthly in EINER Transaktion komplett neu (Lock
|
|
gegen parallele Seitenaufrufe). Rückgabe: Anzahl geschriebener Zeilen."""
|
|
advisory_xact_lock(session, PLANNING_CHART_LOCK)
|
|
totals = planning_monthly_totals(session, today, get_settings().horizon_days)
|
|
session.execute(delete(PlanningMonthly))
|
|
session.add_all(PlanningMonthly(month=m, category=c, total=v)
|
|
for (m, c), v in sorted(totals.items()))
|
|
session.commit()
|
|
return len(totals)
|