rdmsandClaude Opus 5.5
1cded22e90
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
50 lines
2.1 KiB
Python
50 lines
2.1 KiB
Python
"""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)
|