372 lines
17 KiB
Python
372 lines
17 KiB
Python
"""Vorschlagsalgorithmus fuer wiederkehrende Buchungen (Ausbaustufe 9, v2).
|
||
|
||
Ersetzt die reine exakte-Betrags-Gruppierung (v1) durch: Empfaenger-Cluster
|
||
mit Toleranz (haelt Preisdrift in einer Serie zusammen, trennt aber parallele
|
||
Vertraege desselben Anbieters), Rhythmus-Erkennung ueber den Median der
|
||
Buchungsabstaende (monatlich/vierteljaehrlich/jaehrlich statt nur monatlich),
|
||
einen Aktiv-Check (keine "Leichen"-Serien) sowie einen Merge-Pass fuer
|
||
Umfirmierungen (Anbieter aendert den Namen, die Serie laeuft inhaltlich
|
||
weiter). Bindende Spec:
|
||
docs/superpowers/specs/2026-07-20-vorschlags-algorithmus-v2-design.md.
|
||
|
||
Alle Betrags-Toleranzvergleiche verwenden ausschliesslich `Decimal`
|
||
(CLAUDE.md: "Decimal, nicht float" - Rundungsfehler bei Geldbetraegen sind
|
||
inakzeptabel). Tage-Vergleiche (Rhythmus, Aktiv-Check, Merge-Luecke) sind
|
||
ganzzahlige Tage-Arithmetik, niemals float/Decimal-Bruchteile von Tagen.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
import statistics
|
||
from collections import Counter
|
||
from dataclasses import dataclass
|
||
from datetime import date, timedelta
|
||
from decimal import Decimal
|
||
|
||
from sqlalchemy import select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.formats import eur
|
||
from app.models.tables import RecurringItem, Transaction
|
||
|
||
# Betrachtungsfenster (Schritt 1): ~15 Monate. Muss mindestens die zwei
|
||
# Belege einer jaehrlichen Serie (bis zu 400 Tage auseinander) plus etwas
|
||
# Puffer fuer Cluster-/Merge-Bildung abdecken.
|
||
WINDOW_DAYS = 460
|
||
|
||
# Rhythmus-Tabelle: (min_tage, max_tage, mindestbelege) je Rhythmus. Der
|
||
# Median der Buchungsabstaende einer Serie muss ins Intervall fallen, UND es
|
||
# muessen mindestens so viele Buchungen vorliegen (ein einzelner Zufallstreffer
|
||
# mit "passendem" Abstand soll nicht als Serie gelten).
|
||
RHYTHMS: dict[str, tuple[int, int, int]] = {
|
||
"monthly": (25, 36, 3),
|
||
"quarterly": (80, 105, 3),
|
||
"yearly": (330, 400, 2),
|
||
}
|
||
|
||
# Nominelle Schrittweite je Rhythmus in Tagen - Referenzwert fuer Aktiv-Check
|
||
# und Merge-Luecken-Fenster (Schritt 5/6).
|
||
STEP_DAYS: dict[str, int] = {"monthly": 30, "quarterly": 91, "yearly": 365}
|
||
|
||
# Aktiv-Check (Schritt 5): die letzte Buchung darf hoechstens das 1,75-fache
|
||
# der Rhythmus-Schrittweite zurueckliegen, sonst gilt die Serie als beendet
|
||
# ("Leiche") und wird nicht vorgeschlagen. Als Fraction 7/4 ausgedrueckt und
|
||
# ganzzahlig verglichen (delta_tage * 4 <= schrittweite * 7), damit keine
|
||
# Gleitkomma-Rundung ueber "aktiv"/"inaktiv" entscheidet.
|
||
ACTIVITY_FACTOR_NUM = 7
|
||
ACTIVITY_FACTOR_DEN = 4
|
||
|
||
# Relative Toleranzen (immer als Decimal verglichen, nie float):
|
||
CLUSTER_TOL = Decimal("0.35") # Schritt 3: Betrags-Cluster (haelt Preisdrift zusammen)
|
||
MERGE_TOL = Decimal("0.25") # Schritt 6: Umfirmierungs-Merge ueber Gruppengrenzen
|
||
BESTAND_TOL = Decimal("0.10") # Schritt 8: Bestandsabgleich gegen RecurringItem
|
||
|
||
# Merge-Luecke (Schritt 6): die Zeit zwischen dem Ende von Serie A und dem
|
||
# Beginn von Serie B muss zwischen dem 0,4- und 1,6-fachen der
|
||
# Rhythmus-Schrittweite liegen (als ganzzahlige Bruchvergleiche, aus
|
||
# demselben Grund wie beim Aktiv-Check).
|
||
MERGE_GAP_MIN_NUM, MERGE_GAP_MIN_DEN = 4, 10 # 0.4
|
||
MERGE_GAP_MAX_NUM, MERGE_GAP_MAX_DEN = 16, 10 # 1.6
|
||
MERGE_DUE_DAY_TOL = 3 # Schritt 6: Faelligkeitstag-Toleranz in Tagen
|
||
BESTAND_DUE_DAY_TOL = 2 # Schritt 8: Faelligkeitstag-Toleranz in Tagen
|
||
|
||
# Bestandsabgleich, Token-Match (Live-Gate-Fund, Nachtrag 4): kuratierte
|
||
# Fixposten tragen haeufig einen Alias-/Variabel-Namen, der weder Substring
|
||
# noch betragsaehnlich zum automatisch erkannten Vorschlag ist (Muster:
|
||
# ein Sammel-Fixposten fuer eine Kreditkartenabrechnung mit variablem Betrag
|
||
# unter einem Alias-Namen des Anbieters deckt den vom Algorithmus erkannten
|
||
# Vorschlag desselben Anbieters unter seinem regulaeren Empfaenger-Namen
|
||
# nicht ab, weil weder Substring noch Betrags-Toleranz greifen). Ein
|
||
# gemeinsames, hinreichend spezifisches Namens-Token (>=5 Zeichen, um
|
||
# generische Woerter wie "Bank" nicht faelschlich matchen zu lassen) bei
|
||
# gleichem Rhythmus und nahem Faelligkeitstag gilt als ausreichendes Indiz
|
||
# fuer denselben Fixposten.
|
||
TOKEN_MIN_LEN = 5
|
||
|
||
# Volatilitaets-Hinweis (Live-Gate A9-Fund, Nachtrag 4): wenn der
|
||
# Betrags-Cluster-Split (Schritt 3) die neueste Buchung der Empfaenger-Gruppe
|
||
# abgetrennt hat (weil sie zu stark vom Serien-Betrag abweicht), ist der
|
||
# vorgeschlagene Betrag ggf. schon wieder veraltet - keine Unterdrueckung,
|
||
# nur ein Warnhinweis fuer die Nutzerin/den Nutzer.
|
||
VOLATILITAETS_HINWEIS = "Beträge schwanken stark – letzte Buchung weicht ab"
|
||
|
||
|
||
def _norm(name: str) -> str:
|
||
"""Normalisiert einen Empfaenger-Namen fuer Gruppen- und
|
||
Substring-Vergleich: Bankexporte schreiben denselben Empfaenger nicht
|
||
einheitlich (Gross-/Kleinschreibung, mehrfache Leerzeichen), das ist fuer
|
||
die Erkennung irrelevant."""
|
||
return " ".join(name.split()).casefold()
|
||
|
||
|
||
def _tokens(name: str) -> set[str]:
|
||
"""Zerlegt einen normalisierten Namen an Nicht-Alphanumerik in Tokens
|
||
(fuer den Token-Match im Bestandsabgleich, Schritt 8). Nur Tokens ab
|
||
TOKEN_MIN_LEN Zeichen zaehlen, damit kurze generische Woerter ("eG",
|
||
"AG", "Bank") keine falschen Treffer erzeugen."""
|
||
return {tok for tok in re.split(r"[^a-z0-9]+", _norm(name)) if len(tok) >= TOKEN_MIN_LEN}
|
||
|
||
|
||
def _rel_diff(a: Decimal, b: Decimal) -> Decimal:
|
||
"""Relative Differenz von Betrag a zur Referenz b (immer >= 0), als
|
||
Decimal. b=0 kommt praktisch nicht vor (eine Nullbuchung bildet keine
|
||
erkennbare Serie); fuer diesen Sonderfall gilt "keine Aehnlichkeit"."""
|
||
if b == 0:
|
||
return Decimal("Infinity") if a != 0 else Decimal("0")
|
||
return abs(a - b) / abs(b)
|
||
|
||
|
||
@dataclass
|
||
class _Series:
|
||
"""Eine erkannte Serie: chronologisch sortierte Buchungen eines
|
||
Betrags-Clusters mit zugeordnetem Rhythmus.
|
||
|
||
`volatile` markiert, dass der Cluster-Split (Schritt 3) innerhalb der
|
||
Empfaenger-Gruppe eine NEUERE, betragsmaessig abweichende Buchung
|
||
abgetrennt hat - der hier vorgeschlagene Betrag koennte also schon
|
||
wieder veraltet sein (siehe VOLATILITAETS_HINWEIS)."""
|
||
items: list[Transaction]
|
||
rhythm: str
|
||
volatile: bool = False
|
||
|
||
@property
|
||
def first(self) -> Transaction:
|
||
return self.items[0]
|
||
|
||
@property
|
||
def last(self) -> Transaction:
|
||
return self.items[-1]
|
||
|
||
|
||
def _amount_clusters(items: list[Transaction]) -> list[list[Transaction]]:
|
||
"""Schritt 3: teilt chronologisch sortierte Buchungen einer
|
||
Empfaenger-Gruppe in Betrags-Cluster. Eine Buchung haengt sich an das
|
||
Cluster, dessen zuletzt aufgenommenes Mitglied gleiches Vorzeichen und
|
||
eine relative Differenz <= CLUSTER_TOL hat (greedy, erstes passendes
|
||
Cluster gewinnt) - das haelt eine langsam driftende Serie (Preiserhoehung)
|
||
zusammen, trennt aber parallele Vertraege mit deutlich anderem Betrag."""
|
||
clusters: list[list[Transaction]] = []
|
||
for t in items:
|
||
for cluster in clusters:
|
||
last = cluster[-1]
|
||
same_sign = (t.amount > 0) == (last.amount > 0)
|
||
if same_sign and _rel_diff(Decimal(t.amount), Decimal(last.amount)) <= CLUSTER_TOL:
|
||
cluster.append(t)
|
||
break
|
||
else:
|
||
clusters.append([t])
|
||
return clusters
|
||
|
||
|
||
def _classify(dates: list[date]) -> str | None:
|
||
"""Schritt 4: bestimmt den Rhythmus einer Serie ueber den Median der
|
||
Buchungsabstaende (robust gegen einzelne Ausreisser, z.B.
|
||
Wochenend-/Feiertagsverschiebung einer einzelnen Buchung)."""
|
||
if len(dates) < 2:
|
||
return None
|
||
gaps = [(b - a).days for a, b in zip(dates, dates[1:])]
|
||
median_gap = statistics.median(gaps)
|
||
for rhythm, (lo, hi, min_belege) in RHYTHMS.items():
|
||
if len(dates) >= min_belege and lo <= median_gap <= hi:
|
||
return rhythm
|
||
return None
|
||
|
||
|
||
def _merge_gap_ok(gap_days: int, step: int) -> bool:
|
||
"""Schritt 6: Luecke zwischen Serienende und -beginn im Fenster
|
||
[0,4; 1,6] * Schrittweite (ganzzahliger Bruchvergleich, keine Rundung)."""
|
||
return (gap_days * MERGE_GAP_MIN_DEN >= MERGE_GAP_MIN_NUM * step
|
||
and gap_days * MERGE_GAP_MAX_DEN <= MERGE_GAP_MAX_NUM * step)
|
||
|
||
|
||
def _mergeable(a: _Series, b: _Series) -> bool:
|
||
"""Prueft die Umfirmierungs-Merge-Bedingungen aus Schritt 6 fuer ein
|
||
Paar (A endet, B beginnt danach): gleicher Rhythmus, plausible Luecke,
|
||
Faelligkeitstag nah beieinander (Transitionspunkte: letzte Buchung von A
|
||
gegen erste Buchung von B), Betrag nicht sprunghaft veraendert."""
|
||
if a.rhythm != b.rhythm:
|
||
return False
|
||
if a.last.booking_date >= b.first.booking_date:
|
||
return False
|
||
step = STEP_DAYS[a.rhythm]
|
||
gap = (b.first.booking_date - a.last.booking_date).days
|
||
if not _merge_gap_ok(gap, step):
|
||
return False
|
||
if abs(a.last.booking_date.day - b.first.booking_date.day) > MERGE_DUE_DAY_TOL:
|
||
return False
|
||
same_sign = (a.last.amount > 0) == (b.first.amount > 0)
|
||
if not same_sign:
|
||
return False
|
||
return _rel_diff(Decimal(b.first.amount), Decimal(a.last.amount)) <= MERGE_TOL
|
||
|
||
|
||
def _try_merge(series_list: list[_Series]) -> list[_Series]:
|
||
"""Schritt 6: fasst Serien desselben Kontos ueber Gruppengrenzen
|
||
(unterschiedlicher normalisierter Empfaenger-Name, z.B. nach einer
|
||
Umfirmierung) zusammen, solange `_mergeable` zutrifft. Laeuft iterativ
|
||
bis zum Fixpunkt, damit eine bereits gemergte Serie mit einer weiteren,
|
||
noch juengeren Serie erneut zusammengefasst werden kann (z.B. zwei
|
||
Umbenennungen hintereinander)."""
|
||
series_list = list(series_list)
|
||
changed = True
|
||
while changed:
|
||
changed = False
|
||
for i, a in enumerate(series_list):
|
||
for j, b in enumerate(series_list):
|
||
if i == j or not _mergeable(a, b):
|
||
continue
|
||
merged = _Series(
|
||
items=sorted(a.items + b.items, key=lambda t: t.booking_date),
|
||
rhythm=a.rhythm,
|
||
volatile=a.volatile or b.volatile,
|
||
)
|
||
series_list = [s for k, s in enumerate(series_list) if k not in (i, j)]
|
||
series_list.append(merged)
|
||
changed = True
|
||
break
|
||
if changed:
|
||
break
|
||
return series_list
|
||
|
||
|
||
def _covered_by_existing(cand_name: str, cand_amount: Decimal, rhythm: str, due_day: int,
|
||
existing: list[RecurringItem]) -> bool:
|
||
"""Schritt 8 (Bestandsabgleich): ein Vorschlag entfaellt, wenn er bereits
|
||
als Fixposten gepflegt ist - ueber einen von drei Wegen:
|
||
(a) Namens-Substring-Match (normalisiert, in beide Richtungen: sowohl
|
||
Kurz- als auch Langschreibweisen kommen in der Praxis in beiden
|
||
Datenquellen vor);
|
||
(b) Rhythmus + Faelligkeitstag + Betrag innerhalb enger Toleranz (falls
|
||
der Fixposten unter einem ganz anderen Namen gepflegt wurde);
|
||
(c) Token-Match: gleicher Rhythmus, Faelligkeitstag-Differenz <= 2 UND
|
||
mindestens ein gemeinsames Namens-Token (>=5 Zeichen) - faengt
|
||
kuratierte Alias-/Variabel-Fixposten, deren Name UND Betrag stark
|
||
vom automatisch erkannten Vorschlag abweichen (Live-Gate-Fund: ein
|
||
Sammel-Fixposten unter Alias-Namen des Anbieters deckt den
|
||
automatisch erkannten Vorschlag desselben Anbieters unter seinem
|
||
regulaeren Empfaenger-Namen ab, obwohl weder (a) noch (b) greifen)."""
|
||
cand_norm = _norm(cand_name)
|
||
cand_tokens = _tokens(cand_name)
|
||
for item in existing:
|
||
item_norm = _norm(item.name)
|
||
if cand_norm in item_norm or item_norm in cand_norm:
|
||
return True
|
||
if (item.rhythm == rhythm
|
||
and abs(item.due_day - due_day) <= BESTAND_DUE_DAY_TOL
|
||
and _rel_diff(cand_amount, Decimal(item.amount)) <= BESTAND_TOL):
|
||
return True
|
||
if (item.rhythm == rhythm
|
||
and abs(item.due_day - due_day) <= BESTAND_DUE_DAY_TOL
|
||
and cand_tokens & _tokens(item.name)):
|
||
return True
|
||
return False
|
||
|
||
|
||
def suggest_recurring(session: Session, today: date | None = None) -> list[dict]:
|
||
"""Ermittelt Vorschlaege fuer wiederkehrende Posten aus bestaetigten
|
||
Buchungen der letzten WINDOW_DAYS Tage. `today` ist ausschliesslich zu
|
||
Testzwecken injizierbar (deterministischer Aktiv-Check) - im
|
||
Produktivbetrieb liefert der Default `date.today()`. Reihenfolge der
|
||
Schritte gemaess Spec, mit einer bewussten Umstellung gegenueber der
|
||
Nummerierung dort: der Aktiv-Check (Schritt 5) laeuft NACH dem
|
||
Umfirmierungs-Merge (Schritt 6) auf der ggf. gemergten Serie - sonst
|
||
wuerde eine per Umfirmierung fortgesetzte Serie an ihrem alten,
|
||
laengst inaktiven Teil scheitern, bevor der Merge sie retten kann."""
|
||
if today is None:
|
||
today = date.today()
|
||
cutoff = today - timedelta(days=WINDOW_DAYS)
|
||
|
||
# Schritt 1: Datenbasis.
|
||
txs = session.execute(
|
||
select(Transaction)
|
||
.where(Transaction.status == "confirmed", Transaction.booking_date >= cutoff)
|
||
).scalars().all()
|
||
|
||
# Schritt 2: Gruppierung je (Konto, normalisierter Empfaenger).
|
||
groups: dict[tuple[int, str], list[Transaction]] = {}
|
||
for t in txs:
|
||
groups.setdefault((t.account_id, _norm(t.counterparty)), []).append(t)
|
||
|
||
# Schritt 3+4: je Gruppe Betrags-Cluster bilden und Rhythmus klassifizieren.
|
||
series_by_account: dict[int, list[_Series]] = {}
|
||
for (account_id, _name_norm), items in groups.items():
|
||
items_sorted = sorted(items, key=lambda t: t.booking_date)
|
||
clusters = _amount_clusters(items_sorted)
|
||
classified = [(cluster, _classify([t.booking_date for t in cluster]))
|
||
for cluster in clusters]
|
||
# Fuer den Volatilitaets-Check zaehlt eine neuere Buchung nur dann als
|
||
# "abgetrennt", wenn sie NICHT bereits zu einem ANDEREN qualifizierten
|
||
# (klassifizierten) Cluster derselben Gruppe gehoert - sonst waeren
|
||
# zwei parallele, stabile Vertraege (jeder fuer sich eine gueltige
|
||
# eigene Serie) faelschlich als "volatil" markiert, nur weil der
|
||
# jeweils andere Vertrag zufaellig spaeter im Monat faellig ist
|
||
# (Nachtrag 3b, Fable-Gate-Korrektur nach dem ersten Live-Gate-Fund).
|
||
qualified_items = {t for cluster, rhythm in classified if rhythm is not None
|
||
for t in cluster}
|
||
for cluster, rhythm in classified:
|
||
if rhythm is None:
|
||
continue
|
||
# Volatilitaets-Hinweis: hat der Cluster-Split innerhalb DIESER
|
||
# Empfaenger-Gruppe (gleiches Konto, gleiches Vorzeichen) eine
|
||
# NEUERE Buchung in einen UNQUALIFIZIERTEN Cluster abgetrennt
|
||
# (z.B. eine einzelne Ausreisser-Buchung, die allein keine Serie
|
||
# bildet), ist der hier vorgeschlagene (letzte) Betrag ggf. schon
|
||
# veraltet.
|
||
cluster_sign = cluster[-1].amount > 0
|
||
volatile = any(
|
||
(t.amount > 0) == cluster_sign
|
||
and t.booking_date > cluster[-1].booking_date
|
||
and t not in qualified_items
|
||
for t in items_sorted
|
||
)
|
||
series_by_account.setdefault(account_id, []).append(
|
||
_Series(items=cluster, rhythm=rhythm, volatile=volatile))
|
||
|
||
existing = list(session.execute(select(RecurringItem)).scalars())
|
||
|
||
suggestions: list[dict] = []
|
||
for series_list in series_by_account.values():
|
||
# Schritt 6: Umfirmierungs-Merge ueber Gruppengrenzen, je Konto.
|
||
for s in _try_merge(series_list):
|
||
# Schritt 5: Aktiv-Check auf der (ggf. gemergten) finalen Serie.
|
||
step = STEP_DAYS[s.rhythm]
|
||
delta_tage = (today - s.last.booking_date).days
|
||
if delta_tage * ACTIVITY_FACTOR_DEN > step * ACTIVITY_FACTOR_NUM:
|
||
continue
|
||
|
||
# Schritt 7: Vorschlagswerte aus der neuesten Buchung.
|
||
last = s.last
|
||
name = last.counterparty
|
||
amount = Decimal(last.amount)
|
||
due_day = last.booking_date.day
|
||
start_date = last.booking_date if s.rhythm in ("quarterly", "yearly") else None
|
||
cat_counts = Counter(t.category_id for t in s.items if t.category_id is not None)
|
||
category_id = cat_counts.most_common(1)[0][0] if cat_counts else None
|
||
|
||
hinweis = ""
|
||
if len(s.items) >= 2:
|
||
previous = Decimal(s.items[-2].amount)
|
||
# "Gestiegen" bezieht sich auf den Betragswert (Ausgaben sind
|
||
# negativ: gestiegen heisst betragsmaessig groesser, also
|
||
# abs(neu) > abs(alt)), nicht auf das Vorzeichen.
|
||
if abs(amount) > abs(previous):
|
||
hinweis = f"Betrag zuletzt gestiegen (vorher {eur(abs(previous))} €)"
|
||
|
||
if s.volatile:
|
||
hinweis = f"{hinweis} {VOLATILITAETS_HINWEIS}".strip()
|
||
|
||
# Schritt 8: Bestandsabgleich.
|
||
if _covered_by_existing(name, amount, s.rhythm, due_day, existing):
|
||
continue
|
||
|
||
suggestions.append({
|
||
"name": name,
|
||
"amount": amount,
|
||
"rhythm": s.rhythm,
|
||
"due_day": due_day,
|
||
"start_date": start_date,
|
||
"category_id": category_id,
|
||
"hinweis": hinweis,
|
||
})
|
||
return suggestions
|