bookkeeping-automation

What this skill does

# Bookkeeping Automation

## Overview

Automate the manual parts of bookkeeping: importing bank statements in multiple formats, categorizing transactions using keyword rules or AI, deduplicating entries, reconciling account balances, and producing expense reports. This skill is format-agnostic — it handles CSV exports from most banks, OFX/QFX files (used by most US banks and Mint), and QIF files (legacy Quicken format).

## Instructions

### Step 1: Parse bank statements

```python
# parser.py — Parse CSV, OFX, and QIF bank statement formats

import csv
import re
from dataclasses import dataclass, field
from datetime import datetime
from decimal import Decimal
from typing import List

@dataclass
class Transaction:
    date: datetime
    description: str
    amount: Decimal          # Negative = debit, Positive = credit
    raw_id: str = ""         # Bank's transaction ID (for dedup)
    category: str = "Uncategorized"
    notes: str = ""

def parse_csv(filepath: str, date_col: str = "Date", desc_col: str = "Description",
              amount_col: str = "Amount", date_fmt: str = "%m/%d/%Y") -> List[Transaction]:
    """
    Parse a bank CSV export. Column names vary by bank — adjust defaults.

    Common variants:
      Chase:    Date, Description, Amount
      Bank of America: Date, Description, Amount, Running Bal.
      Wells Fargo: date, description, deposits, withdrawals, balance
    """
    transactions = []
    with open(filepath, newline="", encoding="utf-8-sig") as f:
        reader = csv.DictReader(f)
        for row in reader:
            # Handle split debit/credit columns (e.g., Wells Fargo)
            if "deposits" in [k.lower() for k in row]:
                credit = Decimal(row.get("deposits", "0").replace(",", "") or "0")
                debit = Decimal(row.get("withdrawals", "0").replace(",", "") or "0")
                amount = credit - debit
            else:
                raw = row[amount_col].replace(",", "").replace("$", "").strip()
                amount = Decimal(raw)

            transactions.append(Transaction(
                date=datetime.strptime(row[date_col].strip(), date_fmt),
                description=row[desc_col].strip(),
                amount=amount,
                raw_id=row.get("Transaction ID", ""),
            ))
    return transactions

def parse_ofx(filepath: str) -> List[Transaction]:
    """Parse OFX/QFX files (Open Financial Exchange — used by most US banks)."""
    transactions = []
    with open(filepath, encoding="utf-8", errors="ignore") as f:
        content = f.read()

    # Extract STMTTRN blocks
    pattern = re.compile(r"<STMTTRN>(.*?)</STMTTRN>", re.DOTALL)
    for match in pattern.finditer(content):
        block = match.group(1)

        def extract(tag):
            m = re.search(rf"<{tag}>(.*?)(?:<|$)", block)
            return m.group(1).strip() if m else ""

        date_str = extract("DTPOSTED")[:8]  # YYYYMMDD
        transactions.append(Transaction(
            date=datetime.strptime(date_str, "%Y%m%d"),
            description=extract("MEMO") or extract("NAME"),
            amount=Decimal(extract("TRNAMT")),
            raw_id=extract("FITID"),
        ))
    return transactions

def parse_qif(filepath: str) -> List[Transaction]:
    """Parse QIF files (Quicken Interchange Format — legacy but still common)."""
    transactions = []
    current = {}

    with open(filepath, encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if line.startswith("D"):   # Date
                current["date"] = line[1:]
            elif line.startswith("T"): # Amount
                current["amount"] = line[1:].replace(",", "")
            elif line.startswith("P"): # Payee
                current["description"] = line[1:]
            elif line.startswith("N"): # Check number / ID
                current["raw_id"] = line[1:]
            elif line == "^":          # Record separator
                if "date" in current and "amount" in current:
                    # QIF dates vary: M/D/Y, M/D'YY, etc.
                    for fmt in ("%m/%d/%Y", "%m/%d'%y", "%d/%m/%Y"):
                        try:
                            parsed_date = datetime.strptime(current["date"], fmt)
                            break
                        except ValueError:
                            continue
                    transactions.append(Transaction(
                        date=parsed_date,
                        description=current.get("description", ""),
                        amount=Decimal(current["amount"]),
                        raw_id=current.get("raw_id", ""),
                    ))
                current = {}

    return transactions
```

### Step 2: Categorize transactions

```python
# categorizer.py — Rule-based and AI-powered transaction categorization

import re
from typing import List
from parser import Transaction

# Rule-based categorization — extend this dict for your business
CATEGORY_RULES = {
    "Software & SaaS": [
        "aws", "amazon web services", "digitalocean", "cloudflare", "github",
        "vercel", "heroku", "stripe", "twilio", "sendgrid", "datadog",
        "notion", "linear", "figma", "zapier", "openai",
    ],
    "Advertising": [
        "google ads", "facebook ads", "meta ads", "twitter ads", "linkedin ads",
        "reddit ads", "bing ads",
    ],
    "Payroll & Contractors": [
        "gusto", "rippling", "deel", "remote.com", "paylocity", "payroll",
        "wise", "transferwise",
    ],
    "Office & Supplies": [
        "staples", "office depot", "amazon", "best buy",
    ],
    "Travel": [
        "airbnb", "marriott", "hilton", "delta", "united", "american airlines",
        "southwest", "uber", "lyft", "expedia",
    ],
    "Meals & Entertainment": [
        "restaurant", "cafe", "coffee", "starbucks", "doordash", "grubhub",
        "ubereats",
    ],
    "Banking & Fees": [
        "bank fee", "service charge", "wire fee", "monthly fee", "overdraft",
    ],
    "Revenue": [
        "stripe payment", "paypal transfer", "square payment",
    ],
}

def categorize_by_rules(transactions: List[Transaction]) -> List[Transaction]:
    """Apply keyword rules to categorize transactions."""
    for tx in transactions:
        desc_lower = tx.description.lower()
        matched = False
        for category, keywords in CATEGORY_RULES.items():
            if any(kw in desc_lower for kw in keywords):
                tx.category = category
                matched = True
                break
        if not matched:
            tx.category = "Uncategorized"
    return transactions

def categorize_with_ai(transactions: List[Transaction], api_key: str,
                       model: str = "gpt-4o-mini") -> List[Transaction]:
    """
    Use an LLM to categorize transactions that rules couldn't match.
    Only sends uncategorized transactions to reduce API costs.
    """
    import json
    import openai

    client = openai.OpenAI(api_key=api_key)
    uncategorized = [tx for tx in transactions if tx.category == "Uncategorized"]

    if not uncategorized:
        return transactions

    # Batch up to 50 transactions per request
    batch_size = 50
    categories = list(CATEGORY_RULES.keys()) + ["Personal", "Tax", "Insurance", "Other"]

    for i in range(0, len(uncategorized), batch_size):
        batch = uncategorized[i:i + batch_size]
        tx_list = [
            {"id": j, "description": tx.description, "amount": str(tx.amount)}
            for j, tx in enumerate(batch)
        ]

        prompt = f"""Categorize these bank transactions into one of these categories:
{json.dumps(categories, indent=2)}

Transactions:
{json.dumps(tx_list, indent=2)}

Return a JSON array of {{"id": <int>, "category": "<category>"}} objects only."""

        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"},
            tempe