#!/usr/bin/env python3 """ find_leaks.py - Money-Back Agent helper (standard library only, runs offline). Reads one or more bank / credit-card exports (CSV, TSV, semicolon CSV, or text copied from a PDF statement) and prints a ranked "money leak" report: recurring charges, price increases, possible duplicates, overlapping services, converted trials, and bank/ATM/FX fees, each with a yearly cost. Usage: python3 find_leaks.py statement1.csv [statement2.csv ...] python3 find_leaks.py --json statement.csv # machine-readable output python3 find_leaks.py --sign pos --dates DMY file.csv cat statement.txt | python3 find_leaks.py - Options: --sign auto|neg|pos How charges appear in the file (default: auto-detect) --dates auto|MDY|DMY Date order for 01/02/2026-style dates (default: auto) --json Print JSON instead of Markdown Same detection rules as tools/money-leak-finder.html. Not financial advice. """ import sys, re, json, csv, io, statistics from datetime import date CADENCES = [ ("Weekly", 6, 8, 52), ("Every 2 weeks", 13, 16, 26), ("Monthly", 26, 35, 12), ("Every 2 months", 56, 66, 6), ("Quarterly", 85, 96, 4), ("Every 6 months", 175, 190, 2), ("Yearly", 350, 380, 1), ] KNOWN = [ (r"NETFLIX", "Netflix", "Streaming"), (r"SPOTIFY", "Spotify", "Music"), (r"APPLE ?MUSIC", "Apple Music", "Music"), (r"TIDAL|DEEZER|PANDORA|SIRIUS ?XM", None, "Music"), (r"DISNEY ?PLUS|DISNEYPLUS|DISNEY\+", "Disney+", "Streaming"), (r"HULU", "Hulu", "Streaming"), (r"HBO ?MAX|\bMAX\.COM|WBD ?MAX", "Max", "Streaming"), (r"PARAMOUNT", "Paramount+", "Streaming"), (r"PEACOCK", "Peacock", "Streaming"), (r"APPLE ?TV", "Apple TV+", "Streaming"), (r"YOUTUBE ?(PREMIUM|TV)|GOOGLE \*YOUTUBE", "YouTube Premium", "Streaming"), (r"DAZN|NOW ?TV|STARZ|SHOWTIME|FUBO|SLING|CRUNCHYROLL|DISCOVERY\+|BRITBOX|MUBI", None, "Streaming"), (r"AMAZON PRIME|AMZN PRIME|PRIME VIDEO|PRIME MEMBERSHIP", "Amazon Prime", "Memberships"), (r"AUDIBLE", "Audible", "Books & audio"), (r"KINDLE ?UNL", "Kindle Unlimited", "Books & audio"), (r"SCRIBD|EVERAND|STORYTEL", None, "Books & audio"), (r"ICLOUD", "iCloud+", "Cloud storage"), (r"GOOGLE ?\*? ?(ONE|STORAGE)", "Google One", "Cloud storage"), (r"DROPBOX", "Dropbox", "Cloud storage"), (r"APPLE\.COM/BILL|APPLE\.COM BILL|ITUNES|APPLE SERVICES", "Apple (App Store)", "Apps"), (r"GOOGLE ?\*? ?PLAY", "Google Play", "Apps"), (r"MICROSOFT ?365|MSFT ?\*|MICROSOFT ?\*|OFFICE ?365", "Microsoft 365", "Software"), (r"ADOBE", "Adobe", "Software"), (r"CANVA", "Canva", "Software"), (r"NOTION", "Notion", "Software"), (r"GRAMMARLY", "Grammarly", "Software"), (r"ZOOM\.US|ZOOM VIDEO|ZOOM\.COM", "Zoom", "Software"), (r"LINKEDIN", "LinkedIn Premium", "Software"), (r"SQUARESPACE|WIX\.COM|GODADDY|SHOPIFY|NAMECHEAP|WEBFLOW|MAILCHIMP|CALENDLY|SLACK|FIGMA|AIRTABLE|1PASSWORD|LASTPASS|NORDVPN|EXPRESSVPN|SURFSHARK", None, "Software"), (r"OPENAI|CHATGPT", "ChatGPT", "AI tools"), (r"ANTHROPIC|CLAUDE\.AI", "Claude", "AI tools"), (r"MIDJOURNEY", "Midjourney", "AI tools"), (r"PERPLEXITY", "Perplexity", "AI tools"), (r"CURSOR|GITHUB|RUNWAY|ELEVENLABS|JASPER|COPILOT", None, "AI tools"), (r"PLANET ?FIT", "Planet Fitness", "Fitness"), (r"LA FITNESS|EQUINOX|ANYTIME FIT|CRUNCH|24 HOUR FIT|PURE ?GYM|GOLD.?S GYM|ORANGETHEORY|CLASSPASS|PELOTON|STRAVA|LIFE TIME FIT|SOULCYCLE|\bGYM\b|FITNESS|PALESTRA|MCFIT|VIRGIN ACTIVE|DAVID LLOYD|THE GYM GROUP", None, "Fitness"), (r"DASHPASS|DOORDASH", "DoorDash", "Food delivery"), (r"UBER ?ONE", "Uber One", "Memberships"), (r"GRUBHUB|INSTACART|DELIVEROO|JUST ?EAT|GLOVO|HELLO ?FRESH|HOME ?CHEF|BLUE APRON|FACTOR", None, "Food delivery"), (r"WALMART ?\+|WALMART PLUS|WMT PLUS", "Walmart+", "Memberships"), (r"COSTCO|SAM.?S CLUB|BJ.?S WHOLESALE", None, "Memberships"), (r"DUOLINGO|BABBEL|CALM\.COM|HEADSPACE|NOOM|MYFITNESSPAL", None, "Apps"), (r"TINDER|BUMBLE|HINGE|MATCH\.COM|OKCUPID", None, "Dating"), (r"PATREON|ONLYFANS|TWITCH", None, "Creators"), (r"SUBSTACK|NYTIMES|NY TIMES|WSJ|WALL STREET J|WASHINGTON POST|ECONOMIST|MEDIUM\.COM|BLOOMBERG|FINANCIAL TIMES", None, "News & media"), (r"XBOX|PLAYSTATION|SONY INTERACTIVE|NINTENDO|GAME ?PASS|EA PLAY|UBISOFT", None, "Gaming"), (r"VERIZON|VZWRLSS|VZW", "Verizon", "Phone"), (r"AT ?& ?T|ATT ?\*", "AT&T", "Phone"), (r"T-MOBILE|TMOBILE", "T-Mobile", "Phone"), (r"COMCAST|XFINITY", "Xfinity (Comcast)", "Internet & TV"), (r"SPECTRUM|CHARTER COMM", "Spectrum", "Internet & TV"), (r"MINT MOBILE|VISIBLE|CRICKET|BOOST MOBILE|METRO BY|\bEE\b|VODAFONE|\bO2\b|THREE\.CO|GIFFGAFF|LEBARA|\bTIM\b|WIND ?TRE|ILIAD|HO\. ?MOBILE|VERY MOBILE|KENA", None, "Phone"), (r"COX COMM|OPTIMUM|FRONTIER|CENTURYLINK|GOOGLE FIBER|DIRECTV|DISH NETWORK|\bBT\b|VIRGIN MEDIA|\bSKY\b|TALKTALK|PLUSNET|FASTWEB|STARLINK|TISCALI", None, "Internet & TV"), (r"GEICO|PROGRESSIVE|STATE FARM|ALLSTATE|LIBERTY MUTUAL|USAA|LEMONADE|NATIONWIDE|FARMERS INS|TRAVELERS|AVIVA|ADMIRAL|DIRECT LINE|UNIPOL|GENERALI|ALLIANZ|\bAXA\b|PRIMA\.IT|INSURANCE|ASSICURAZ", None, "Insurance"), (r"ELECTRIC|ENERGY|POWER CO|\bGAS CO|WATER|PG ?& ?E|CON ?ED|DUKE ENERGY|EDISON|NATIONAL GRID|DOMINION|XCEL|ENEL|\bHERA\b|A2A|IREN|OCTOPUS|BRITISH GAS|\bEDF\b|E\.ON|UTILIT", None, "Utilities"), (r"SIMPLISAFE|\bADT\b|RING\.COM|RING PROTECT|ARLO|NEST AWARE", None, "Home & security"), (r"CHEWY|BARK ?BOX|PET ?INSUR|TRUPANION", None, "Pets"), ] KNOWN = [(re.compile(p), n, c) for p, n, c in KNOWN] SUB_CATS = {"Streaming", "Music", "Memberships", "Books & audio", "Cloud storage", "Apps", "Software", "AI tools", "Fitness", "Food delivery", "Dating", "Creators", "News & media", "Gaming", "Home & security"} BILL_CATS = {"Phone", "Internet & TV", "Insurance", "Utilities"} OVERLAP_CATS = {"Streaming": "streaming services", "Music": "music services", "Cloud storage": "cloud storage plans", "AI tools": "AI subscriptions", "Food delivery": "delivery memberships", "Dating": "dating apps", "Fitness": "fitness memberships", "Books & audio": "book/audio plans", "Memberships": "memberships"} FEE_RE = re.compile(r"\b(FEES?|SERVICE CHARGE|MAINTENANCE|MONTHLY ACCOUNT CHARGE|OVERDRAFT|OD CHARGE|NSF|INSUFFICIENT FUNDS|RETURNED ITEM|NON-?STERLING|FOREIGN TRANS\w*|FOREIGN EXCH\w*|FX (CHARGE|MARKUP)|INTL (TRANS\w*|TXN)|INTERNATIONAL TRANS\w*|CROSS.?BORDER|ATM SURCHARGE|ATM CHARGE|LATE CHARGE|INTEREST CHARGE\w*|PURCHASE INTEREST|CASH ADVANCE INTEREST|COMMISSION[EI]|COMMISSIONS?|CANONE( MENSILE)?( CONTO)?|SPESE|IMPOSTA DI BOLLO|GEB(U|Ü)HR\w*|ENTGELT|KONTOF(U|Ü)HRUNG\w*|FRAIS|COMISI(O|Ó)N)\b") EXCLUDE_RE = re.compile(r"\b(TRANSFER|XFER|ZELLE|VENMO|CASH ?APP|PAYMENT THANK|THANK YOU|AUTOPAY|AUTO ?PAY(MENT)?|CREDIT CARD PAYMENT|CARD PAYMENT|ONLINE PAYMENT|PAYMENT RECEIVED|PAYROLL|SALARY|DIRECT DEP\w*|DEPOSIT|INTEREST PAID|INTEREST EARNED|REFUND|REVERSAL|RENT|MORTGAGE|LOAN|ATM WITHDRAWAL|ATM CASH|CASH WITHDRAWAL|WITHDRAWAL|STIPENDIO|BONIFICO|GIROCONTO|PRELIEVO|IRS|HMRC|SAVINGS|INVESTMENT|BROKERAGE|ROBINHOOD|VANGUARD|FIDELITY|SCHWAB)\b") MONTHS = {"jan": 1, "feb": 2, "mar": 3, "apr": 4, "may": 5, "jun": 6, "jul": 7, "aug": 8, "sep": 9, "oct": 10, "nov": 11, "dec": 12, "gen": 1, "mag": 5, "giu": 6, "lug": 7, "ago": 8, "set": 9, "ott": 10, "dic": 12, "mär": 3, "mrz": 3, "mai": 5, "okt": 10, "dez": 12, "ene": 1, "abr": 4, "fév": 2, "fev": 2, "avr": 4, "aoû": 8, "aou": 8, "déc": 12} H_DATE = re.compile(r"(^|\b)(date|data|datum|fecha|posted|booking|buchungstag|valuta|operazione)(\b|$)", re.I) H_DESC = [re.compile(p, re.I) for p in (r"description|descrizione|beschreibung|libell|concepto|narrative", r"merchant|payee|beneficiar|empf", r"details|memo|causale|verwendungszweck|reference|particulars", r"name|transaction$|^transaction")] H_AMT = re.compile(r"amount|importo|betrag|montant|importe|^value$|^sum$|^total$", re.I) H_DEBIT = re.compile(r"debit|withdrawal|paid out|money out|^out$|uscite|addebit|^dare$|^soll$|spent|charges?$", re.I) H_CREDIT = re.compile(r"credit|deposit|paid in|money in|^in$|entrate|accredit|^avere$|^haben$|received", re.I) H_TYPE = re.compile(r"^(type|transaction type|credit/debit|debit/credit|cr/dr|dr/cr|dc)$", re.I) H_SKIP = re.compile(r"balance|saldo|solde|kontostand|category|categoria|card|account|status|id$|reference no|check|currency", re.I) def parse_amount(raw, dec_comma=False): if raw is None: return None s = str(raw).strip() if not s or re.search(r"[A-Za-z]{4,}", s) or re.search(r"\d/\d", s) or re.search(r"\d-\d", s): return None neg = False if s.startswith("(") and s.endswith(")"): neg, s = True, s[1:-1] if re.search(r"\bCR\.?$", s, re.I): s = re.sub(r"\s*CR\.?$", "", s, flags=re.I) elif re.search(r"\b(DR|DB)\.?$", s, re.I): neg, s = True, re.sub(r"\s*(DR|DB)\.?$", "", s, flags=re.I) s = re.sub(r"[A-Za-z$€£\s  ']", "", s) if s.endswith("-"): neg, s = not neg, s[:-1] s = s.lstrip("+") if s.startswith("-"): neg, s = not neg, s[1:] if not re.fullmatch(r"[\d.,]+", s or "x") or not re.search(r"\d", s): return None lc, ld = s.rfind(","), s.rfind(".") try: if lc > -1 and ld > -1: n = float(s.replace(".", "").replace(",", ".")) if lc > ld else float(s.replace(",", "")) elif lc > -1: after = len(s) - lc - 1 n = float(s.replace(",", "")) if (s.count(",") > 1 or (after == 3 and not dec_comma)) else float(s.replace(",", ".")) elif ld > -1: after = len(s) - ld - 1 n = float(s.replace(".", "")) if (s.count(".") > 1 or (after == 3 and dec_comma)) else float(s) else: n = float(s) except ValueError: return None return -n if neg else n def _clean_date(raw): s = str(raw).strip().strip('"') return re.sub(r"[ T]\d{1,2}:\d{2}(:\d{2})?(\.\d+)?\s*(Z|[+-]\d{2}:?\d{2})?\s*(AM|PM)?$", "", s, flags=re.I).strip() def _mk(y, m, d): try: if not m or not (1990 <= y <= 2100): return None return date(y, m, d).toordinal() except ValueError: return None def _month(w): w = w.lower().replace(".", "") return MONTHS.get(w) or MONTHS.get(w[:4]) or MONTHS.get(w[:3]) def _yr(s): return 2000 + int(s) if len(s) == 2 else int(s) def parse_date(raw, order="MDY"): if raw is None: return None s = _clean_date(raw) m = re.fullmatch(r"(\d{4})[-/.](\d{1,2})[-/.](\d{1,2})", s) if m: return _mk(int(m[1]), int(m[2]), int(m[3])) m = re.fullmatch(r"(20\d{2})(\d{2})(\d{2})", s) if m: return _mk(int(m[1]), int(m[2]), int(m[3])) m = re.fullmatch(r"(\d{1,2})[-/.](\d{1,2})[-/.](\d{2}|\d{4})", s) if m: a, b, y = int(m[1]), int(m[2]), _yr(m[3]) return _mk(y, b, a) if order == "DMY" else _mk(y, a, b) m = re.fullmatch(r"(\d{1,2})[\s\-/.]+([A-Za-zÀ-ÿ]{3,9})\.?[\s\-/.,]+(\d{2}|\d{4})", s) if m: return _mk(_yr(m[3]), _month(m[2]), int(m[1])) m = re.fullmatch(r"([A-Za-zÀ-ÿ]{3,9})\.?[\s\-/.]+(\d{1,2})(?:st|nd|rd|th)?,?[\s\-/.]+(\d{2}|\d{4})", s, re.I) if m: return _mk(_yr(m[3]), _month(m[1]), int(m[2])) return None def looks_date(v): return parse_date(v, "MDY") is not None or parse_date(v, "DMY") is not None def detect_order(strs, fallback): dmy = mdy = 0 for raw in strs: m = re.fullmatch(r"(\d{1,2})[-/.](\d{1,2})[-/.](\d{2}|\d{4})", _clean_date(raw)) if m: dmy += int(m[1]) > 12 mdy += int(m[2]) > 12 if dmy != mdy: return "DMY" if dmy > mdy else "MDY" if any(re.match(r"\d{1,2}\.\d{1,2}\.\d{2,4}", str(r).strip()) for r in strs): return "DMY" return fallback def split_rows(lines, delim): return [[c.strip() for c in row] for row in csv.reader(lines, delimiter=delim)] def detect_delim(lines): best, best_score = None, 0 for d in [",", ";", "\t", "|"]: try: counts = [len(r) for r in split_rows(lines[:40], d)] except csv.Error: continue med = statistics.median(counts) if counts else 0 if med < 2: continue score = sum(1 for c in counts if c == med) / len(counts) * 10 + min(med, 8) if score > best_score: best, best_score = d, score return best def map_header(h): mp = dict(date=-1, desc=-1, amt=-1, debit=-1, credit=-1, type=-1) used = set() for key, test in (("date", lambda c: H_DATE.search(c) and not H_AMT.search(c)), ("amt", lambda c: H_AMT.search(c) and not H_SKIP.search(c)), ("type", lambda c: H_TYPE.search(c.strip())), ("debit", lambda c: H_DEBIT.search(c) and not H_SKIP.search(c)), ("credit", lambda c: H_CREDIT.search(c) and not H_SKIP.search(c))): for i, c in enumerate(h): if i not in used and mp[key] < 0 and test(c): mp[key] = i used.add(i) for rx in H_DESC: if mp["desc"] >= 0: break for i, c in enumerate(h): if i not in used and rx.search(c) and not H_SKIP.search(c): mp["desc"] = i break return mp def infer_columns(rows, dec_comma): n = max((len(r) for r in rows[:50]), default=0) stats = [] for c in range(n): d = a = neg = tot = ln = 0 for r in rows[:200]: v = r[c] if c < len(r) else "" if not v: continue tot += 1 ln += len(v) if looks_date(v): d += 1 else: x = parse_amount(v, dec_comma) if x is not None: a += 1 neg += x < 0 stats.append(dict(c=c, d=d / tot if tot else 0, a=a / tot if tot else 0, neg=neg, len=ln / tot if tot else 0)) mp = dict(date=-1, desc=-1, amt=-1, debit=-1, credit=-1, type=-1) ds = sorted([s for s in stats if s["d"] > 0.6], key=lambda s: -s["d"]) if ds: mp["date"] = ds[0]["c"] as_ = [s for s in stats if s["a"] > 0.6 and s["c"] != mp["date"]] with_neg = [s for s in as_ if s["neg"]] if with_neg: mp["amt"] = with_neg[0]["c"] elif as_: mp["amt"] = as_[0]["c"] txt = sorted([s for s in stats if s["c"] not in (mp["date"], mp["amt"]) and s["a"] < 0.5 and s["d"] < 0.5], key=lambda s: -s["len"]) if txt: mp["desc"] = txt[0]["c"] return mp LOOSE_DATE = re.compile(r"^(\d{4}[-/.]\d{1,2}[-/.]\d{1,2}|\d{1,2}[-/.]\d{1,2}[-/.]\d{2,4}|\d{1,2}[\s\-][A-Za-zÀ-ÿ]{3,9}\.?[\s\-,]+\d{2,4}|[A-Za-z]{3,9}\.?\s\d{1,2},?\s\d{2,4})\s+(.*)$") LOOSE_AMT = re.compile(r"\s+(\(?[-+]?\s?[$€£]?\s?\d[\d.,]*\)?-?(?:\s?(?:CR|DR))?)$", re.I) def parse_statement(text, sign="auto", date_order="auto"): lines = [l for l in text.replace("\r\n", "\n").replace("\r", "\n").split("\n") if l.strip()] res = dict(txs=[], skipped_rows=0, credits=0, currency=None, date_order=None, spend_negative=None) if not lines: return res head = "\n".join(lines[:60]) cs = {"USD": head.count("$"), "GBP": head.count("£"), "EUR": head.count("€") + len(re.findall(r"EUR\b", head))} best = max(cs, key=cs.get) res["currency"] = best if cs[best] else None delim = detect_delim(lines) dec_comma = delim == ";" or res["currency"] == "EUR" or (delim == "\t" and re.search(r"\d,\d{2}\b", head) is not None) if not res["currency"] and delim == ";": res["currency"] = "EUR" raw, fixed_neg = [], None if delim: rows = split_rows(lines, delim) hi = -1 for i, r in enumerate(rows[:20]): score = sum(1 for c in r if H_DATE.search(c) or H_AMT.search(c) or H_DEBIT.search(c) or H_CREDIT.search(c) or any(x.search(c) for x in H_DESC)) if score >= 2 and not any(looks_date(c) for c in r): hi = i break if hi >= 0: mp, rows = map_header(rows[hi]), rows[hi + 1:] else: mp = infer_columns(rows, dec_comma) if mp["date"] < 0 or (mp["amt"] < 0 and mp["debit"] < 0) or mp["desc"] < 0: inf = infer_columns(rows, dec_comma) for k in ("date", "amt", "desc"): if mp[k] < 0 and not (k == "amt" and mp["debit"] >= 0): mp[k] = inf[k] if mp["date"] >= 0 and (mp["amt"] >= 0 or mp["debit"] >= 0): fixed_neg = mp["amt"] < 0 g = lambda r, i: r[i] if 0 <= i < len(r) else "" for r in rows: dv, desc, amt = g(r, mp["date"]), g(r, mp["desc"]), None if mp["amt"] >= 0: amt = parse_amount(g(r, mp["amt"]), dec_comma) if amt is not None and mp["type"] >= 0: t = g(r, mp["type"]) if re.search(r"debit|^dr$|withdraw|sale|purchase|charge|addebito", t, re.I): amt = -abs(amt) elif re.search(r"credit|^cr$|deposit|payment|refund|accredito", t, re.I): amt = abs(amt) fixed_neg = True else: de = parse_amount(g(r, mp["debit"]), dec_comma) cr = parse_amount(g(r, mp["credit"]), dec_comma) if mp["credit"] >= 0 else None if de: amt = -abs(de) elif cr: amt = abs(cr) if not dv or amt is None: res["skipped_rows"] += 1 continue raw.append((dv, desc, amt)) if len(raw) < max(2, len(lines) * 0.3): loose = [] for l in lines: m = LOOSE_DATE.match(l.strip()) if not m: continue rest, amts = m[2], [] for _ in range(3): mm = LOOSE_AMT.search(rest) if not mm: break v = parse_amount(mm[1], dec_comma) if v is None: break amts.insert(0, v) rest = rest[:mm.start()] if amts and rest.strip(): loose.append((m[1], rest.strip(), amts[0])) if len(loose) > len(raw): raw, fixed_neg = loose, None order = date_order if date_order in ("MDY", "DMY") else detect_order([r[0] for r in raw], "DMY" if (dec_comma or res["currency"] in ("GBP", "EUR")) else "MDY") res["date_order"] = order if sign == "neg": spend_neg = True elif sign == "pos": spend_neg = False elif fixed_neg: spend_neg = True else: spend_neg = (sum(1 for r in raw if r[2] < 0) / len(raw) >= 0.3) if raw else True res["spend_negative"] = spend_neg for dv, desc, amt in raw: day = parse_date(dv, order) if day is None: res["skipped_rows"] += 1 continue spend = -amt if spend_neg else amt if spend <= 0: res["credits"] += 1 continue res["txs"].append(dict(day=day, desc=re.sub(r"\s+", " ", str(desc)).strip(), amt=round(spend, 2))) return res STATES = r"AL|AK|AZ|AR|CA|CO|CT|DE|DC|FL|GA|HI|ID|IL|IN|IA|KS|KY|LA|ME|MD|MA|MI|MN|MS|MO|MT|NE|NV|NH|NJ|NM|NY|NC|ND|OH|OK|OR|PA|RI|SC|SD|TN|TX|UT|VT|VA|WA|WV|WI|WY|GB|UK|US|USA|IT|ITA" def clean_merchant(desc): s = " " + str(desc).upper() + " " s = re.sub(r"PURCHASE AUTHORI[SZ]ED ON \d{1,2}/\d{1,2}", " ", s) s = re.sub(r"\b(POS|ACH|DEBIT CARD|CREDIT CARD|CHECK ?CARD|VISA|MASTERCARD|MC|PURCHASE|RECURRING|PRE-?AUTHORI[SZ]ED|ONLINE|WEB|POINT OF SALE|ELECTRONIC|BILL ?PAY|PMT|PAYMENT TO|DIRECT DEBIT|D/D|DD|CONTACTLESS|CARD|PAGAMENTO|ADDEBITO|SDD|OPERAZIONE|CARTA|LASTSCHRIFT|KARTENZAHLUNG|PRELEVEMENT|SEPA)\b", " ", s) s = re.sub(r"\bPAYPAL ?\*", " ", s) s = re.sub(r"\b(SQ|TST|SP|PY|DD|IN|FS|GOOGLE|APL|AMZ|PP)\s?\*\s?", lambda m: "GOOGLE " if m[1] == "GOOGLE" else " ", s) s = re.sub(r"\bAMZN\.COM/BILL\b", " ", s) s = re.sub(r"\bAMZN ?MKTP\b.*|\bAMAZON\.COM\b.*|\bAMAZON MKTPL?\b.*", " AMAZON MARKETPLACE ", s) s = re.sub(r"\*.*$", " ", s) s = re.sub(r"\b(X{2,}|\*{2,}|#)\d+", " ", s) s = re.sub(r"\+?\d[\d\-. ()]{8,}\d", " ", s) s = re.sub(r"\d{1,2}[/.-]\d{1,2}([/.-]\d{2,4})?", " ", s) s = re.sub(r"WWW\.|\.COM\b|\.CO\.UK\b|\.IT\b|\.NET\b|\.IO\b|HTTPS?://", " ", s) s = re.sub(r"\b[A-Z]*\d[A-Z\d]*\b", " ", s) s = re.sub(r"\s+", " ", re.sub(r"[^A-Z&' +]", " ", s)).strip() s = re.sub(r"\b(INC|LLC|LTD|LIMITED|SPA|S P A|SRL|GMBH|CORP|PLC|FATTURA|SUBSCR|SUBSCRIPTION|SUBS|MEMBERSHIP|BILLING)\b", " ", s) s = re.sub(r"\s+", " ", s).strip() s = re.sub(r"\s(" + STATES + r")$", "", s) return " ".join(s.split()[:3]) def title(s): return re.sub(r"(^|[\s&'-])([a-z])", lambda m: m[1] + m[2].upper(), s.lower()).replace("'S ", "'s ") def classify(desc): u = str(desc).upper() for rx, name, cat in KNOWN: if rx.search(u): nm = name or title(clean_merchant(desc) or u[:24]) return nm.upper(), nm, cat, True c = clean_merchant(desc) or re.sub(r"\s+", " ", u).strip()[:24] return c, title(c), "Other", False def fee_type(u): if "ATM" in u: return "ATM fees" if re.search(r"FOREIGN|NON-?STERLING|FX|INTL|INTERNATIONAL|CROSS.?BORDER|EXCH", u): return "Foreign transaction fees" if re.search(r"OVERDRAFT|OD CHARGE|NSF|INSUFFICIENT|RETURNED ITEM", u): return "Overdraft / NSF fees" if "INTEREST" in u: return "Interest charges" if "LATE" in u: return "Late fees" if re.search(r"MAINTENANCE|MONTHLY|SERVICE CHARGE|ACCOUNT CHARGE|CANONE|KONTOF|SPESE|BOLLO", u): return "Account maintenance fees" return "Other bank fees" def cadence_of(gm): for c in CADENCES: if c[1] <= gm <= c[2]: return c return None def consistent(txs, cat): amts = [t["amt"] for t in txs] ma = statistics.median(amts) if not ma: return False if sum(1 for a in amts if abs(a - ma) / ma <= 0.05) / len(amts) >= 0.8: return True levels = [] for a in amts: if not levels or abs(levels[-1] - a) > 0.004: levels.append(a) if len(levels) <= 2: return True if cat in BILL_CATS: return sum(1 for a in amts if abs(a - ma) / ma <= 0.35) / len(amts) >= 0.75 return False def series(txs, cat): if len(txs) < 2: return None gaps = [txs[i]["day"] - txs[i - 1]["day"] for i in range(1, len(txs))] cad = cadence_of(statistics.median(gaps)) if not cad: return None if sum(1 for g in gaps if cad[1] - 1 <= g <= cad[2] + 1) / len(gaps) < 0.6: return None return cad if consistent(txs, cat) else None def fmt_day(d): return date.fromordinal(d).strftime("%b %d, %Y") def action_for(it): f = {x["t"] for x in it["flags"]} if it["kind"] == "fee": if it["name"].startswith("Foreign"): return "Ask for a refund, then switch to a no-FX-fee card", "09" if it["name"].startswith("Interest"): return "Ask for a one-time interest refund", "09" return "Ask for a fee waiver", "09" if not it["active"]: return "Check it is really cancelled", "01" if "dup" in f: return "Dispute the duplicate charge", "06" if "new" in f: return "Trial converted? Cancel + ask for a refund", "08" if "up" in f and it["cat"] in BILL_CATS: return ("Re-quote: the price went up", "11") if it["cat"] == "Insurance" else ("Negotiate the increase", "04") if "overlap" in f: return "Keep one, cancel or rotate the rest", "01" if "up" in f: return "Push back or downgrade", "18" if it["cat"] in ("Phone", "Internet & TV"): return "Negotiate a lower rate", "04" if it["cat"] == "Insurance": return "Re-quote every 12 months", "11" if it["cat"] == "Utilities": return "Check for a cheaper plan or tariff", "05" if it["cat"] == "Fitness": return "Using it? If not, cancel or freeze", "10" return "Keep, downgrade or cancel", "01" def analyze(txs): out = dict(items=[], dups=[], skipped=0) if not txs: out.update(recurring_yearly=0, fees_yearly=0, dup_total=0, span_days=0) return out start, end = min(t["day"] for t in txs), max(t["day"] for t in txs) span = end - start + 1 out["span_days"] = span groups, fees = {}, {} for t in txs: u = t["desc"].upper() if FEE_RE.search(u) and not re.search(r"REFUND|REVERSAL|WAIVED|REBATE", u): fees.setdefault(fee_type(u), []).append(t) continue if EXCLUDE_RE.search(u): out["skipped"] += 1 continue key, name, cat, known = classify(t["desc"]) groups.setdefault(key, dict(name=name, cat=cat, known=known, txs=[]))["txs"].append(t) def push(g, lst, cad, label=None): first, last = lst[0], lst[-1] it = dict(name=label or g["name"], cat=g["cat"], cadence=cad[0], per_year=cad[3], last_amt=last["amt"], first_amt=first["amt"], count=len(lst), first_seen=fmt_day(first["day"]), last_seen=fmt_day(last["day"]), yearly=round(last["amt"] * cad[3], 2), flags=[], active=True, kind="sub") diff = last["amt"] - first["amt"] if diff >= 0.5 and last["amt"] > first["amt"] * 1.02: pct = round(diff / first["amt"] * 100) it["flags"].append(dict(t="up", text=f"{'Intro price ended? ' if pct >= 50 else 'Price up '}{pct}% ({first['amt']:.2f} -> {last['amt']:.2f})", extra=round(diff * cad[3], 2))) if span >= 60 and first["day"] - start > cad[2] + 3 and (g["cat"] in SUB_CATS or g["cat"] == "Other"): it["flags"].append(dict(t="new", text=f"New since {fmt_day(first['day'])}: trial that converted?")) if end - last["day"] > cad[2] * 1.5 + 3: it["active"] = False it["flags"].append(dict(t="stopped", text=f"No charge since {fmt_day(last['day'])}: already cancelled?")) out["items"].append(it) return it for g in groups.values(): g["txs"].sort(key=lambda t: t["day"]) clean, dups = [], [] for t in g["txs"]: prev = next((c for c in reversed(clean) if t["day"] - c["day"] <= 3 and abs(c["amt"] - t["amt"]) < 0.005), None) if prev: dups.append(dict(name=g["name"], amt=t["amt"], date=fmt_day(t["day"]), first_date=fmt_day(prev["day"]))) else: clean.append(t) made = [] cad = series(clean, g["cat"]) if cad: made.append(push(g, clean, cad)) elif len(clean) >= 2: clusters = [] for t in sorted(clean, key=lambda t: t["amt"]): if clusters and abs(t["amt"] - clusters[-1][0]["amt"]) <= max(0.01, clusters[-1][0]["amt"] * 0.03): clusters[-1].append(t) else: clusters.append([t]) for c in clusters: if len(c) < 2: continue c.sort(key=lambda t: t["day"]) cc = series(c, g["cat"]) if cc: made.append(push(g, c, cc, f"{g['name']} - {c[-1]['amt']:.2f}")) if not made and len(clean) == 1 and g["known"] and (g["cat"] in SUB_CATS or g["cat"] in BILL_CATS): t = clean[0] big = t["amt"] >= 40 it = push(g, [t], CADENCES[6] if big else CADENCES[2]) it["cadence"] = ("Yearly" if big else "Monthly") + "?" it["flags"] = [f for f in it["flags"] if f["t"] != "stopped"] it["active"] = True if not big and end - t["day"] > 45: it["active"] = False it["flags"].append(dict(t="stopped", text=f"Seen once ({fmt_day(t['day'])}): one-off or cancelled?")) else: it["flags"].append(dict(t="est", text="Seen once: cadence guessed")) if dups and (made or g["known"] or any(d["amt"] >= 25 for d in dups)): out["dups"].extend(dups) for it in made: it["flags"].insert(0, dict(t="dup", text=f"{len(dups)} possible duplicate{'s' if len(dups) > 1 else ''}")) sp = max(30, span) for ft, lst in fees.items(): total = sum(t["amt"] for t in lst) out["items"].append(dict(name=ft, cat="Bank fees", cadence=f"{len(lst)}x in {max(1, round(span / 30.4))} mo", per_year=365 / sp, last_amt=round(total / len(lst), 2), first_amt=0, count=len(lst), yearly=round(total * 365 / sp, 2), active=True, kind="fee", flags=[dict(t="fee", text=f"{len(lst)} charge(s), {total:.2f} total")])) by_cat = {} for it in out["items"]: if it["active"] and it["kind"] == "sub" and it["cat"] in OVERLAP_CATS: by_cat.setdefault(it["cat"], []).append(it) for cat, lst in by_cat.items(): if len(lst) >= 2: for it in lst: it["flags"].append(dict(t="overlap", text=f"1 of {len(lst)} {OVERLAP_CATS[cat]}")) for it in out["items"]: it["action"], it["template"] = action_for(it) out["items"].sort(key=lambda i: (not i["active"], -i["yearly"])) out["recurring_yearly"] = round(sum(i["yearly"] for i in out["items"] if i["active"] and i["kind"] == "sub"), 2) out["fees_yearly"] = round(sum(i["yearly"] for i in out["items"] if i["kind"] == "fee"), 2) out["dup_total"] = round(sum(d["amt"] for d in out["dups"]), 2) return out def money(n, cur): sym = {"USD": "$", "GBP": "£", "EUR": "€"}.get(cur or "USD", "$") return f"{sym}{n:,.2f}" def report_md(p, r): cur = p["currency"] or "USD" L = [] L.append(f"Read {len(p['txs'])} charges over {r.get('span_days', 0)} days " f"(dates read as {p['date_order']}, charges as {'negative' if p['spend_negative'] else 'positive'} numbers; " f"{p['credits']} credits/deposits ignored, {p['skipped_rows']} rows skipped, {r['skipped']} transfers/rent/payroll ignored).\n") L.append(f"**Recurring charges: {money(r['recurring_yearly'], cur)}/year** (about {money(r['recurring_yearly'] / 12, cur)}/month)") L.append(f"**Bank fees & interest: {money(r['fees_yearly'], cur)}/year** (annualized)") L.append(f"**Possible duplicates: {money(r['dup_total'], cur)}** (one-time, to claim back)\n") L.append("| # | Merchant | Category | How often | Charge | Per year | What we found | Next move | Template |") L.append("|---|---|---|---|---|---|---|---|---|") for i, it in enumerate(r["items"], 1): found = "; ".join(f["text"] for f in it["flags"]) or "Looks normal" name = it["name"] if it["active"] else f"~~{it['name']}~~ (inactive)" L.append(f"| {i} | {name} | {it['cat']} | {it['cadence']} | {money(it['last_amt'], cur)} | {money(it['yearly'], cur)} | {found} | {it['action']} | {it['template']} |") if r["dups"]: L.append("\n**Possible duplicate charges**") for d in r["dups"]: L.append(f"- {d['name']}: {money(d['amt'], cur)} on {d['first_date']} and again on {d['date']}") if r.get("span_days", 0) and r["span_days"] < 60: L.append("\nNote: less than 60 days of data. Monthly charges are hard to confirm; ask for 3+ months.") return "\n".join(L) def main(argv): sign, order, as_json, files = "auto", "auto", False, [] i = 0 while i < len(argv): a = argv[i] if a == "--sign": sign = argv[i + 1]; i += 2; continue if a == "--dates": order = argv[i + 1]; i += 2; continue if a == "--json": as_json = True; i += 1; continue files.append(a); i += 1 if not files: print(__doc__) return 1 txs, meta = [], None for f in files: text = sys.stdin.read() if f == "-" else open(f, encoding="utf-8-sig", errors="replace").read() p = parse_statement(text, sign, order) txs.extend(p["txs"]) if meta is None: meta = p else: for k in ("credits", "skipped_rows"): meta[k] += p[k] meta["txs"] = txs r = analyze(txs) if as_json: print(json.dumps(dict(meta={k: v for k, v in meta.items() if k != "txs"}, transactions=len(txs), **r), indent=2)) else: print(report_md(meta, r)) return 0 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))