#!/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") ap.add_argument( "--write", action="store_true", help="actually insert the splits (settled=true). Without this, nothing is written.", ) 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.write: # Imported as settled: these obligations were discharged on a platform # we no longer run, and the residual is already carried by transaction # 2348. Writing them unsettled would re-open ~$40k of debts that were # paid years ago. See ACTIVE_OBLIGATION in analytics-sql.ts -- settled # rows stay in spend and leave every owed figure, which is exactly the # point: this import exists to correct historical SPEND. SETTLED_ON = "2026-01-09" # the carryover's date written = 0 for row, tx in rep.matched: if tx["has_split"]: continue payer_share = round(100 - row.ower_share, 2) pairs = [(row.ower, row.ower_share)] if payer_share > 0: pairs.append((row.payer, payer_share)) for pid, share in pairs: cur.execute( """ INSERT INTO transaction_splits (transaction_id, participant_id, share_percent, settled, settled_at) VALUES (%s, %s, %s, true, %s) ON CONFLICT (transaction_id, participant_id) DO NOTHING """, (tx["id"], pid, share, SETTLED_ON), ) written += cur.rowcount conn.commit() print(f"\nWROTE {written} split rows (settled=true, settled_at={SETTLED_ON}).") 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())