fix: DKB-Empfaenger/Zweck-Split per CSV-Ground-Truth, Vergleichswerkzeug

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

finance/scripts/parser_vs_csv.py

@@ -0,0 +1,166 @@
+"""Vergleichswerkzeug: PDF-Parser-Ergebnis gegen CSV-Ground-Truth (Task 5, Ausbaustufe 3).
+
+NUR LOKAL VERWENDEN - die Ausgabe enthaelt echte Kontodaten (Betraege,
+Gegenparteien, Verwendungszwecke) und darf nicht in Commits, Reports oder
+Tickets uebernommen werden.
+
+Matcht die aus einem Kontoauszugs-PDF geparsten Transaktionen gegen die
+Zeilen einer oder mehrerer Kontoumsatz-CSV-Dateien (auf den vom PDF
+abgedeckten Zeitraum eingeschraenkt) anhand von (booking_date, amount) und
+vergleicht je Match Gegenpartei (counterparty) und Verwendungszweck
+(purpose) mittels difflib.SequenceMatcher.
+
+Aufruf:
+    PYTHONPATH=. .venv/bin/python scripts/parser_vs_csv.py <pdf> <csv> [<csv>...]
+"""
+import statistics
+import sys
+from collections import defaultdict
+from dataclasses import dataclass
+from datetime import date
+from decimal import Decimal
+from difflib import SequenceMatcher
+from pathlib import Path
+
+from app.parsers.base import ParsedTransaction
+from app.parsers.csv_formats import parse_csv
+from app.parsers.registry import parse_pdf
+
+TRUNC = 60
+
+
+@dataclass
+class CsvRow:
+    booking_date: date
+    amount: Decimal
+    counterparty: str
+    purpose: str
+
+
+def _norm(s: str) -> str:
+    return " ".join((s or "").casefold().split())
+
+
+def _sim(a: str, b: str) -> float:
+    return SequenceMatcher(None, _norm(a), _norm(b)).ratio()
+
+
+def _trunc(s: str) -> str:
+    s = s or ""
+    return s if len(s) <= TRUNC else s[: TRUNC - 1] + "…"
+
+
+def load_csv_rows(paths: list[Path], period_start: date | None,
+                   period_end: date | None) -> list[CsvRow]:
+    rows: list[CsvRow] = []
+    for p in paths:
+        parsed = parse_csv(p)
+        for t in parsed.statement.transactions:
+            if period_start is not None and t.booking_date < period_start:
+                continue
+            if period_end is not None and t.booking_date > period_end:
+                continue
+            rows.append(CsvRow(booking_date=t.booking_date, amount=t.amount,
+                                counterparty=t.counterparty, purpose=t.purpose))
+    return rows
+
+
+def match(pdf_txs: list[ParsedTransaction], csv_rows: list[CsvRow]):
+    """Greedy 1:1 match auf (booking_date, amount); Reihenfolge = PDF-Reihenfolge.
+
+    Returns (matches, unmatched_pdf, unmatched_csv, ambiguous_keys).
+    matches: list[tuple[ParsedTransaction, CsvRow]]
+    """
+    buckets: dict[tuple[date, Decimal], list[CsvRow]] = defaultdict(list)
+    for r in csv_rows:
+        buckets[(r.booking_date, r.amount)].append(r)
+
+    ambiguous_keys = {k for k, v in buckets.items() if len(v) > 1}
+
+    matches: list[tuple[ParsedTransaction, CsvRow]] = []
+    unmatched_pdf: list[ParsedTransaction] = []
+    for t in pdf_txs:
+        key = (t.booking_date, t.amount)
+        bucket = buckets.get(key)
+        if bucket:
+            row = bucket.pop(0)
+            matches.append((t, row))
+        else:
+            unmatched_pdf.append(t)
+
+    unmatched_csv = [r for bucket in buckets.values() for r in bucket]
+    return matches, unmatched_pdf, unmatched_csv, ambiguous_keys
+
+
+def purpose_sim(pdf_purpose: str, csv_counterparty: str, csv_purpose: str) -> float:
+    direct = _sim(pdf_purpose, csv_purpose)
+    combined = _sim(pdf_purpose, f"{csv_counterparty} {csv_purpose}")
+    return max(direct, combined)
+
+
+def main(argv: list[str]) -> int:
+    if len(argv) < 3:
+        print(f"Usage: {argv[0]} <pdf> <csv> [<csv>...]", file=sys.stderr)
+        return 2
+
+    pdf_path = Path(argv[1])
+    csv_paths = [Path(p) for p in argv[2:]]
+
+    stmt = parse_pdf(pdf_path)
+    csv_rows = load_csv_rows(csv_paths, stmt.period_start, stmt.period_end)
+
+    print(f"PDF: {pdf_path.name}  bank={stmt.bank}  "
+          f"period={stmt.period_start}..{stmt.period_end}  n_pdf={len(stmt.transactions)}")
+    print(f"CSV: {', '.join(p.name for p in csv_paths)}  "
+          f"n_csv(in Periode)={len(csv_rows)}")
+    print("-" * 100)
+
+    matches, unmatched_pdf, unmatched_csv, ambiguous_keys = match(
+        stmt.transactions, csv_rows)
+
+    cp_sims: list[float] = []
+    p_sims: list[float] = []
+
+    for i, (t, r) in enumerate(matches, start=1):
+        cp_s = _sim(t.counterparty, r.counterparty)
+        p_s = purpose_sim(t.purpose, r.counterparty, r.purpose)
+        cp_sims.append(cp_s)
+        p_sims.append(p_s)
+        print(f"#{i} | {t.booking_date} | {t.amount:>10} "
+              f"| PDF-cp={_trunc(t.counterparty)} | CSV-cp={_trunc(r.counterparty)} "
+              f"| cp-sim={cp_s:.2f} "
+              f"| PDF-p={_trunc(t.purpose)} | CSV-p={_trunc(r.purpose)} "
+              f"| p-sim={p_s:.2f}")
+
+    print("-" * 100)
+    if unmatched_pdf:
+        print(f"Unmatched PDF-Transaktionen ({len(unmatched_pdf)}):")
+        for t in unmatched_pdf:
+            print(f"  {t.booking_date} | {t.amount:>10}")
+    if unmatched_csv:
+        print(f"Unmatched CSV-Zeilen ({len(unmatched_csv)}):")
+        for r in unmatched_csv:
+            print(f"  {r.booking_date} | {r.amount:>10}")
+
+    print("=" * 100)
+    print("Summary")
+    print(f"  n_pdf            = {len(stmt.transactions)}")
+    print(f"  n_csv(in Periode) = {len(csv_rows)}")
+    print(f"  n_matched        = {len(matches)}")
+    print(f"  ambiguities      = {len(ambiguous_keys)} "
+          f"(Buchungs-Keys mit >1 CSV-Zeile gleichen Datums/Betrags)")
+    if cp_sims:
+        print(f"  cp-sim mean/median = {statistics.mean(cp_sims):.3f} / "
+              f"{statistics.median(cp_sims):.3f}")
+        print(f"  p-sim  mean/median = {statistics.mean(p_sims):.3f} / "
+              f"{statistics.median(p_sims):.3f}")
+        low_cp = sum(1 for s in cp_sims if s < 0.5)
+        print(f"  cp-sim < 0.5     = {low_cp} von {len(matches)} Matches")
+    else:
+        print("  (keine Matches)")
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(sys.argv))