GuidoandClaude Opus 5
6cd1c100c0
Ein Franken-in-Euro-Tausch stand zweimal da: die CHF-Seite als Uebertrag nach Revolut Euro, die EUR-Seite weiterhin als eigene Einnahme. Revolut Euro war damit doppelt gutgeschrieben. Drei Fehler auf einmal. Erstens prueft "zum Uebertrag machen" jetzt, ob die Buchung bereits mit ihrer Gegenseite VERBUNDEN ist. Dann sind beide Seiten gebucht und zaehlen schon in keiner Auswertung; ein Uebertrag bewegt das Geld auf dem Zielkonto ein zweites Mal. Das laesst sich auch nicht mit "trotzdem" wegdruecken – es ist nicht Ansichtssache. Zweitens wird auf dem Zielkonto nach einer passenden gebuchten Gegenseite gesehen, auch ohne Verknuepfung: gleiche Tage, passender Betrag, ueber Waehrungen hinweg mit Spielraum. Findet sich eine, kommt die Warnung samt dem richtigen Weg – die beiden verbinden statt einen Uebertrag anzulegen. Der Treffer ist unscharf, deshalb hier mit "trotzdem" uebergehbar. Drittens der beobachtete Punkt: Der nachtraegliche Weg reichte den Eingangs-Eintrag nicht durch und rechnete deshalb IMMER mit dem EZB-Kurs, auch wenn die Bank ihren eigenen genannt hatte. Jetzt wird er nachgeschlagen, und der Kurs der Bank landet an der Buchung. Und Loeschen raeumt die Eingangs-Eintraege mit. Der Fremdschluessel setzt sonst nur ihre Verweisspalte auf NULL, und sie blieben als "gebucht" ohne Buchung stehen – bei einem Uebertrag sind das zwei. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
351 lines
14 KiB
Python
351 lines
14 KiB
Python
"""Etwas zum Kontouebertrag machen – aus dem Eingang oder nachtraeglich.
|
||
|
||
Zwei Wege fuehren hierher: ein offener Vorschlag im Eingang und eine schon
|
||
gebuchte Ausgabe oder Einnahme. Beide muessen dieselben vier Dinge einhalten,
|
||
deshalb liegen sie hier und nicht in den Routern:
|
||
|
||
1. Die Richtung kommt aus dem Vorzeichen. Ein Abgang geht von hier weg, ein
|
||
Zugang kommt vom Gegenkonto her.
|
||
2. Der Betrag der Gegenseite ist bekannt oder geschaetzt – das entscheidet
|
||
ueber Punkt 3.
|
||
3. Ist er geschaetzt und haengt das Zielkonto selbst an einer Bank, wird
|
||
nachgefragt: Die Bank meldet den Zugang von allein, und ein geschaetzter
|
||
Betrag laesst sich beim Abgleich nicht wiedererkennen.
|
||
4. Auf dem Zielkonto entsteht ein Eingangs-Eintrag, der auf DIESELBE Buchung
|
||
zeigt. Er faengt die Gegenseite ab – die schon vorliegende, indem er sie
|
||
verbraucht, und die spaeter eintreffende, weil ``detect_rows`` gegen
|
||
gebuchte Eingangs-Eintraege prueft.
|
||
|
||
Ohne Punkt 4 stuende das Geld zweimal auf dem Zielkonto: Der Duplikatschutz
|
||
vergleicht ``import_hash`` – den hat eine Uebertrags-Buchung nicht – und
|
||
Eingangs-Eintraege, und ohne diesen legte der Uebertrag auf dem Zielkonto
|
||
keinen an.
|
||
|
||
Nicht betroffen ist das Buchungsformular: Wer dort von Hand einen Uebertrag
|
||
anlegt, waehlt beide Konten bewusst und hat keine Gegenseite, die noch kommt.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import timedelta
|
||
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..models import (
|
||
Account,
|
||
BankAccountLink,
|
||
InboxItem,
|
||
InboxSource,
|
||
InboxStatus,
|
||
Transaction,
|
||
TransferLink,
|
||
TxnType,
|
||
User,
|
||
)
|
||
from . import buchung, csv_import, currency
|
||
from .csv_import import INBOX_TAGE
|
||
from .uebertraege import FX_TOLERANZ
|
||
|
||
|
||
class UebertragFehler(RuntimeError):
|
||
"""Fehler mit einem Text, den man dem Nutzer zeigen kann."""
|
||
|
||
def __init__(self, text: str, nachfragen: bool = False):
|
||
super().__init__(text)
|
||
# True heisst: Es ist kein Fehler, sondern eine Warnung, die sich mit
|
||
# ``trotzdem`` uebergehen laesst.
|
||
self.nachfragen = nachfragen
|
||
|
||
|
||
def ist_angebunden(db: Session, user: User, konto: Account) -> bool:
|
||
"""Holt Certain die Umsaetze dieses Kontos selbst von der Bank?
|
||
|
||
Entscheidend fuer einen Uebertrag: Bei einem angebundenen Gegenkonto
|
||
liefert die Bank die Gegenbuchung selbst.
|
||
"""
|
||
return db.query(BankAccountLink.id).filter(
|
||
BankAccountLink.owner_id == user.id,
|
||
BankAccountLink.account_id == konto.id,
|
||
).first() is not None
|
||
|
||
|
||
def _eingang_zu(db: Session, user: User, txn: Transaction) -> InboxItem | None:
|
||
"""Der Eingangs-Eintrag, aus dem diese Buchung entstand.
|
||
|
||
Er traegt, was die Bank ueber die Umrechnung gesagt hat. Ohne ihn bliebe
|
||
nur der EZB-Kurs – auch dann, wenn die Bank ihren eigenen genannt hat.
|
||
"""
|
||
return db.query(InboxItem).filter(
|
||
InboxItem.owner_id == user.id,
|
||
InboxItem.transaction_id == txn.id,
|
||
InboxItem.account_id == txn.account_id,
|
||
).first()
|
||
|
||
|
||
def _schon_verknuepft(db: Session, user: User, txn: Transaction) -> bool:
|
||
"""Haengt diese Buchung schon als Paar an ihrer Gegenseite?"""
|
||
return db.query(TransferLink.id).filter(
|
||
TransferLink.owner_id == user.id,
|
||
(TransferLink.from_txn_id == txn.id) | (TransferLink.to_txn_id == txn.id),
|
||
).first() is not None
|
||
|
||
|
||
def _gebuchte_gegenseite(db: Session, user: User, konto: Account, ziel: Account,
|
||
eigener_typ: TxnType, erwartet: int | None, datum,
|
||
exakt: bool) -> Transaction | None:
|
||
"""Steht auf dem Zielkonto schon eine passende Buchung?
|
||
|
||
Dann waere ein Uebertrag falsch: Er wuerde das Geld dort ein zweites Mal
|
||
bewegen. Richtig ist in dem Fall, die beiden zu VERKNUEPFEN – beide
|
||
Buchungen bleiben stehen und zaehlen nur nicht mehr in der Auswertung.
|
||
|
||
Ueber Waehrungen hinweg wird mit Spielraum verglichen: Der erwartete Betrag
|
||
ist dann gerechnet, der echte kommt von der Bank.
|
||
"""
|
||
if not erwartet:
|
||
return None
|
||
gegen_typ = TxnType.income if eigener_typ == TxnType.expense else TxnType.expense
|
||
spielraum = 0 if exakt else max(1, round(erwartet * FX_TOLERANZ))
|
||
return next(
|
||
(k for k in db.query(Transaction).filter(
|
||
Transaction.owner_id == user.id,
|
||
Transaction.account_id == ziel.id,
|
||
Transaction.type == gegen_typ,
|
||
Transaction.date >= datum - timedelta(days=INBOX_TAGE),
|
||
Transaction.date <= datum + timedelta(days=INBOX_TAGE),
|
||
) if abs(k.amount_cents - erwartet) <= spielraum),
|
||
None,
|
||
)
|
||
|
||
|
||
def _pruefe_gegenseite(db: Session, user: User, konto: Account, ziel: Account,
|
||
eigener_typ: TxnType, erwartet: int | None, datum,
|
||
exakt: bool, trotzdem: bool) -> None:
|
||
if trotzdem:
|
||
return
|
||
schon = _gebuchte_gegenseite(db, user, konto, ziel, eigener_typ, erwartet,
|
||
datum, exakt)
|
||
if schon is None:
|
||
return
|
||
raise UebertragFehler(
|
||
f"Auf '{ziel.name}' steht schon eine passende Buchung vom "
|
||
f"{schon.date.strftime('%d.%m.%Y')}. Beide Seiten sind also bereits "
|
||
f"gebucht – ein Übertrag würde das Geld dort ein zweites Mal bewegen. "
|
||
f"Richtig ist hier, die beiden zu verbinden: Sie bleiben stehen und "
|
||
f"zählen nur nicht mehr als Ausgabe und Einnahme.",
|
||
nachfragen=True,
|
||
)
|
||
|
||
|
||
def _gegenbetrag(db: Session, konto: Account, ziel: Account, betrag: int,
|
||
datum, eintrag: InboxItem | None = None,
|
||
) -> tuple[int | None, float | None, bool]:
|
||
"""Betrag auf der Zielseite, Kurs, und ob der Betrag BEKANNT ist.
|
||
|
||
Drei Stufen, absteigend nach Verlaesslichkeit:
|
||
|
||
1. Der Gegenbetrag, den die Bank selbst genannt hat – exakt.
|
||
2. Ihr Kurs. Genauer als die EZB, aber der Betrag bleibt gerechnet.
|
||
3. Der EZB-Kurs. Eine Naeherung; die Bank rechnet mit ihrem eigenen Kurs.
|
||
|
||
Nur in Stufe 1 gilt der Betrag als bekannt. Daran haengt, ob bei einem
|
||
angebundenen Zielkonto nachgefragt wird: Ein gerechneter Betrag laesst sich
|
||
beim naechsten Abruf nicht als dieselbe Bewegung wiedererkennen.
|
||
"""
|
||
if ziel.currency == konto.currency:
|
||
return betrag, None, True
|
||
|
||
if (eintrag is not None and eintrag.counter_amount_cents
|
||
and (eintrag.counter_currency or "").upper() == ziel.currency.upper()):
|
||
return abs(eintrag.counter_amount_cents), None, True
|
||
|
||
kurs = float(eintrag.fx_rate) if eintrag is not None and eintrag.fx_rate else None
|
||
umgerechnet = currency.convert(db, betrag, konto.currency, ziel.currency, datum, kurs)
|
||
return umgerechnet, kurs, False
|
||
|
||
|
||
def _pruefe(db: Session, user: User, konto: Account, ziel: Account,
|
||
exakt: bool, trotzdem: bool) -> None:
|
||
if ziel.id == konto.id:
|
||
raise UebertragFehler("Ein Übertrag braucht ein anderes Zielkonto")
|
||
if exakt or trotzdem:
|
||
return
|
||
if not ist_angebunden(db, user, ziel):
|
||
return
|
||
raise UebertragFehler(
|
||
f"Der Betrag auf '{ziel.name}' ist eine Umrechnung über den EZB-Kurs, "
|
||
f"nicht der Betrag, den die Bank meldet. Da das Konto selbst angebunden "
|
||
f"ist, kommt der Zugang von dort noch einmal – und liesse sich mit einem "
|
||
f"geschätzten Betrag nicht als derselbe erkennen. Sicherer: hier als "
|
||
f"Ausgabe buchen und beide Seiten danach verbinden lassen.",
|
||
nachfragen=True,
|
||
)
|
||
|
||
|
||
def _lege_an(db: Session, user: User, konto: Account, ziel: Account,
|
||
betrag_signiert: int, ziel_betrag: int | None, datum,
|
||
payee: str | None, beschreibung: str | None,
|
||
kurs: float | None = None) -> Transaction:
|
||
"""Die Uebertrags-Buchung, richtig herum.
|
||
|
||
Bei einem Zugang kennen wir den Betrag auf DIESEM Konto exakt – der gehoert
|
||
dann auf die Zielseite der Buchung, und der Abgang wird umgerechnet.
|
||
"""
|
||
betrag = abs(betrag_signiert)
|
||
if betrag_signiert < 0:
|
||
von, nach = konto, ziel
|
||
von_betrag, nach_betrag = betrag, ziel_betrag
|
||
else:
|
||
von, nach = ziel, konto
|
||
von_betrag, nach_betrag = (ziel_betrag if ziel_betrag is not None else betrag), betrag
|
||
|
||
txn = Transaction(
|
||
owner_id=user.id,
|
||
type=TxnType.transfer,
|
||
date=datum,
|
||
account_id=von.id,
|
||
to_account_id=nach.id,
|
||
amount_cents=von_betrag,
|
||
to_amount_cents=nach_betrag,
|
||
fx_rate=kurs,
|
||
payee=payee,
|
||
description=beschreibung,
|
||
)
|
||
buchung.setze_betraege(db, txn, von, nach)
|
||
db.add(txn)
|
||
db.flush()
|
||
return txn
|
||
|
||
|
||
def _gegenseite(db: Session, user: User, txn: Transaction, ziel: Account,
|
||
betrag_signiert: int, ziel_betrag: int, datum) -> InboxItem:
|
||
"""Den Eingangs-Eintrag auf dem Zielkonto – verbrauchen oder anlegen.
|
||
|
||
Verbraucht wird ein noch OFFENER Eintrag: Er beschreibt dieselbe Bewegung,
|
||
nur von der anderen Seite gesehen, und wuerde sonst ein zweites Mal gebucht.
|
||
"""
|
||
# Auf dem Zielkonto ist die Bewegung anders herum.
|
||
gesucht = -1 if betrag_signiert > 0 else 1
|
||
gesucht *= ziel_betrag
|
||
|
||
offen = [
|
||
(eid, edatum, ebetrag)
|
||
for eid, edatum, ebetrag in db.query(
|
||
InboxItem.id, InboxItem.date, InboxItem.amount_cents
|
||
).filter(
|
||
InboxItem.owner_id == user.id,
|
||
InboxItem.account_id == ziel.id,
|
||
InboxItem.status == InboxStatus.pending,
|
||
)
|
||
]
|
||
# Derselbe Abgleich wie beim Import – nicht eine zweite Vergleichslogik.
|
||
treffer = csv_import.finde_eingangs_treffer(offen, set(), datum, gesucht)
|
||
|
||
eintrag = db.get(InboxItem, treffer) if treffer is not None else None
|
||
if eintrag is None:
|
||
eintrag = InboxItem(
|
||
owner_id=user.id,
|
||
source=InboxSource.transfer,
|
||
account_id=ziel.id,
|
||
date=datum,
|
||
amount_cents=gesucht,
|
||
currency=ziel.currency,
|
||
payee=txn.payee,
|
||
note=txn.description,
|
||
)
|
||
db.add(eintrag)
|
||
eintrag.status = InboxStatus.booked
|
||
eintrag.transaction_id = txn.id
|
||
# Ein Uebertrag traegt keine Kategorie – auch nicht auf der Gegenseite.
|
||
eintrag.category_id = None
|
||
db.flush()
|
||
return eintrag
|
||
|
||
|
||
def aus_eingang(db: Session, user: User, eintrag: InboxItem, ziel: Account,
|
||
trotzdem: bool = False) -> Transaction:
|
||
"""Einen offenen Vorschlag als Kontouebertrag buchen."""
|
||
if eintrag.status == InboxStatus.booked or eintrag.transaction_id is not None:
|
||
raise UebertragFehler("Dieser Eintrag ist schon gebucht")
|
||
if eintrag.account_id is None:
|
||
raise UebertragFehler("Kein Konto zugeordnet")
|
||
konto = db.get(Account, eintrag.account_id)
|
||
if konto is None or konto.owner_id != user.id:
|
||
raise UebertragFehler("Kein Konto zugeordnet")
|
||
|
||
betrag = abs(eintrag.amount_cents)
|
||
ziel_betrag, kurs, exakt = _gegenbetrag(db, konto, ziel, betrag, eintrag.date, eintrag)
|
||
_pruefe(db, user, konto, ziel, exakt, trotzdem)
|
||
_pruefe_gegenseite(
|
||
db, user, konto, ziel,
|
||
TxnType.expense if eintrag.amount_cents < 0 else TxnType.income,
|
||
ziel_betrag, eintrag.date, exakt, trotzdem)
|
||
|
||
txn = _lege_an(db, user, konto, ziel, eintrag.amount_cents, ziel_betrag,
|
||
eintrag.date, eintrag.payee, eintrag.note, kurs)
|
||
eintrag.transaction_id = txn.id
|
||
eintrag.status = InboxStatus.booked
|
||
eintrag.category_id = None
|
||
_gegenseite(db, user, txn, ziel, eintrag.amount_cents,
|
||
ziel_betrag if ziel_betrag is not None else betrag, eintrag.date)
|
||
return txn
|
||
|
||
|
||
def aus_buchung(db: Session, user: User, txn: Transaction, ziel: Account,
|
||
trotzdem: bool = False) -> Transaction:
|
||
"""Eine schon gebuchte Ausgabe oder Einnahme zum Uebertrag machen.
|
||
|
||
Nicht ueber den PATCH-Weg des Buchungs-Routers: Der dreht die Richtung bei
|
||
einer Einnahme nicht um und legt keine Gegenseite an.
|
||
"""
|
||
if txn.type == TxnType.transfer:
|
||
raise UebertragFehler("Diese Buchung ist bereits ein Übertrag")
|
||
konto = db.get(Account, txn.account_id)
|
||
if konto is None or konto.owner_id != user.id:
|
||
raise UebertragFehler("Buchung ohne Konto")
|
||
|
||
# Ist sie schon mit ihrer Gegenseite verknuepft, sind BEIDE gebucht. Ein
|
||
# Uebertrag wuerde das Geld auf dem Zielkonto ein zweites Mal bewegen –
|
||
# und das laesst sich nicht mit "trotzdem" wegdruecken, es ist einfach
|
||
# falsch.
|
||
if _schon_verknuepft(db, user, txn):
|
||
raise UebertragFehler(
|
||
"Diese Buchung ist bereits mit ihrer Gegenseite verbunden – beide "
|
||
"Seiten sind gebucht und zählen schon jetzt in keiner Auswertung. "
|
||
"Ein Übertrag würde das Geld auf dem Zielkonto ein zweites Mal "
|
||
"bewegen. Wenn du es wirklich anders willst, löse zuerst die "
|
||
"Verknüpfung."
|
||
)
|
||
|
||
signiert = txn.amount_cents if txn.type == TxnType.income else -txn.amount_cents
|
||
# Der Eingangs-Eintrag traegt den Kurs der Bank – ohne ihn bliebe hier nur
|
||
# die EZB-Naeherung, auch wenn die Bank ihren eigenen genannt hat.
|
||
eintrag = _eingang_zu(db, user, txn)
|
||
ziel_betrag, kurs, exakt = _gegenbetrag(db, konto, ziel, abs(signiert),
|
||
txn.date, eintrag)
|
||
_pruefe(db, user, konto, ziel, exakt, trotzdem)
|
||
_pruefe_gegenseite(db, user, konto, ziel, txn.type, ziel_betrag, txn.date,
|
||
exakt, trotzdem)
|
||
|
||
# Die vorhandene Buchung umschreiben statt eine zweite anzulegen: Alles, was
|
||
# auf sie zeigt – Eingangs-Eintrag, Verknuepfungen – bleibt damit gueltig.
|
||
if signiert < 0:
|
||
von, nach = konto, ziel
|
||
txn.amount_cents, txn.to_amount_cents = abs(signiert), ziel_betrag
|
||
else:
|
||
von, nach = ziel, konto
|
||
txn.amount_cents = ziel_betrag if ziel_betrag is not None else abs(signiert)
|
||
txn.to_amount_cents = abs(signiert)
|
||
txn.type = TxnType.transfer
|
||
txn.account_id = von.id
|
||
txn.to_account_id = nach.id
|
||
txn.category_id = None
|
||
if kurs:
|
||
txn.fx_rate = kurs # der Kurs der Bank schlaegt die EZB
|
||
buchung.setze_betraege(db, txn, von, nach)
|
||
db.flush()
|
||
|
||
_gegenseite(db, user, txn, ziel, signiert,
|
||
ziel_betrag if ziel_betrag is not None else abs(signiert), txn.date)
|
||
return txn
|