bin/finance/app/services/projection_service.py

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 schedule_for
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 = [schedule_for(l) 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