""" Afton Bowen-Family Transfer Audit ================================== Comprehensive audit of all transfer activity between Afton's RBFCU 9015 and external accounts (J. Bowen, T. Bowen, other accounts) over the bank CSV period (12/01/25 - 5/11/26). Cross-references each transfer against book transactions on acct 61 to identify: - Transfers IN (owner contributions) — are they booked? to what account? - Transfers OUT (owner distributions / family transfers) — are they booked? to what? Output: afton_transfers_audit_report.md """ from __future__ import annotations import csv import json import re import sys from collections import defaultdict from datetime import datetime from pathlib import Path if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8", errors="replace") SCRIPT_DIR = Path(__file__).resolve().parent QBO_APP_DIR = SCRIPT_DIR.parent.parent / "Atlas" / "app" sys.path.insert(0, str(QBO_APP_DIR)) import qbo_client as qbo # type: ignore import requests CLIENT = "afton" ACCT_61 = "61" BANK_CSV = SCRIPT_DIR / "20260513-8529015.CSV" START = "2025-12-01" END = "2026-05-11" OUT_MD = SCRIPT_DIR / "afton_transfers_audit_report.md" OUT_JSON = SCRIPT_DIR / "afton_transfers_audit.json" def parse_bank_csv(path: Path) -> list[dict]: txns = [] with open(path, encoding="utf-8") as f: for row in csv.DictReader(f): amt = float(row["Amount"]) d = datetime.strptime(row["Post Date"], "%m/%d/%Y").date() txns.append({ "date": d, "amount": amt, "payee": row.get("Payee", ""), "memo": row.get("Memo", ""), }) return txns def categorize_transfer(payee: str, amount: float) -> str: p = payee.upper() if "TRANSFER" not in p and "INTERNET" not in p: return None # Not a transfer direction = "IN" if amount > 0 else "OUT" if "J. BOWEN" in p or "J BOWEN" in p: return f"J. Bowen ({direction})" if "T. BOWEN" in p or "T BOWEN" in p: return f"T. Bowen ({direction})" if "*3905" in p: return f"Acct 3905 ({direction})" if "*4778" in p: return f"Acct 4778 ({direction})" if "*7223" in p: return f"Acct 7223 ({direction})" if "*3507" in p: return f"Acct 3507 / J. Bowen ({direction})" if "*1467" in p: return f"Acct 1467 CML ({direction})" if "*0202" in p: return f"Acct 0202 ({direction})" return f"Other transfer ({direction})" def pull_book_gl(token: str) -> list[dict]: cfg = qbo.get_client(CLIENT) realm = cfg["realm_id"] url = f"{qbo.api_base()}/v3/company/{realm}/reports/GeneralLedger" resp = requests.get(url, params={ "account": ACCT_61, "start_date": START, "end_date": END, "minorversion": "73", "columns": "tx_date,txn_type,doc_num,name,memo,split_acc,subt_nat_amount", }, headers={"Authorization": f"Bearer {token}", "Accept": "application/json"}, timeout=120) data = resp.json() txns = [] def walk(rows): for r in rows or []: if "Rows" in r: walk(r["Rows"].get("Row", [])) else: cd = r.get("ColData", []) if len(cd) < 7: continue txn_type = cd[1].get("value", "") if not txn_type or txn_type == "Beginning Balance" or "Total" in txn_type: continue try: d = datetime.strptime(cd[0].get("value", ""), "%Y-%m-%d").date() except ValueError: continue try: amt = float(cd[6].get("value", "0").replace(",", "")) except ValueError: continue txns.append({ "date": d, "amount": amt, "txn_type": txn_type, "doc_num": cd[2].get("value", ""), "name": cd[3].get("value", ""), "memo": cd[4].get("value", ""), "split_acc": cd[5].get("value", ""), }) walk(data.get("Rows", {}).get("Row", [])) return txns def main() -> None: token = qbo.refresh_access_token(CLIENT) print("Pulling bank CSV...") bank = parse_bank_csv(BANK_CSV) start_d = datetime.fromisoformat(START).date() end_d = datetime.fromisoformat(END).date() bank = [b for b in bank if start_d <= b["date"] <= end_d] print(f" {len(bank)} bank txns total in period") # Filter to transfers transfers = [] for b in bank: cat = categorize_transfer(b["payee"], b["amount"]) if cat: b["category"] = cat transfers.append(b) print(f" {len(transfers)} bank transfer txns") # Group by category cat_groups = defaultdict(list) for t in transfers: cat_groups[t["category"]].append(t) print("\nPulling book GL...") book = pull_book_gl(token) book_transfers = [b for b in book if "transfer" in b["txn_type"].lower() or "transfer" in b["memo"].lower() or "transfer" in b["name"].lower()] print(f" {len(book)} book txns, of which {len(book_transfers)} look transfer-related") # Match each bank transfer against book transactions (by date+amount ±3 days) book_used: set = set() matched_pairs = [] unmatched_bank = [] for t in transfers: best = None for i, b in enumerate(book): if i in book_used: continue if abs(b["amount"] - t["amount"]) > 0.01: continue day_delta = abs((b["date"] - t["date"]).days) if day_delta > 3: continue if best is None or day_delta < best[1]: best = (i, day_delta) if best: book_used.add(best[0]) matched_pairs.append({"bank": t, "book": book[best[0]], "day_delta": best[1]}) else: unmatched_bank.append(t) # ── Output ──────────────────────────────────────────────────────────── lines = [ "# Afton — Bowen-Family / Transfer Audit", f"_Period: {START} to {END}_", f"_Generated: {datetime.utcnow().isoformat()}Z_", f"_Source: bank CSV ({BANK_CSV.name})_", "", f"Total bank transfers in period: **{len(transfers)}** totaling **${sum(t['amount'] for t in transfers):,.2f}** net", "", "## Summary by category", "", "| Category | Count | $ Total | Notes |", "|---|---|---|---|", ] for cat in sorted(cat_groups, key=lambda c: -sum(t['amount'] for t in cat_groups[c]) if any(t['amount']>0 for t in cat_groups[c]) else sum(t['amount'] for t in cat_groups[c])): ts = cat_groups[cat] total = sum(t["amount"] for t in ts) notes = "" if "(IN)" in cat: notes = "INFLOW — potential owner contribution / external funding" elif "(OUT)" in cat: notes = "OUTFLOW — potential distribution / family transfer" lines.append(f"| {cat} | {len(ts)} | ${total:,.2f} | {notes} |") lines.append("") # Detail per category for cat in sorted(cat_groups): ts = cat_groups[cat] total = sum(t["amount"] for t in ts) lines.append(f"## {cat} — {len(ts)} txns, ${total:,.2f}") lines.append("") lines.append("| Date | Amount | Payee | Book match? | Book split-acct |") lines.append("|---|---|---|---|---|") for t in sorted(ts, key=lambda x: x["date"]): # Find if this is matched match = next((m for m in matched_pairs if m["bank"] is t), None) if match: book_split = match["book"].get("split_acc", "?") match_str = f"✓ matched ({match['day_delta']}d)" else: book_split = "—" match_str = "✗ UNMATCHED" lines.append(f"| {t['date']} | ${t['amount']:,.2f} | {t['payee'][:50]} | {match_str} | {book_split} |") lines.append("") OUT_MD.write_text("\n".join(lines), encoding="utf-8") OUT_JSON.write_text(json.dumps({ "transfers": [{**t, "date": t["date"].isoformat()} for t in transfers], "matched": len(matched_pairs), "unmatched_bank": len(unmatched_bank), "by_category": {cat: {"count": len(ts), "total": sum(t["amount"] for t in ts)} for cat, ts in cat_groups.items()}, }, indent=2, default=str), encoding="utf-8") print(f"\nWrote {OUT_MD}") print(f"\nMatched: {len(matched_pairs)}, Unmatched-bank: {len(unmatched_bank)}") print("\nCategory summary:") for cat, ts in sorted(cat_groups.items()): total = sum(t['amount'] for t in ts) print(f" {cat:35s} {len(ts):3d} txns ${total:>12,.2f}") if __name__ == "__main__": main()