feat: Erkennung möglicher Doppelzählung Kredit/Fixposten (A12-T2)

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

finance/app/services/double_count.py

@@ -0,0 +1,127 @@
+"""Doppelzählungs-Erkennung Kredit ↔ wiederkehrender Posten (Ausbaustufe 12).
+
+Ein monatlicher Ausgabe-Posten gilt als vermutlich doppelt zu einem Kredit,
+wenn sein Betrag innerhalb der Toleranz der regulären Monatsrate liegt und
+sich die Zeiträume überschneiden. Bewusst kein Namensvergleich
+(„Autokredit" vs. „Rate VW Bank" ist nicht zuverlässig zuordenbar).
+Reine Logik auf Wertobjekten, damit auch noch nicht gespeicherte bzw. per
+PATCH veränderte Zustände geprüft werden können. Alle Beträge Decimal.
+"""
+from dataclasses import dataclass
+from datetime import date
+from decimal import ROUND_HALF_UP, Decimal
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.engine.loans import add_months, annuity_payment
+from app.models.tables import Loan, RecurringItem
+
+CODE = "doppelzaehlung_kredit"
+TOLERANZ_PROZENT = Decimal("3")
+TOLERANZ_MIN = Decimal("1.00")
+CENT = Decimal("0.01")
+
+
+@dataclass(frozen=True)
+class PostenDaten:
+    id: int | None
+    name: str
+    amount: Decimal
+    rhythm: str
+    start_date: date | None
+    end_date: date | None
+
+
+@dataclass(frozen=True)
+class KreditDaten:
+    id: int | None
+    name: str
+    principal: Decimal
+    annual_rate_pct: Decimal
+    term_months: int
+    payout_date: date
+    repayment_type: str
+
+
+@dataclass(frozen=True)
+class Konflikt:
+    posten_id: int | None
+    posten_name: str
+    posten_betrag: Decimal
+    kredit_id: int | None
+    kredit_name: str
+    kredit_rate: Decimal
+
+    @property
+    def paar(self) -> tuple[int | None, int | None]:
+        return (self.posten_id, self.kredit_id)
+
+
+def posten_aus(item: RecurringItem) -> PostenDaten:
+    return PostenDaten(item.id, item.name, Decimal(item.amount), item.rhythm,
+                       item.start_date, item.end_date)
+
+
+def kredit_aus(loan: Loan) -> KreditDaten:
+    return KreditDaten(loan.id, loan.name, Decimal(loan.principal),
+                       Decimal(loan.annual_rate_pct), loan.term_months,
+                       loan.payout_date, loan.repayment_type)
+
+
+def regular_rate(k: KreditDaten) -> Decimal | None:
+    """Reguläre Monatsrate (positiv): Annuität bzw. bei endfälligen Krediten
+    die monatlichen Zinsen. Ohne reguläre Rate (0 % endfällig, Laufzeit <= 0)
+    -> None, dann nie Konflikt. Die Schlusszahlung wird nicht betrachtet."""
+    if k.term_months <= 0:
+        return None
+    if k.repayment_type == "bullet":
+        if k.annual_rate_pct == 0:
+            return None
+        return (k.principal * k.annual_rate_pct / Decimal(100) / Decimal(12)
+                ).quantize(CENT, ROUND_HALF_UP)
+    return annuity_payment(k.principal, k.annual_rate_pct, k.term_months)
+
+
+def rate_period(k: KreditDaten) -> tuple[date, date]:
+    """Erste und letzte Rate, identisch zu engine/loans.loan_schedule."""
+    return add_months(k.payout_date, 1), add_months(k.payout_date, k.term_months)
+
+
+def find_conflict(p: PostenDaten, k: KreditDaten) -> Konflikt | None:
+    if p.rhythm != "monthly" or p.amount >= 0:
+        return None
+    rate = regular_rate(k)
+    if rate is None or rate <= 0:
+        return None
+    toleranz = max(TOLERANZ_MIN, rate * TOLERANZ_PROZENT / Decimal(100))
+    if abs(-p.amount - rate) > toleranz:
+        return None
+    erste, letzte = rate_period(k)
+    if p.start_date is not None and p.start_date > letzte:
+        return None
+    if p.end_date is not None and p.end_date < erste:
+        return None
+    return Konflikt(p.id, p.name, p.amount, k.id, k.name, rate)
+
+
+def conflicts_for_recurring(session: Session, p: PostenDaten) -> list[Konflikt]:
+    loans = session.execute(select(Loan).order_by(Loan.name, Loan.id)).scalars()
+    return [c for c in (find_conflict(p, kredit_aus(l)) for l in loans) if c]
+
+
+def conflicts_for_loan(session: Session, k: KreditDaten) -> list[Konflikt]:
+    items = session.execute(select(RecurringItem)
+                            .order_by(RecurringItem.name, RecurringItem.id)).scalars()
+    return [c for c in (find_conflict(posten_aus(r), k) for r in items) if c]
+
+
+def all_conflicts(session: Session) -> list[Konflikt]:
+    kredite = [kredit_aus(l) for l in session.execute(
+        select(Loan).order_by(Loan.name, Loan.id)).scalars()]
+    out: list[Konflikt] = []
+    for r in session.execute(select(RecurringItem)
+                             .order_by(RecurringItem.name, RecurringItem.id)).scalars():
+        p = posten_aus(r)
+        out.extend(c for c in (find_conflict(p, k) for k in kredite) if c)
+    return out