feat(scripts): dry-run matcher for the SplitMyExpenses history

Matches the five CSV exports against transactions already in the ledger and
reports what it would do. Writes nothing -- importing is a separate step, and
rehearsing it first is what catches the defects that tests do not.

What it found, and why the number is what it is: 676 of 1,536 shareable rows
match (44%). The ceiling is ledger coverage, not matcher quality. The CSVs
describe 678 shared expenses in 2024 alone; the ledger holds 591 rows for the
whole of that year, 3 to 72 a month, which is far less than a household
actually spends. Most 2024 CSV rows have no transaction to attach a split to
and never will. South Korea April 2024 matches 4 of 158 for that reason.

Three decisions are encoded deliberately:

  - Date format is decided per FILE, not per row. The household export writes
    D/M/YYYY and the four trip exports write ISO, and 474 rows parse validly
    under both readings -- per-row guessing silently swaps January and February
    for some rows and not others.
  - A person's column is their net balance impact, not their share. The payer
    is whoever is positive; the other's share is |their negative| / cost. So a
    +cost/-cost row means the other party owes 100%, not that the expense was
    unshared -- the reading that would fake an arrangement change.
  - Matching is one-to-one, best pair first. The NZ trip has two identical
    $10.16 Uber rows against three ledger rows and four PayMyPark rows in the
    same shape; without this a ledger row is claimed repeatedly and the second
    CSV row looks matched while being unrepresented.

scripts/split_csv_match.py

@@ -0,0 +1,342 @@
+#!/usr/bin/env python3
+"""Match SplitMyExpenses CSV rows to transactions already in the ledger.
+
+Dry-run by default. It prints what it would do and writes nothing; `--write`
+is a separate step (task #9) and is deliberately not implemented here.
+
+Why this exists
+---------------
+The CSVs are the record of how expenses were actually shared before this app
+existed. Importing them is what makes historical *spend* correct: without a
+split row, a $200 grocery shop counts as $200 of my spending when half of it
+was never mine. The balances are already settled by carryover transaction 2348,
+so these splits are imported as `settled = true` and move no balance.
+
+Three things make the matching harder than "same date, same amount":
+
+1.  **Dates are ambiguous across files.** The household file writes D/M/YYYY;
+    the four trip files write ISO. 474 rows parse validly under both readings,
+    so the format is decided per file, from the file, and never guessed per row.
+
+2.  **The sign convention is not "who paid".** A person's column is their net
+    balance impact: positive means they are owed. So the payer is whoever is
+    positive, and the other person's share is |their negative| / cost. A row
+    reading +cost / -cost therefore means the other party owes 100% -- NOT that
+    the expense was unshared, which is the reading that would fake an
+    arrangement change.
+
+3.  **A settlement is not an expense.** Rows where one person hands the other
+    money must not become split transactions; they are already represented by
+    the carryover.
+
+Usage:
+    .venv/bin/python scripts/split_csv_match.py [--verbose] [--file NAME]
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import glob
+import os
+import re
+import sys
+from collections import Counter
+from dataclasses import dataclass, field
+from datetime import date, datetime, timedelta
+
+import psycopg2
+import psycopg2.extras
+
+DUMP_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "dump")
+
+# Read from the same place the app does rather than hardcoding a container IP,
+# which changes on every recreate.
+def db_url() -> str:
+    url = os.environ.get("DATABASE_URL")
+    if url:
+        return url
+    for envfile in (".env", ".env.test"):
+        path = os.path.join(os.path.dirname(DUMP_DIR), envfile)
+        if not os.path.exists(path):
+            continue
+        for line in open(path):
+            if line.startswith("DATABASE_URL"):
+                return line.split("=", 1)[1].strip().strip('"').strip("'")
+    sys.exit("No DATABASE_URL found (env, .env, or .env.test)")
+
+
+# The two people in these files. The CSV writes full names; the ledger uses
+# first names.
+CSV_ME = "Siddharth Bose"
+CSV_THEM = "Meghalee"
+PARTICIPANT_ME = 1
+PARTICIPANT_THEM = 4
+
+# A settlement transfers money; it is not a shared cost. These are the
+# descriptions SplitMyExpenses uses for them.
+SETTLEMENT_PAT = re.compile(
+    r"payment|settle|debts? remainder|reimburse|transfer to|paid back", re.I
+)
+
+
+@dataclass
+class CsvRow:
+    source: str
+    line: int
+    when: date
+    description: str
+    category: str
+    cost: float
+    currency: str
+    net_me: float
+    net_them: float
+
+    # Filled in by classify()
+    mode: str = ""
+    payer: int = 0            # participant id who paid
+    ower: int = 0             # participant id who owes
+    ower_share: float = 0.0   # 0-100
+
+    def __str__(self) -> str:
+        return f"{self.source}:{self.line} {self.when} {self.description[:38]!r} ${self.cost:.2f} [{self.mode}]"
+
+
+@dataclass
+class MatchReport:
+    rows: list = field(default_factory=list)
+    matched: list = field(default_factory=list)
+    ambiguous: list = field(default_factory=list)
+    unmatched: list = field(default_factory=list)
+    skipped: list = field(default_factory=list)
+
+
+def sniff_date_format(sample: list[str]) -> str:
+    """Decide ISO vs D/M/YYYY for a whole file.
+
+    Deciding per row is what produces a ledger where January and February are
+    silently swapped for some rows and not others. A file is written by one
+    exporter in one format, so the file is the unit of decision.
+    """
+    if not sample:
+        return "%Y-%m-%d"
+    slashes = sum(1 for s in sample if "/" in s)
+    return "%d/%m/%Y" if slashes > len(sample) / 2 else "%Y-%m-%d"
+
+
+def parse_rows(path: str) -> list[CsvRow]:
+    with open(path, newline="", encoding="utf-8-sig") as fh:
+        reader = list(csv.DictReader(fh))
+    if not reader:
+        return []
+
+    # Header names vary in quoting between exports.
+    def col(row: dict, *names: str):
+        for n in names:
+            for k in row:
+                if k.strip().strip('"') == n:
+                    return row[k]
+        return None
+
+    fmt = sniff_date_format([col(r, "Date") or "" for r in reader[:40]])
+    out: list[CsvRow] = []
+    for i, r in enumerate(reader, start=2):
+        raw_date = (col(r, "Date") or "").strip()
+        try:
+            when = datetime.strptime(raw_date, fmt).date()
+        except ValueError:
+            continue
+        try:
+            cost = float(col(r, "Cost") or 0)
+            net_me = float(col(r, CSV_ME) or 0)
+            net_them = float(col(r, CSV_THEM) or 0)
+        except ValueError:
+            continue
+        out.append(
+            CsvRow(
+                source=os.path.basename(path),
+                line=i,
+                when=when,
+                description=(col(r, "Description") or "").strip(),
+                category=(col(r, "Category") or "").strip(),
+                cost=cost,
+                currency=(col(r, "Currency") or "AUD").strip(),
+                net_me=net_me,
+                net_them=net_them,
+            )
+        )
+    return out
+
+
+def classify(row: CsvRow) -> CsvRow:
+    """Work out who paid and what share the other person owes.
+
+    A person's column is their net balance impact, not their share: positive
+    means they are owed money. So the payer is whoever is positive.
+    """
+    if row.cost == 0:
+        row.mode = "zero-cost"
+        return row
+
+    if SETTLEMENT_PAT.search(row.description):
+        row.mode = "settlement"
+        return row
+
+    # Both zero against a real cost: recorded but not shared.
+    if abs(row.net_me) < 0.005 and abs(row.net_them) < 0.005:
+        row.mode = "unshared"
+        return row
+
+    if row.net_me > 0:
+        row.payer, row.ower, owed = PARTICIPANT_ME, PARTICIPANT_THEM, abs(row.net_them)
+    else:
+        row.payer, row.ower, owed = PARTICIPANT_THEM, PARTICIPANT_ME, abs(row.net_me)
+
+    row.ower_share = round(owed / row.cost * 100, 2)
+    if abs(row.ower_share - 50) < 0.6:
+        row.mode = "50/50"
+    elif abs(row.ower_share - 100) < 0.6:
+        row.mode = "other-owes-all"
+    else:
+        row.mode = f"uneven-{row.ower_share:.0f}"
+    return row
+
+
+def norm(s: str) -> set[str]:
+    return {w for w in re.split(r"[^a-z0-9]+", (s or "").lower()) if len(w) > 2}
+
+
+def score(row: CsvRow, tx: dict) -> float:
+    """How well a ledger row matches a CSV row. Amount and date gate it;
+    description only ranks among survivors."""
+    days = abs((tx["transaction_date"] - row.when).days)
+    s = 100.0 - days * 4
+    overlap = norm(row.description) & (norm(tx["description"]) | norm(tx["merchant_normalized"]))
+    s += 12 * len(overlap)
+    return s
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--verbose", action="store_true")
+    ap.add_argument("--file", help="only this CSV (substring match)")
+    ap.add_argument("--window", type=int, default=5, help="date tolerance in days")
+    args = ap.parse_args()
+
+    paths = sorted(glob.glob(os.path.join(DUMP_DIR, "*SplitMyExpenses*.csv")))
+    if args.file:
+        paths = [p for p in paths if args.file in os.path.basename(p)]
+    if not paths:
+        sys.exit("No SplitMyExpenses CSVs found in dump/")
+
+    conn = psycopg2.connect(db_url())
+    cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
+    # Superseded rows are duplicates; matching against them would attach a split
+    # to a row nothing else counts.
+    cur.execute(
+        """
+        SELECT t.id, t.transaction_date, COALESCE(t.description,'') AS description,
+               COALESCE(t.merchant_normalized,'') AS merchant_normalized,
+               COALESCE(t.amount_aud, t.amount)::float AS amount,
+               t.superseded_by_id,
+               EXISTS(SELECT 1 FROM transaction_splits x WHERE x.transaction_id=t.id) AS has_split
+        FROM transactions t
+        LEFT JOIN statements s ON s.id = t.statement_id
+        WHERE COALESCE(t.owner_id, s.owner_id) IN (%s, %s)
+          AND t.superseded_by_id IS NULL
+          AND NOT (t.statement_id IS NULL AND t.reconciled_with_id IS NOT NULL)
+          AND t.transaction_type IN ('debit','fee','interest')
+        """,
+        (PARTICIPANT_ME, PARTICIPANT_THEM),
+    )
+    txs = cur.fetchall()
+
+    # Index by rounded amount: the amount must agree, so it is the only cheap
+    # gate that never needs fuzzy comparison.
+    by_amount: dict[float, list[dict]] = {}
+    for t in txs:
+        by_amount.setdefault(round(t["amount"], 2), []).append(t)
+
+    rep = MatchReport()
+    modes: Counter = Counter()
+    candidates: list = []
+    for path in paths:
+        for row in parse_rows(path):
+            classify(row)
+            rep.rows.append(row)
+            modes[row.mode] += 1
+
+            if row.mode in ("settlement", "zero-cost", "unshared"):
+                rep.skipped.append(row)
+                continue
+            if row.currency != "AUD":
+                rep.skipped.append(row)
+                continue
+
+            cands = [
+                t for t in by_amount.get(round(row.cost, 2), [])
+                if abs((t["transaction_date"] - row.when).days) <= args.window
+            ]
+            if not cands:
+                rep.unmatched.append(row)
+            else:
+                candidates.append((row, cands))
+
+    # Assign one-to-one, best pair first.
+    #
+    # Without this a ledger row can be claimed by several CSV rows. That is not
+    # hypothetical: the NZ trip has two identical $10.16 Uber trips on one day
+    # and three matching ledger rows, and four PayMyPark rows in the same shape.
+    # Attaching a split twice is harmless (the unique key absorbs it) but it
+    # leaves the second CSV row silently unrepresented while looking matched,
+    # which is a lie in the report rather than a defect in the data.
+    scored = sorted(
+        ((score(row, t), row, t) for row, cands in candidates for t in cands),
+        key=lambda x: -x[0],
+    )
+    taken_tx: set[int] = set()
+    taken_row: set[int] = set()
+    for s, row, t in scored:
+        if id(row) in taken_row or t["id"] in taken_tx:
+            continue
+        taken_row.add(id(row))
+        taken_tx.add(t["id"])
+        rep.matched.append((row, t))
+    for row, cands in candidates:
+        if id(row) not in taken_row:
+            rep.ambiguous.append((row, cands))
+
+    total = len(rep.rows)
+    considered = total - len(rep.skipped)
+    print(f"CSV rows              {total}")
+    print(f"  skipped             {len(rep.skipped)}  (settlements, zero-cost, unshared, non-AUD)")
+    print(f"  considered          {considered}")
+    print(f"    matched           {len(rep.matched)}  ({len(rep.matched)/max(considered,1)*100:.1f}%)")
+    print(f"    ambiguous         {len(rep.ambiguous)}")
+    print(f"    unmatched         {len(rep.unmatched)}")
+    print()
+    print("Row modes:")
+    for m, n in modes.most_common():
+        print(f"  {m:<18} {n}")
+
+    already = sum(1 for _, t in rep.matched if t["has_split"])
+    print()
+    print(f"Of the matched, {already} already carry a split and would be left alone;")
+    print(f"{len(rep.matched) - already} would gain one.")
+
+    if args.verbose:
+        print("\n--- ambiguous ---")
+        for row, cands in rep.ambiguous[:40]:
+            print(f"  {row}")
+            for t in cands[:3]:
+                print(f"      -> #{t['id']} {t['transaction_date']} {t['description'][:44]!r}")
+        print("\n--- unmatched ---")
+        for row in rep.unmatched[:60]:
+            print(f"  {row}")
+
+    conn.close()
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())