fin/.claude/skills/fin-diagnostics-and-tooling/scripts/audit_linked_legs.py

#!/usr/bin/env python3
"""Audit linked transaction legs (transfers, settlements, splitwise pairs).

READ-ONLY. Usage: python3 audit_linked_legs.py /path/to/fin.db [--allow-legacy] [--json]

Ground truth: api/app/crud.py (create_transfer, create_settlement,
create_split_expense, create_splitwise_i_paid, create_funded_buy,
delete_transaction legacy fallback) and api/app/models.py.

Rules encoded:
- Every 'transfer'/'settlement' row must have a partner via linked_transaction_id,
  EXCEPT the funded-buy cash leg: a single 'transfer' row whose linked_account_id
  points to an investment/rsu account (create_funded_buy writes no partner row).
- Rows written before linked_transaction_id existed are "legacy": NULL link but
  linked_account_id set. delete_transaction pairs them by reversed accounts +
  same date + same type. --allow-legacy accepts legacy rows that have such a match.
- linked_transaction_id must resolve, must not be self, and for
  transfer/settlement/splitwise_pay the partner must link back (reciprocal).
  The split-expense 'expense' row links one-way into its transfer group by
  design and is exempt from reciprocity.
- Same-currency transfer/settlement pairs must cancel: amount_a + amount_b == 0
  (fees live in the fee column and are excluded). Cross-currency transfer legs
  legitimately differ (to_amount conversion) and are skipped.
- splitwise_pay pairs never cancel by design (bank leg = -total, splitwise leg
  = +partner_share) and are never amount-checked.

Exit 0 = clean, 1 = anomalies found.
"""
import argparse
import json
import sqlite3
import sys

EPS = 0.005
PARTNER_REQUIRED = ("transfer", "settlement")
RECIPROCAL = ("transfer", "settlement", "splitwise_pay")


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("db")
    ap.add_argument("--allow-legacy", action="store_true",
                    help="accept legacy NULL-link legs that have a crud-style fallback match")
    ap.add_argument("--json", action="store_true")
    args = ap.parse_args()

    con = sqlite3.connect(f"file:{args.db}?mode=ro", uri=True)
    con.row_factory = sqlite3.Row

    accounts = {r["id"]: dict(r) for r in con.execute(
        "SELECT id, name, type FROM accounts")}
    txns = {r["id"]: dict(r) for r in con.execute(
        "SELECT id, date, account_id, amount, currency, type, fee, "
        "linked_account_id, linked_transaction_id, description "
        "FROM transactions")}

    anomalies: list[dict] = []
    info = {"transactions": len(txns), "checked_pairs": 0,
            "cross_currency_pairs": 0, "funded_buy_legs": 0, "legacy_matched": 0}

    def flag(check, txn_id, detail):
        anomalies.append({"check": check, "txn_id": txn_id, "detail": detail})

    for t in txns.values():
        tid, ttype = t["id"], t["type"]
        if t["account_id"] not in accounts:
            flag("missing-account", tid,
                 f"account_id={t['account_id']} does not exist")
        la = t["linked_account_id"]
        if la is not None and la not in accounts:
            flag("missing-linked-account", tid,
                 f"linked_account_id={la} does not exist")
        if la is not None and la == t["account_id"]:
            flag("self-account-link", tid, "linked_account_id == account_id")

        lid = t["linked_transaction_id"]
        if lid is not None:
            if lid == tid:
                flag("self-link", tid, "linked_transaction_id == id")
                continue
            partner = txns.get(lid)
            if partner is None:
                flag("orphan-link", tid,
                     f"linked_transaction_id={lid} does not exist")
                continue
            if ttype in RECIPROCAL and partner["linked_transaction_id"] != tid:
                flag("non-reciprocal", tid,
                     f"partner {lid} links to {partner['linked_transaction_id']!r}, not back")
            if la is not None and partner["account_id"] != la:
                flag("partner-account-mismatch", tid,
                     f"partner {lid} lives on account {partner['account_id']}, "
                     f"but linked_account_id={la}")
            # amount cancellation — once per pair, same type, same currency only
            if (ttype in ("transfer", "settlement") and partner["type"] == ttype
                    and tid < lid):
                if t["currency"] == partner["currency"]:
                    info["checked_pairs"] += 1
                    resid = t["amount"] + partner["amount"]
                    if abs(resid) > EPS:
                        flag("amounts-dont-cancel", tid,
                             f"pair ({tid},{lid}): {t['amount']} + {partner['amount']} "
                             f"= {resid:+.2f} {t['currency']}")
                else:
                    info["cross_currency_pairs"] += 1
        elif ttype in PARTNER_REQUIRED:
            la_type = accounts.get(la, {}).get("type") if la is not None else None
            if ttype == "transfer" and la_type in ("investment", "rsu"):
                info["funded_buy_legs"] += 1  # single-leg by design
                continue
            legacy_ok = False
            if args.allow_legacy and la is not None:
                # replicate delete_transaction's legacy fallback match
                for o in txns.values():
                    if (o["id"] != tid
                            and o["account_id"] == la
                            and o["linked_account_id"] == t["account_id"]
                            and o["linked_transaction_id"] is None
                            and o["date"] == t["date"]
                            and o["type"] == ttype):
                        legacy_ok = True
                        break
            if legacy_ok:
                info["legacy_matched"] += 1
            else:
                flag("unlinked-leg", tid,
                     f"{ttype} with NULL linked_transaction_id "
                     f"(linked_account_id={la}, date={t['date']}, amount={t['amount']})")

    if args.json:
        print(json.dumps({"info": info, "anomalies": anomalies}, indent=2))
    else:
        print(f"audit_linked_legs: {info['transactions']} transactions, "
              f"{info['checked_pairs']} same-currency pairs amount-checked, "
              f"{info['cross_currency_pairs']} cross-currency pairs skipped, "
              f"{info['funded_buy_legs']} funded-buy single legs, "
              f"{info['legacy_matched']} legacy pairs accepted")
        for a in anomalies:
            print(f"ANOMALY [{a['check']}] txn {a['txn_id']}: {a['detail']}")
        print("CLEAN" if not anomalies else f"{len(anomalies)} anomalies")
    return 1 if anomalies else 0


if __name__ == "__main__":
    sys.exit(main())