rdmsandClaude Opus 5.5
aaf05b9209
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
131 lines
5.1 KiB
Python
131 lines
5.1 KiB
Python
"""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 wie in build_cashflows (gleiche Helfer); Stichtag je Fixposten ist sein
|
|
Fälligkeitstag im Monat, je Einmalposten das Fälligkeitsdatum.
|
|
"""
|
|
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, due_in_month
|
|
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,
|
|
on=due_in_month(first, r.due_day))
|
|
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, on=p.due)
|
|
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]
|