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) 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.""" 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 Einmalzahlungen zaehlen IMMER fuer dieses Szenario - # include_planned steuert nur die globalen PlannedItems. planned += [PlainPlanned(p.name, Decimal(p.amount), p.due, None) 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) 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