GuidoandClaude Opus 5
adbc3e0aa4
Jedes Paar steht in der Kandidatenliste zweimal, einmal je Richtung. Was verknuepft wird, faellt ueber "verbraucht" heraus - ein blosser Vorschlag aber nicht. Also bot /uebertraege/vorschlaege dieselbe Zahlung zweimal zum Verbinden an, und wer das zweite anklickte, bekam "bereits verknuepft" fuer etwas, das er gerade selbst verknuepft hatte. Betrifft auch die Weboberflaeche: Uebertragsvorschlaege.jsx zeigt die Liste unveraendert an und hatte denselben Doppeleintrag. Gefunden beim Bauen der Arbeitsliste fuer die iOS-App - die Pruefdatei brachte den Vorschlag doppelt zurueck. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
426 lines
18 KiB
Python
426 lines
18 KiB
Python
"""Gleichwährige Überträge zwischen eigenen Konten erkennen.
|
||
|
||
Ein Übertrag steht in zwei Auszügen: als Abgang beim einen Konto, als Zugang
|
||
beim anderen. Beide werden gebucht – das ist richtig, denn beide Salden bewegen
|
||
sich. Nur zählen sie ohne Zusammenhang als Ausgabe UND als Einnahme in der
|
||
Auswertung, obwohl kein Geld ausgegeben wurde.
|
||
|
||
Dieses Modul findet solche Paare und legt eine ``TransferLink`` an. An den
|
||
Buchungen selbst wird nichts geändert; die Verknüpfung ist eine Schicht
|
||
obendrauf. Eine falsche Verknüpfung ist damit ein Etikett, das man abzieht –
|
||
kein Eingriff, der rückgängig gemacht werden müsste.
|
||
|
||
Die Regeln sind bewusst eng. Der Vorgänger hat Zielkonten aus Texten geraten
|
||
und Beträge über Wechselkurse geschätzt; beides hat Salden verfälscht. Hier
|
||
gilt: exakt gleicher Betrag bei gleicher Währung, gleicher Tag.
|
||
|
||
Bei VERSCHIEDENEN Währungen wird der EZB-Kurs des Tages herangezogen – aber nur
|
||
als Prüfung, ob zwei Beträge zueinander passen können. Gespeichert wird nichts
|
||
davon; beide Buchungen behalten den Betrag, der in ihrem Auszug steht. Genau
|
||
das war der Unterschied zum alten Ansatz: Der hat den Gegenbetrag geschätzt und
|
||
in die Buchung geschrieben, und damit Salden verfälscht. Eine Verknüpfung kann
|
||
das nicht – sie ist nur ein Etikett.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from collections import defaultdict
|
||
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..models import (
|
||
Account,
|
||
AccountIdentifier,
|
||
AccountType,
|
||
IdentifierKind,
|
||
LinkIgnored,
|
||
Transaction,
|
||
TransferLink,
|
||
TxnType,
|
||
User,
|
||
)
|
||
from . import currency
|
||
from .csv_import import INBOX_TAGE, normalize_text
|
||
|
||
# Wörter, die eine Bewegung zwischen eigenen Konten ankündigen. Ein Treffer
|
||
# allein verknüpft noch nichts – er erlaubt nur, dass es ohne Rückfrage
|
||
# geschieht. Ohne Treffer wird vorgeschlagen.
|
||
HINWEISE = (
|
||
"übertrag", "uebertrag", "umbuchung", "einzahlung von", "auszahlung",
|
||
"transfer", "gutschrift von", "von konto", "auf konto", "eigenübertrag",
|
||
"umgetauscht", "umtausch", "exchange", "wechsel",
|
||
)
|
||
|
||
# Wie weit ein Betrag bei Währungswechsel vom EZB-Kurs abweichen darf. Banken
|
||
# nehmen einen Aufschlag; Revolut liegt werktags nahe am Referenzkurs, eine
|
||
# klassische Überweisung deutlich darüber. Enger als 2,5 % würde echte Wechsel
|
||
# verwerfen, weiter würde beliebige Beträge zusammenwürfeln.
|
||
FX_TOLERANZ = 0.025
|
||
|
||
|
||
def _trifft_als_wort(nadel: str, heu: str) -> bool:
|
||
"""Nur an Wortgrenzen suchen, nicht mitten im Wort.
|
||
|
||
Ein kurzer Kontoname als blosser Teilstring richtet Schaden an: „Euro"
|
||
steckt auch in „PayPal Europe" und „Europa-Park". Genau daran ist die alte
|
||
Transfer-Erkennung gescheitert.
|
||
"""
|
||
if not nadel:
|
||
return False
|
||
return re.search(rf"(?<![0-9a-zäöüß]){re.escape(nadel)}(?![0-9a-zäöüß])", heu) is not None
|
||
|
||
|
||
def _text(txn: Transaction) -> str:
|
||
return normalize_text(f"{txn.payee or ''} {txn.description or ''}")
|
||
|
||
|
||
def _kontobezeichner(db: Session, user: User) -> dict[int, list[str]]:
|
||
"""Je Konto die Wörter, an denen es im Buchungstext erkennbar ist:
|
||
sein Name und seine Erkennungs-Texte."""
|
||
je_konto: dict[int, list[str]] = defaultdict(list)
|
||
for konto in db.query(Account).filter(Account.owner_id == user.id):
|
||
name = normalize_text(konto.name)
|
||
if len(name) >= 4:
|
||
je_konto[konto.id].append(name)
|
||
for acc_id, kind, wert in (
|
||
db.query(AccountIdentifier.account_id, AccountIdentifier.kind, AccountIdentifier.value_norm)
|
||
.join(Account, Account.id == AccountIdentifier.account_id)
|
||
.filter(Account.owner_id == user.id)
|
||
):
|
||
if kind == IdentifierKind.text and wert and len(wert) >= 3:
|
||
je_konto[acc_id].append(normalize_text(wert))
|
||
return je_konto
|
||
|
||
|
||
def _kontotypen(db: Session, user: User) -> dict[int, AccountType]:
|
||
return dict(db.query(Account.id, Account.type).filter(Account.owner_id == user.id))
|
||
|
||
|
||
def _ist_kartenzahlung(a: Transaction, b: Transaction,
|
||
typen: dict[int, AccountType]) -> bool:
|
||
"""Geht hier eine Kreditkartenrechnung auf?
|
||
|
||
Eine Karte bekommt Geld nur aus einem Grund: weil ihre Rechnung bezahlt
|
||
wird. Das ist Hinweis genug – und noetig, denn im Buchungstext steht bei
|
||
so einer Zahlung oft weder ein Schlagwort noch der Name des Gegenkontos.
|
||
|
||
Entscheidend ist die RICHTUNG, nicht bloss, dass eine Karte beteiligt ist.
|
||
Die weitere Fassung „Kartenkonto auf einer Seite" wuerde einen
|
||
Karteneinkauf ueber 50,00 mit einer zufaelligen Gutschrift ueber 50,00 am
|
||
selben Tag verknuepfen – eine echte Ausgabe UND eine echte Einnahme fielen
|
||
still aus der Auswertung. Genau der Schaden, gegen den _hat_hinweis
|
||
gebaut wurde.
|
||
|
||
Zwei Karten untereinander bleiben aussen vor: Karten zahlen einander
|
||
nicht.
|
||
"""
|
||
for karte, quelle in ((a, b), (b, a)):
|
||
if (karte.type == TxnType.income
|
||
and typen.get(karte.account_id) == AccountType.kreditkarte
|
||
and typen.get(quelle.account_id) != AccountType.kreditkarte):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _hat_hinweis(a: Transaction, b: Transaction, bezeichner: dict[int, list[str]],
|
||
typen: dict[int, AccountType] | None = None) -> bool:
|
||
"""Deutet mindestens eine der beiden Zeilen auf einen Übertrag hin?
|
||
|
||
Ohne diese Hürde würde ein zufälliger Treffer – 50.00 im Laden bezahlt, am
|
||
selben Tag 50.00 auf einem anderen Konto eingegangen – als „eindeutig"
|
||
gelten und eine echte Ausgabe samt echter Einnahme aus der Statistik
|
||
nehmen. Der Saldo bliebe richtig, die Auswertung nicht.
|
||
"""
|
||
if typen and _ist_kartenzahlung(a, b, typen):
|
||
return True
|
||
for txn, gegen in ((a, b), (b, a)):
|
||
text = _text(txn)
|
||
if any(h in text for h in HINWEISE):
|
||
return True
|
||
# Der Name oder Erkennungs-Text des ANDEREN Kontos im Text – an
|
||
# Wortgrenzen geprüft, damit sich „Euro" nicht wieder „Europe" einfängt.
|
||
for wort in bezeichner.get(gegen.account_id, ()):
|
||
if _trifft_als_wort(wort, text):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _kandidaten(db: Session, user: User,
|
||
fenster: int = 0) -> dict[int, list[Transaction]]:
|
||
"""Zu jeder unverknüpften Buchung die möglichen Gegenstücke.
|
||
|
||
``fenster`` ist der erlaubte Abstand in Tagen. 0 heisst derselbe Tag – so
|
||
eng muss es bleiben, wo von selbst verknüpft wird.
|
||
|
||
Für reine VORSCHLÄGE darf es weiter sein, denn dort entscheidet der Nutzer.
|
||
Sonst klafft eine Lücke: Die Übertrags-Prüfung sucht die Gegenseite über
|
||
±INBOX_TAGE und schickt den Nutzer zum Verbinden – die Vorschlagsliste aber
|
||
kannte nur den gleichen Tag und bot dieses Paar nie an. Die Bank bucht die
|
||
Gegenseite regelmässig ein bis drei Tage später.
|
||
"""
|
||
verknuepft = set()
|
||
for a, b in db.query(TransferLink.from_txn_id, TransferLink.to_txn_id).filter(
|
||
TransferLink.owner_id == user.id
|
||
):
|
||
verknuepft.update((a, b))
|
||
ignoriert = {
|
||
tuple(sorted(p))
|
||
for p in db.query(LinkIgnored.a_txn_id, LinkIgnored.b_txn_id).filter(
|
||
LinkIgnored.owner_id == user.id
|
||
)
|
||
}
|
||
waehrung = dict(db.query(Account.id, Account.currency).filter(Account.owner_id == user.id))
|
||
|
||
buchungen = [
|
||
t for t in db.query(Transaction).filter(
|
||
Transaction.owner_id == user.id,
|
||
Transaction.type.in_((TxnType.expense, TxnType.income)),
|
||
)
|
||
if t.id not in verknuepft
|
||
]
|
||
|
||
# Nach Betrag bündeln, der Tag wird danach geprüft – nur so lässt sich ein
|
||
# Fenster zulassen. Innerhalb eines Bündels nach Datum sortiert, damit die
|
||
# Suche abbrechen kann, sobald der Abstand zu gross wird.
|
||
nach_betrag: dict[int, list[Transaction]] = defaultdict(list)
|
||
for t in buchungen:
|
||
nach_betrag[t.amount_cents].append(t)
|
||
|
||
kandidaten: dict[int, list[Transaction]] = defaultdict(list)
|
||
for gruppe in nach_betrag.values():
|
||
if len(gruppe) < 2:
|
||
continue
|
||
gruppe.sort(key=lambda t: (t.date, t.id))
|
||
for i, a in enumerate(gruppe):
|
||
for b in gruppe[i + 1:]:
|
||
if (b.date - a.date).days > fenster:
|
||
break # sortiert: alles Weitere liegt noch ferner
|
||
if a.account_id == b.account_id:
|
||
continue
|
||
if a.type == b.type:
|
||
continue # eine Ausgabe, eine Einnahme
|
||
if waehrung.get(a.account_id) != waehrung.get(b.account_id):
|
||
continue # 100 CHF und 100 EUR sind nicht derselbe
|
||
# Betrag, nur dieselbe Zahl
|
||
if tuple(sorted((a.id, b.id))) in ignoriert:
|
||
continue
|
||
kandidaten[a.id].append(b)
|
||
kandidaten[b.id].append(a)
|
||
|
||
# Währungswechsel: unterschiedliche Beträge. Hier entscheidet der EZB-Kurs,
|
||
# ob die beiden zueinander passen können.
|
||
kandidaten = _ergaenze_waehrungswechsel(db, user, buchungen, waehrung,
|
||
ignoriert, kandidaten, fenster)
|
||
|
||
# Das nächstliegende Datum zuerst: Wo mehrere passen, ist der gleiche Tag
|
||
# das wahrscheinlichere Paar als der drei Tage später.
|
||
datum_je_id = {t.id: t.date for t in buchungen}
|
||
for a_id, liste in kandidaten.items():
|
||
a_datum = datum_je_id[a_id]
|
||
liste.sort(key=lambda b, d=a_datum: (abs((b.date - d).days), b.id))
|
||
return kandidaten
|
||
|
||
|
||
def _passt_ueber_kurs(db: Session, a: Transaction, b: Transaction,
|
||
von: str, nach: str) -> bool:
|
||
"""Passt ``b`` zum umgerechneten ``a``?
|
||
|
||
Fehlt der Kurs für den Tag, wird NICHT geraten – dann gibt es kein Paar.
|
||
"""
|
||
erwartet = currency.convert(db, a.amount_cents, von, nach, a.date)
|
||
if erwartet is None or erwartet <= 0:
|
||
return False
|
||
return abs(erwartet - b.amount_cents) <= erwartet * FX_TOLERANZ
|
||
|
||
|
||
def _ergaenze_waehrungswechsel(db, user, buchungen, waehrung, ignoriert, kandidaten,
|
||
fenster: int = 0):
|
||
nach_datum = sorted(buchungen, key=lambda t: (t.date, t.id))
|
||
for i, a in enumerate(nach_datum):
|
||
for b in nach_datum[i + 1:]:
|
||
if (b.date - a.date).days > fenster:
|
||
break # sortiert: alles Weitere liegt noch ferner
|
||
if a.account_id == b.account_id or a.type == b.type:
|
||
continue
|
||
va, vb = waehrung.get(a.account_id), waehrung.get(b.account_id)
|
||
if not va or not vb or va == vb:
|
||
continue # gleiche Währung ist oben schon erledigt
|
||
if a.amount_cents == b.amount_cents:
|
||
continue # zufällig gleiche Zahl, andere Währung
|
||
if tuple(sorted((a.id, b.id))) in ignoriert:
|
||
continue
|
||
# Die Umrechnung hat eine Richtung – beide prüfen, sonst hinge es
|
||
# davon ab, welche der beiden zufaellig zuerst einsortiert war.
|
||
if _passt_ueber_kurs(db, a, b, va, vb):
|
||
kandidaten[a.id].append(b)
|
||
if _passt_ueber_kurs(db, b, a, vb, va):
|
||
kandidaten[b.id].append(a)
|
||
return kandidaten
|
||
|
||
|
||
def _eindeutig(a: Transaction, moegliche: list[Transaction],
|
||
kandidaten: dict[int, list[Transaction]],
|
||
verbraucht: set[int]) -> Transaction | None:
|
||
"""Gibt es genau eine sinnvolle Zuordnung?
|
||
|
||
Zwei Fälle gelten als zwingend:
|
||
|
||
1. Genau ein Gegenstück, und dieses hat seinerseits nur diese eine
|
||
Möglichkeit.
|
||
2. Alle Gegenstücke sind untereinander austauschbar – gleiches Konto,
|
||
gleicher Betrag, gleicher Tag. Dann ist jede Zuordnung dieselbe, und
|
||
eine beliebige ist richtig. Das ist der Alltag: Revolut-Aufrundungen
|
||
sind hunderte gleiche Beträge am selben Tag zwischen denselben Konten.
|
||
"""
|
||
if not moegliche:
|
||
return None
|
||
if len(moegliche) == 1:
|
||
gegen = moegliche[0]
|
||
# Nur die noch OFFENEN Gegenstücke zählen. Zählte man alle, bliebe beim
|
||
# letzten Paar einer Reihe gleicher Beträge immer eines übrig: Sein
|
||
# Gegenstück hätte rechnerisch viele Kandidaten, obwohl alle anderen
|
||
# längst vergeben sind.
|
||
offen_dort = [x for x in kandidaten.get(gegen.id, ())
|
||
if x.id not in verbraucht and x.id != a.id]
|
||
return gegen if not offen_dort else None
|
||
konten = {b.account_id for b in moegliche}
|
||
return moegliche[0] if len(konten) == 1 else None
|
||
|
||
|
||
def finde(db: Session, user: User, nur_vorschlaege: bool = False) -> dict:
|
||
"""Paare suchen. Eindeutige verknüpfen, den Rest als Vorschlag melden.
|
||
|
||
Mit ``nur_vorschlaege`` wird NICHTS verknüpft – dann kommt alles als
|
||
Vorschlag zurück. Das braucht jede Ansicht, die bloss fragen will („ist das
|
||
ein Übertrag?"), ohne beim Hinsehen schon etwas zu verändern.
|
||
"""
|
||
# Von selbst verknüpft wird nur am gleichen Tag. Vorschläge dürfen weiter
|
||
# greifen – sie ändern nichts, bis der Nutzer zustimmt.
|
||
kandidaten = _kandidaten(db, user, INBOX_TAGE if nur_vorschlaege else 0)
|
||
bezeichner = _kontobezeichner(db, user)
|
||
typen = _kontotypen(db, user)
|
||
nach_id = {}
|
||
for a_id, liste in kandidaten.items():
|
||
for b in liste:
|
||
nach_id[b.id] = b
|
||
for t in db.query(Transaction).filter(Transaction.id.in_(list(kandidaten.keys()) or [0])):
|
||
nach_id[t.id] = t
|
||
|
||
verbraucht: set[int] = set()
|
||
verknuepft = 0
|
||
vorschlaege: list[tuple[Transaction, Transaction]] = []
|
||
# Jedes Paar steht in ``kandidaten`` zweimal, einmal je Richtung. Was
|
||
# verknüpft wird, fällt über ``verbraucht`` heraus – ein Vorschlag aber
|
||
# nicht, und so kam dieselbe Zahlung zweimal zum Verbinden angeboten.
|
||
vorgeschlagen: set[tuple[int, int]] = set()
|
||
|
||
def schlage_vor(a: Transaction, b: Transaction) -> None:
|
||
paar = (min(a.id, b.id), max(a.id, b.id))
|
||
if paar in vorgeschlagen:
|
||
return
|
||
vorgeschlagen.add(paar)
|
||
vorschlaege.append((a, b))
|
||
|
||
for a_id in sorted(kandidaten):
|
||
if a_id in verbraucht:
|
||
continue
|
||
a = nach_id.get(a_id)
|
||
if a is None:
|
||
continue
|
||
offen = [b for b in kandidaten[a_id] if b.id not in verbraucht]
|
||
gegen = _eindeutig(a, offen, kandidaten, verbraucht)
|
||
if gegen is None:
|
||
if offen:
|
||
schlage_vor(a, offen[0])
|
||
continue
|
||
# Automatisch nur ohne Kategorie und mit Hinweis – siehe _hat_hinweis.
|
||
if a.category_id is not None or gegen.category_id is not None:
|
||
schlage_vor(a, gegen)
|
||
continue
|
||
if not _hat_hinweis(a, gegen, bezeichner, typen):
|
||
schlage_vor(a, gegen)
|
||
continue
|
||
if nur_vorschlaege:
|
||
schlage_vor(a, gegen)
|
||
continue
|
||
verbinde(db, user, a, gegen, automatisch=True)
|
||
verbraucht.update((a.id, gegen.id))
|
||
verknuepft += 1
|
||
|
||
if not nur_vorschlaege:
|
||
db.commit()
|
||
return {
|
||
"verknuepft": verknuepft,
|
||
"vorschlaege": [
|
||
{"a": _kurz(db, x), "b": _kurz(db, y)}
|
||
for x, y in vorschlaege
|
||
if x.id not in verbraucht and y.id not in verbraucht
|
||
],
|
||
}
|
||
|
||
|
||
def _kurz(db: Session, t: Transaction) -> dict:
|
||
konto = db.get(Account, t.account_id)
|
||
return {
|
||
"id": t.id, "date": t.date.isoformat(), "type": t.type.value,
|
||
"amount_cents": t.amount_cents, "currency": konto.currency if konto else "",
|
||
"account_id": t.account_id, "account_name": konto.name if konto else "",
|
||
"payee": t.payee, "description": t.description, "category_id": t.category_id,
|
||
}
|
||
|
||
|
||
def verbinde(db: Session, user: User, a: Transaction, b: Transaction,
|
||
automatisch: bool = False) -> TransferLink:
|
||
"""Verknüpfung anlegen. Abgang zuerst, damit die Richtung stimmt."""
|
||
von, nach = (a, b) if a.type == TxnType.expense else (b, a)
|
||
link = TransferLink(owner_id=user.id, from_txn_id=von.id, to_txn_id=nach.id,
|
||
automatisch=automatisch)
|
||
db.add(link)
|
||
db.flush()
|
||
return link
|
||
|
||
|
||
def linkliste(db: Session, user: User) -> dict[int, dict]:
|
||
"""Je Buchungs-ID die Gegenseite – für die Anzeige in der Buchungsliste.
|
||
|
||
Das Gegenkonto kommt mit: Die andere Buchung steht oft gar nicht in der
|
||
gerade gefilterten Liste, und „Übertrag" ohne Angabe wohin sagt nichts.
|
||
"""
|
||
links = db.query(TransferLink).filter(TransferLink.owner_id == user.id).all()
|
||
beteiligt = {i for l in links for i in (l.from_txn_id, l.to_txn_id)}
|
||
# Einmal nachschlagen statt je Verknüpfung – die Liste kann lang werden.
|
||
konto_von = dict(
|
||
db.query(Transaction.id, Transaction.account_id)
|
||
.filter(Transaction.id.in_(beteiligt or [0]))
|
||
.all()
|
||
)
|
||
|
||
ergebnis: dict[int, dict] = {}
|
||
for link in links:
|
||
for eigen, gegen, richtung in (
|
||
(link.from_txn_id, link.to_txn_id, "raus"),
|
||
(link.to_txn_id, link.from_txn_id, "rein"),
|
||
):
|
||
ergebnis[eigen] = {
|
||
"link_id": link.id, "other_id": gegen, "richtung": richtung,
|
||
"other_account_id": konto_von.get(gegen),
|
||
"automatisch": link.automatisch,
|
||
}
|
||
return ergebnis
|
||
|
||
|
||
def verknuepfte_ids(db: Session, user: User) -> set[int]:
|
||
"""Alle Buchungen, die Teil einer Verknüpfung sind.
|
||
|
||
Die Auswertung überspringt sie – so wie sie heute schon neutrale
|
||
Kategorien überspringt. Umkategorisiert wird nichts: Löst man die
|
||
Verknüpfung, zählt die Buchung mit ihrer alten Kategorie wieder mit.
|
||
"""
|
||
ids: set[int] = set()
|
||
for a, b in db.query(TransferLink.from_txn_id, TransferLink.to_txn_id).filter(
|
||
TransferLink.owner_id == user.id
|
||
):
|
||
ids.update((a, b))
|
||
return ids
|