from dataclasses import dataclass from datetime import date, timedelta from datetime import datetime, timezone from decimal import Decimal from sqlalchemy import delete, func, select from sqlalchemy.orm import Session from app.config import get_settings from app.db import advisory_xact_lock from app.engine.loans import loan_schedule from app.engine.projection import project from app.engine.scenario import (PlainModifier, PlainPlanned, PlainRecurring, build_cashflows) from app.models.tables import (Loan, PlannedItem, ProjectionPoint, ProjectionResult, RecurringItem, Scenario, ScenarioLoan, ScenarioModifier, ScenarioPlannedItem, ScenarioRecurringItem) from app.services.balances import total_balance @dataclass(frozen=True) class ScenarioInputs: """Eingaben einer Szenario-Rechnung (Projektion wie Aufteilung).""" recurring: list[PlainRecurring] planned: list[PlainPlanned] loans: list[Loan] modifiers: list[PlainModifier] def load_scenario_inputs(session: Session, scenario: Scenario | None) -> ScenarioInputs: """Lädt die Eingaben eines Szenarios. scenario=None ist die reine Planung („System-Plan"): alle Fixposten, alle globalen Einmalposten, alle Kredite, keine Modifikatoren, keine Szenario-Einmalposten und keine szenario-eigenen wiederkehrenden Zahlungen.""" recurring = [] if scenario is None or scenario.include_recurring: recurring = [PlainRecurring(r.id, r.name, Decimal(r.amount), r.rhythm, r.due_day, r.start_date, r.end_date, r.category_id) for r in session.execute( select(RecurringItem).order_by(RecurringItem.id)).scalars()] planned = [] if scenario is None or scenario.include_planned: planned = [PlainPlanned(p.name, Decimal(p.amount), p.due, p.category_id) for p in session.execute( select(PlannedItem).order_by(PlannedItem.id)).scalars()] if scenario is None: loans = list(session.execute(select(Loan).order_by(Loan.id)).scalars()) return ScenarioInputs(recurring, planned, loans, []) # Szenario-eigene Zahlungen zählen IMMER für dieses Szenario - # include_recurring/include_planned steuern nur die globalen Posten. # id=None: Posten-Einträge (target_type="recurring") treffen sie nie, # die IDs zweier Tabellen könnten sonst kollidieren. recurring += [PlainRecurring(None, r.name, Decimal(r.amount), r.rhythm, r.due_day, r.start_date, r.end_date, r.category_id) for r in session.execute(select(ScenarioRecurringItem).where( ScenarioRecurringItem.scenario_id == scenario.id) .order_by(ScenarioRecurringItem.id)).scalars()] planned += [PlainPlanned(p.name, Decimal(p.amount), p.due, p.category_id) for p in session.execute(select(ScenarioPlannedItem).where( ScenarioPlannedItem.scenario_id == scenario.id) .order_by(ScenarioPlannedItem.id)).scalars()] loans = list(session.execute( select(Loan).join(ScenarioLoan, ScenarioLoan.loan_id == Loan.id) .where(ScenarioLoan.scenario_id == scenario.id) .order_by(Loan.id)).scalars()) modifiers = [PlainModifier(m.target_type, m.target_id, m.kind, Decimal(m.value), m.end_date, m.start_date) for m in session.execute(select(ScenarioModifier).where( ScenarioModifier.scenario_id == scenario.id) .order_by(ScenarioModifier.id)).scalars()] return ScenarioInputs(recurring, planned, loans, modifiers) def run_projection(session: Session, scenario: Scenario, horizon_days: int, start_date: date) -> ProjectionResult: end = start_date + timedelta(days=horizon_days) inputs = load_scenario_inputs(session, scenario) schedules = [loan_schedule(Decimal(l.principal), Decimal(l.annual_rate_pct), 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] flows = build_cashflows(inputs.recurring, inputs.planned, schedules, payouts, inputs.modifiers, start_date, end) proj = project(total_balance(session), start_date, flows, horizon_days, threshold=get_settings().warn_threshold) session.execute(delete(ProjectionPoint).where( ProjectionPoint.scenario_id == scenario.id)) for d, bal in proj.series: session.add(ProjectionPoint(scenario_id=scenario.id, day=d, balance=bal)) result = session.get(ProjectionResult, scenario.id) or ProjectionResult( scenario_id=scenario.id, computed_at=datetime.now(timezone.utc), low_point_date=proj.low_point[0], low_point_balance=proj.low_point[1]) result.computed_at = datetime.now(timezone.utc) result.low_point_date, result.low_point_balance = proj.low_point result.below_zero_date = proj.first_below_zero result.below_threshold_date = proj.first_below_threshold session.add(result) session.commit() return result # Advisory-Lock-Schlüssel (beliebig, aber projektweit eindeutig). STALE_PROJECTION_LOCK = 110_001 def refresh_stale_projections(session: Session, today: date) -> int: """Rechnet jedes bereits durchgerechnete Szenario neu, dessen Projektion vor `today` beginnt (erster gespeicherter Punkt <= today, d.h. Rechen-Start < today). Maßstab ist der Projektionsbeginn, nicht computed_at (UTC- Zeitstempel, würde um Mitternacht falsch schneiden). Nie gerechnete Szenarien bleiben unberührt. Je Szenario: Lock nehmen, Staleness NEU prüfen (ein paralleler Seitenaufruf kann schon gerechnet haben), rechnen; run_projection committet und gibt damit den Lock frei. Rückgabe: Anzahl neu gerechneter Szenarien.""" horizon = get_settings().horizon_days ids = list(session.execute(select(ProjectionResult.scenario_id)).scalars()) refreshed = 0 for sid in ids: advisory_xact_lock(session, STALE_PROJECTION_LOCK) first = session.execute(select(func.min(ProjectionPoint.day)) .where(ProjectionPoint.scenario_id == sid)).scalar() scenario = session.get(Scenario, sid) if scenario is None or first is None or first > today: session.commit() # Lock freigeben continue run_projection(session, scenario, horizon, today) refreshed += 1 return refreshed