bin/finance/scripts/parser_vs_csv.py

"""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))