From 0a1c5b2838b7686a6de8671a31c9a2b76576e682 Mon Sep 17 00:00:00 2001 From: Ryan Kegel Date: Thu, 8 Oct 2026 16:48:03 -0400 Subject: [PATCH] add transaction CSV export and category suggestion features --- .gitignore | 2 + AGENTS.md | 25 ++++ README.md | 18 +++ category_suggester.py | 263 +++++++++++++++++++++++++++++++++++++++++ main.py | 268 +++++++++++++++++++++++++++++++++++++++--- test_import.py | 219 +++++++++++++++++++++++++++++++++- ynab_client.py | 33 ++++++ 7 files changed, 810 insertions(+), 18 deletions(-) create mode 100644 AGENTS.md create mode 100644 category_suggester.py diff --git a/.gitignore b/.gitignore index 5f424d8..90cdcc8 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,5 @@ *.json /ynab.yml /.venv/ +/__pycache__/ +transactions.csv diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..51f3b50 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,25 @@ +# Repository Guidelines + +## Project Structure & Modules + +This repository is a small, flat Python application. Core scripts live in the root: `main.py` is the extraction and import CLI, `wf_*_extractor.py` handles Wells Fargo payloads, `transaction_processor.py` transforms transactions, `ynab_client.py` connects to YNAB, `category_suggester.py` learns category patterns, and `reconcile.py` compares transactions. `test_import.py` contains focused import-helper checks. Usage and module documentation is in `README.md`, `QUICKSTART.md`, `PROJECT_SUMMARY.md`, and `YNAB_MODULE.md`. There are no separate source, test, or asset directories. + +## Build, Test, and Development + +There is no build step. Use Python 3 and install pinned dependencies with `python -m pip install -r requirements.txt`. Run the CLI with `python main.py --input posted.dat`; add `--help` to inspect options. Run the assertion checks with `python test_import.py` (or `python -m pytest test_import.py` if pytest is installed). `python reconcile.py` runs the Wells Fargo/YNAB reconciliation flow and requires valid local YNAB configuration. + +## Coding Style & Naming + +Follow the existing Python style: four-space indentation, `snake_case` for functions and variables, and `PascalCase` for classes. Keep modules focused on their current responsibilities and preserve the existing CLI and programmatic interfaces unless a change requires updating their documentation and examples. No formatter or linter is configured. + +## Testing Guidelines + +Add focused assertions to `test_import.py` for import conversion, classification, and formatting behavior. Name test functions `test_` and include representative transaction cases, especially amount signs, duplicate IDs, and posted versus pending status. Run the script directly before submitting changes that affect these helpers. + +## Commits and Pull Requests + +Recent history uses brief, lowercase summaries such as `added batch file.` and `made fix.`. Keep commit subjects concise and describe the change. Pull requests should explain the user-visible behavior, list relevant manual or automated checks, and note changes to CLI usage or configuration. Include sample output when it clarifies an extraction or reconciliation change. + +## Security and Local Data + +Keep real bank exports, `transactions.csv`, and generated JSON out of commits; `.gitignore` excludes `*.dat`, `*.json` (including `category_memory.json`), and `transactions.csv`. Treat `ynab.yml` as secret-bearing local configuration and do not commit API tokens or account details. diff --git a/README.md b/README.md index 5a1b444..269ea84 100644 --- a/README.md +++ b/README.md @@ -126,6 +126,24 @@ print(f"Balance: ${balance / 1000:.2f}") See `ynab_example.py` for a complete example that also saves all transactions to `ynab_transactions.json`. +### Export YNAB Transactions to CSV + +Write transactions from the configured credit and debit accounts for the past 60 days: + +```powershell +python .\main.py --record-transactions +``` + +Set a different lookback (1–240 days) and output path: + +```powershell +python .\main.py --record-transactions=90 --transactions-csv .\exports\transactions.csv +``` + +The default output is `transactions.csv`. It contains only `payee`, `category`, `inflow`, and `outflow` columns. Split transactions are written as one row per subcategory, keeping the parent transaction's payee on each row. The misspelled `--record-trasactions` option is also accepted as an alias. + +When importing bank transactions with `--ynab-import`, the importer uses `transactions.csv` and a reusable `category_memory.json` sidecar to suggest categories from similar payees, transaction direction, and amount. Confident suggestions are sent to YNAB; uncertain ones appear with a `?` in the preview and remain uncategorized. Each explicit `--record-transactions` export merges new patterns into the sidecar, so they remain available after older rows leave the CSV. Keep or sync both files between machines; the sidecar contains normalized payee/category/amount patterns but no transaction dates or IDs. + ### ynab.yml Format ```yaml diff --git a/category_suggester.py b/category_suggester.py new file mode 100644 index 0000000..d8c9734 --- /dev/null +++ b/category_suggester.py @@ -0,0 +1,263 @@ +"""Learn conservative YNAB category suggestions from transaction CSV history.""" + +import csv +from dataclasses import dataclass +from decimal import Decimal, InvalidOperation, ROUND_HALF_UP +from difflib import SequenceMatcher +import json +import math +from pathlib import Path +import re +import unicodedata + + +_MEMORY_VERSION = 1 +_MIN_PAYEE_SIMILARITY = 0.70 +_MIN_CATEGORY_SHARE = 0.80 +_AMOUNT_SMOOTHING_CENTS = 100 + + +@dataclass(frozen=True, order=True) +class CategoryPattern: + """An aggregated payee/category/amount pattern, without transaction dates.""" + + payee: str + category: str + direction: str + amount_cents: int + count: int = 1 + + +@dataclass(frozen=True) +class CategorySuggestion: + """A category candidate and whether evidence is strong enough to apply it.""" + + category_name: str + confidence: float + category_id: str | None + applied: bool + + +def normalize_payee(payee: str) -> str: + """Normalize formatting and discard long numeric or mixed reference codes.""" + normalized = unicodedata.normalize("NFKD", payee).encode("ascii", "ignore").decode() + tokens = re.findall(r"[a-z0-9]+", normalized.casefold()) + stable_tokens = [ + token + for token in tokens + if not ( + (token.isdigit() and len(token) >= 3) + or ( + len(token) >= 5 + and re.search(r"[a-z]", token) + and re.search(r"[0-9]", token) + ) + ) + ] + return " ".join(stable_tokens) + + +def _amount_cents(value: str) -> int | None: + cleaned = value.strip().replace("$", "").replace(",", "") + if not cleaned: + return None + try: + amount = Decimal(cleaned) + except InvalidOperation: + return None + if not amount.is_finite(): + return None + cents = (amount * 100).quantize(Decimal("1"), rounding=ROUND_HALF_UP) + return int(cents) + + +class CategoryHistory: + """Memory of unique merchant/category/amount patterns and snapshot support counts.""" + + def __init__( + self, + patterns: dict[tuple[str, str, str, int], CategoryPattern] | None = None, + ) -> None: + self.patterns = patterns or {} + + @staticmethod + def _key(pattern: CategoryPattern) -> tuple[str, str, str, int]: + return ( + pattern.payee, + pattern.category.casefold(), + pattern.direction, + pattern.amount_cents, + ) + + @staticmethod + def memory_path_for_csv(csv_path: str | Path) -> Path: + """Return the portable category-memory sidecar path beside the CSV.""" + return Path(csv_path).with_name("category_memory.json") + + @classmethod + def load(cls, path: str | Path) -> "CategoryHistory": + """Load and validate a category-memory sidecar.""" + memory_path = Path(path) + payload = json.loads(memory_path.read_text(encoding="utf-8")) + if not isinstance(payload, dict) or payload.get("version") != _MEMORY_VERSION: + raise ValueError(f"Unsupported category memory version in {memory_path}") + + patterns: dict[tuple[str, str, str, int], CategoryPattern] = {} + for item in payload.get("patterns", []): + payee = str(item["payee"]).strip() + category = str(item["category"]).strip() + direction = str(item["direction"]).strip().lower() + amount_cents = int(item["amount_cents"]) + count = max(1, int(item.get("count", 1))) + if payee and category and direction in ("inflow", "outflow") and amount_cents > 0: + pattern = CategoryPattern(payee, category, direction, amount_cents, count) + patterns[cls._key(pattern)] = pattern + return cls(patterns) + + @classmethod + def from_csv(cls, csv_path: str | Path) -> "CategoryHistory": + """Read the repository's four-column transaction CSV as category examples.""" + path = Path(csv_path) + patterns: dict[tuple[str, str, str, int], CategoryPattern] = {} + with path.open("r", encoding="utf-8-sig", newline="") as csv_file: + reader = csv.DictReader(csv_file) + required = {"payee", "category", "inflow", "outflow"} + if not reader.fieldnames or not required.issubset(reader.fieldnames): + raise ValueError( + f"{path} must contain payee, category, inflow, and outflow columns" + ) + + for row in reader: + payee = normalize_payee(row.get("payee") or "") + category = (row.get("category") or "").strip() + inflow = _amount_cents(row.get("inflow") or "") + outflow = _amount_cents(row.get("outflow") or "") + if not payee or not category: + continue + if inflow is not None and inflow > 0 and (outflow is None or outflow == 0): + pattern = CategoryPattern(payee, category, "inflow", inflow) + key = cls._key(pattern) + existing = patterns.get(key) + patterns[key] = CategoryPattern( + payee, category, "inflow", inflow, + (existing.count if existing else 0) + 1, + ) + elif outflow is not None and outflow > 0 and (inflow is None or inflow == 0): + pattern = CategoryPattern(payee, category, "outflow", outflow) + key = cls._key(pattern) + existing = patterns.get(key) + patterns[key] = CategoryPattern( + payee, category, "outflow", outflow, + (existing.count if existing else 0) + 1, + ) + return cls(patterns) + + def merge(self, other: "CategoryHistory") -> int: + """Union patterns, returning how many new patterns were added.""" + changes = 0 + for key, pattern in other.patterns.items(): + existing = self.patterns.get(key) + if existing is None or pattern.count > existing.count: + self.patterns[key] = pattern + changes += 1 + return changes + + def save(self, path: str | Path) -> None: + """Write a deterministic, portable JSON sidecar using an atomic replace.""" + memory_path = Path(path) + memory_path.parent.mkdir(parents=True, exist_ok=True) + payload = { + "version": _MEMORY_VERSION, + "patterns": [ + { + "payee": pattern.payee, + "category": pattern.category, + "direction": pattern.direction, + "amount_cents": pattern.amount_cents, + "count": pattern.count, + } + for pattern in sorted( + self.patterns.values(), + key=lambda item: ( + item.payee, + item.category.casefold(), + item.direction, + item.amount_cents, + ), + ) + ], + } + temporary_path = memory_path.with_name(memory_path.name + ".tmp") + temporary_path.write_text( + json.dumps(payload, indent=2, ensure_ascii=False) + "\n", + encoding="utf-8", + ) + temporary_path.replace(memory_path) + + def suggest( + self, + payee: str, + amount_milli: int, + category_ids_by_name: dict[str, list[str]], + ) -> CategorySuggestion | None: + """Rank historical categories using payee similarity, direction, and amount.""" + payee_key = normalize_payee(payee) + if not payee_key or not self.patterns: + return None + + direction = "inflow" if amount_milli > 0 else "outflow" + amount_cents = round(abs(amount_milli) / 10) + category_scores: dict[str, float] = {} + category_names: dict[str, str] = {} + category_support: dict[str, int] = {} + category_exact: dict[str, bool] = {} + + for pattern in self.patterns.values(): + if pattern.direction != direction: + continue + payee_similarity = SequenceMatcher( + None, payee_key, pattern.payee, autojunk=False + ).ratio() + if payee_similarity < _MIN_PAYEE_SIMILARITY: + continue + + ratio = (amount_cents + _AMOUNT_SMOOTHING_CENTS) / ( + pattern.amount_cents + _AMOUNT_SMOOTHING_CENTS + ) + amount_similarity = 1 / (1 + abs(math.log(ratio))) + weight = payee_similarity * (amount_similarity ** 2) * math.sqrt(pattern.count) + category_key = pattern.category.casefold() + category_scores[category_key] = category_scores.get(category_key, 0.0) + weight + category_names.setdefault(category_key, pattern.category) + category_support[category_key] = ( + category_support.get(category_key, 0) + pattern.count + ) + if payee_key == pattern.payee: + category_exact[category_key] = True + else: + category_exact.setdefault(category_key, False) + + total_score = sum(category_scores.values()) + if total_score <= 0: + return None + + best_category = max(category_scores, key=category_scores.get) + confidence = category_scores[best_category] / total_score + sufficiently_supported = ( + category_exact.get(best_category, False) + or category_support[best_category] >= 2 + ) + applied = ( + confidence >= _MIN_CATEGORY_SHARE + and sufficiently_supported + ) + + category_name = category_names[best_category] + matching_ids = category_ids_by_name.get(category_name.casefold(), []) + category_id = matching_ids[0] if applied and len(matching_ids) == 1 else None + return CategorySuggestion( + category_name=category_name, + confidence=confidence, + category_id=category_id, + applied=category_id is not None, + ) diff --git a/main.py b/main.py index 6f3ae14..e3fc768 100644 --- a/main.py +++ b/main.py @@ -1,11 +1,14 @@ import argparse +import csv import json from datetime import date, datetime, timedelta, timezone +from decimal import Decimal from pathlib import Path from typing import Any import ynab +from category_suggester import CategoryHistory from wf_credit_extractor import WellsFargoPayloadExtractor as CreditExtractor from wf_debit_extractor import WellsFargoPayloadExtractor as DebitExtractor from ynab_client import YNABClient, YNABTransaction @@ -166,6 +169,167 @@ def _run_ynab_transactions(days: int) -> None: print(json.dumps(transactions, indent=2)) +def _parse_record_days(value: str) -> int: + """Validate the supported lookback for transaction CSV exports.""" + try: + days = int(value) + except ValueError as exc: + raise argparse.ArgumentTypeError("days must be an integer from 1 to 240") from exc + if not 1 <= days <= 240: + raise argparse.ArgumentTypeError("days must be between 1 and 240") + return days + + +def _write_transactions_csv( + transactions: list[YNABTransaction], + output_path: Path, +) -> int: + """Write YNAB transactions as one CSV row per category line item.""" + output_path.parent.mkdir(parents=True, exist_ok=True) + fieldnames = ["payee", "category", "inflow", "outflow"] + row_count = 0 + + with output_path.open("w", encoding="utf-8", newline="") as csv_file: + writer = csv.DictWriter(csv_file, fieldnames=fieldnames) + writer.writeheader() + + for transaction in transactions: + line_items = transaction.get("subtransactions") or [transaction] + for line_item in line_items: + amount = Decimal(line_item["amount_milliCurrency"]) / Decimal(1000) + if amount >= 0: + inflow = f"{amount:.2f}" + outflow = "" + else: + inflow = "" + outflow = f"{abs(amount):.2f}" + + writer.writerow({ + "payee": transaction.get("payee_name") or "", + "category": line_item.get("category_name") or "", + "inflow": inflow, + "outflow": outflow, + }) + row_count += 1 + + return row_count + + +def _record_transactions_csv( + ynab_client: YNABClient, + days: int, + output_path: Path, +) -> None: + """Fetch credit and debit account transactions and save their CSV rows.""" + target_extractors = set(FLAG_TO_EXTRACTOR.values()) + target_names = {"Credit Card", "Debit Card", "Wells Fargo Bank"} + target_account_ids = { + account["id"] + for account in ynab_client.account_configs + if account.get("extractor") in target_extractors + or account.get("name") in target_names + } + if not target_account_ids: + raise ValueError( + "No Credit Card or Wells Fargo Bank accounts found in ynab.yml" + ) + + transactions = ynab_client.get_transactions_for_days( + days=days, + include_subtransactions=True, + ) + selected_transactions = [ + transaction + for transaction in transactions + if transaction["account_id"] in target_account_ids + ] + + history, memory_was_readable = _load_category_history( + output_path, + report_empty=False, + ) + row_count = _write_transactions_csv(selected_transactions, output_path) + memory_path = CategoryHistory.memory_path_for_csv(output_path) + try: + history.merge(CategoryHistory.from_csv(output_path)) + history.save(memory_path) + action = "Updated" if memory_was_readable else "Rebuilt" + print(f"[category] {action} reusable category patterns in {memory_path}") + except (OSError, ValueError) as exc: + print(f"[category] Could not update {memory_path}: {exc}") + print( + f"[ynab] Wrote {row_count} transaction row(s) from the past {days} day(s) " + f"to {output_path}" + ) + + +def _load_category_history( + csv_path: Path, + *, + report_empty: bool = True, +) -> tuple[CategoryHistory, bool]: + """Load persistent patterns and merge the current CSV without losing old data.""" + memory_path = CategoryHistory.memory_path_for_csv(csv_path) + memory_exists = memory_path.exists() + can_update_memory = True + + if memory_exists: + try: + history = CategoryHistory.load(memory_path) + except ( + OSError, + ValueError, + TypeError, + KeyError, + AttributeError, + json.JSONDecodeError, + ) as exc: + print(f"[category] Could not read {memory_path}: {exc}") + history = CategoryHistory() + can_update_memory = False + else: + history = CategoryHistory() + + try: + current_csv = CategoryHistory.from_csv(csv_path) + added_patterns = history.merge(current_csv) + if can_update_memory and (not memory_exists or added_patterns): + history.save(memory_path) + except FileNotFoundError: + pass + except (OSError, ValueError, csv.Error) as exc: + print(f"[category] Could not read transaction history {csv_path}: {exc}") + + if report_empty and not history.patterns: + print( + f"[category] No usable history found in {csv_path} or {memory_path}; " + "new transactions will be imported without category suggestions." + ) + return history, can_update_memory + + +def _apply_category_suggestions( + transactions: list[dict], + history: CategoryHistory, + category_ids_by_name: dict[str, list[str]], +) -> None: + """Annotate import candidates and set category_id only for confident matches.""" + for transaction in transactions: + suggestion = history.suggest( + transaction.get("payee_name") or "", + transaction["amount"], + category_ids_by_name, + ) + if suggestion is None: + continue + + transaction["_category_suggestion"] = suggestion.category_name + transaction["_category_confidence"] = suggestion.confidence + transaction["_category_applied"] = suggestion.applied + if suggestion.category_id is not None: + transaction["category_id"] = suggestion.category_id + + def _filter_wf_from_date(wf_transactions: list[dict], start_iso: str) -> list[dict]: """Keep only WF transactions whose transactionDate is on or after start_iso.""" return [ @@ -353,46 +517,69 @@ def _preview_import(classification: _ImportClassification) -> None: """Print a formatted preview of the import classification.""" col_date = 12 col_amount = 14 - col_desc = 50 - header = f" {'Date':<{col_date}} {'Amount':>{col_amount}} {'Description':<{col_desc}}" - separator = f" {'-' * col_date} {'-' * col_amount} {'-' * col_desc}" + col_desc = 34 + col_category = 28 + report_width = col_date + col_amount + col_desc + col_category + 10 + header = ( + f" {'Date':<{col_date}} {'Amount':>{col_amount}} " + f"{'Description':<{col_desc}} {'Category suggestion':<{col_category}}" + ) + separator = ( + f" {'-' * col_date} {'-' * col_amount} {'-' * col_desc} " + f"{'-' * col_category}" + ) + + def category_text(txn: dict) -> str: + suggestion = txn.get("_category_suggestion") + if not suggestion: + return "(uncategorized)" + confidence = txn.get("_category_confidence", 0.0) + prefix = "" if txn.get("_category_applied") else "? " + return f"{prefix}{suggestion} ({confidence:.0%})" + + def row_text(txn: dict, category: str) -> str: + desc = (txn.get("payee_name") or "")[:col_desc] + category = category[:col_category] + return ( + f" {txn['date']:<{col_date}} " + f"{_format_amount_dollars(txn['amount']):>{col_amount}} " + f"{desc:<{col_desc}} {category:<{col_category}}" + ) # --- New --- - print(f"\n{'=' * 80}") + print(f"\n{'=' * report_width}") print(f" NEW TRANSACTIONS ({len(classification.new)}) — will be created with BLUE flag") - print(f"{'=' * 80}") + print(f"{'=' * report_width}") if classification.new: print(header) print(separator) for txn in classification.new: - desc = (txn.get("payee_name") or "")[:col_desc] - print(f" {txn['date']:<{col_date}} {_format_amount_dollars(txn['amount']):>{col_amount}} {desc:<{col_desc}}") + print(row_text(txn, category_text(txn))) else: print(" (none)") # --- Pending→Posted --- - print(f"\n{'=' * 80}") + print(f"\n{'=' * report_width}") print(f" PENDING → POSTED ({len(classification.pending_to_posted)}) — existing uncleared will be marked cleared") - print(f"{'=' * 80}") + print(f"{'=' * report_width}") if classification.pending_to_posted: print(header) print(separator) for txn, matched in classification.pending_to_posted: - desc = (txn.get("payee_name") or "")[:col_desc] - print(f" {txn['date']:<{col_date}} {_format_amount_dollars(txn['amount']):>{col_amount}} {desc:<{col_desc}}") + existing_category = matched.get("category_name") or "(unchanged)" + print(row_text(txn, existing_category)) else: print(" (none)") # --- Duplicates --- - print(f"\n{'=' * 80}") + print(f"\n{'=' * report_width}") print(f" POSSIBLE DUPLICATES ({len(classification.duplicates)}) — will be created with RED flag") - print(f"{'=' * 80}") + print(f"{'=' * report_width}") if classification.duplicates: print(header) print(separator) for txn in classification.duplicates: - desc = (txn.get("payee_name") or "")[:col_desc] - print(f" {txn['date']:<{col_date}} {_format_amount_dollars(txn['amount']):>{col_amount}} {desc:<{col_desc}}") + print(row_text(txn, category_text(txn))) else: print(" (none)") @@ -420,6 +607,7 @@ def _execute_import( date=date.fromisoformat(t["date"]), amount=t["amount"], payee_name=t["payee_name"], + category_id=t.get("category_id"), cleared=ynab.TransactionClearedStatus(t["cleared"]), flag_color=ynab.TransactionFlagColor(t["flag_color"]), import_id=t["import_id"], @@ -478,6 +666,24 @@ def main() -> None: metavar="DAYS", help="List YNAB transactions for the past DAYS (excluding reconciled)", ) + parser.add_argument( + "--record-transactions", + "--record-trasactions", + dest="record_transactions", + type=_parse_record_days, + nargs="?", + const=60, + default=None, + metavar="DAYS", + help="Write Credit Card and Wells Fargo Bank transactions to CSV (default: 60 days; max: 240)", + ) + parser.add_argument( + "--transactions-csv", + type=Path, + default=Path("transactions.csv"), + metavar="PATH", + help="CSV history/export path for category learning (default: transactions.csv)", + ) parser.add_argument( "--ynab-import", action="store_true", @@ -488,6 +694,12 @@ def main() -> None: # Load YNAB config to map extractors -> account IDs ynab_client = YNABClient("ynab.yml") + if args.record_transactions is not None: + _record_transactions_csv( + ynab_client, + args.record_transactions, + args.transactions_csv, + ) account_configs = ynab_client.account_configs # Build extractor_name -> [account_id, ...] mapping from ynab.yml @@ -568,6 +780,15 @@ def main() -> None: if args.ynab_import and extracted_by_extractor: all_ynab_ready: list[dict] = [] balances: dict[str, float] = {} # label -> balance amount + category_history, _ = _load_category_history(args.transactions_csv) + try: + category_ids_by_name = ynab_client.get_category_ids_by_name() + except Exception as exc: + print( + f"[category] Could not load YNAB category IDs ({exc}); " + "suggestions will not be assigned." + ) + category_ids_by_name = {} for extractor_name, wf_txns in extracted_by_extractor.items(): account_ids_for_ext = extractor_to_accounts.get(extractor_name, []) @@ -596,6 +817,11 @@ def main() -> None: ynab_ready = _convert_wf_to_ynab_transactions( import_candidates, target_account_id, ) + _apply_category_suggestions( + ynab_ready, + category_history, + category_ids_by_name, + ) # Classify against existing YNAB transactions classification = _classify_for_import(ynab_ready, scoped_ynab) @@ -658,8 +884,16 @@ def main() -> None: if args.ynab_transactions is not None: _run_ynab_transactions(args.ynab_transactions) - if not args.credit and not args.debit and args.ynab_transactions is None: - print("No input provided. Use --credit, --debit, --ynab-import, and/or --ynab-transactions.") + if ( + not args.credit + and not args.debit + and args.ynab_transactions is None + and args.record_transactions is None + ): + print( + "No input provided. Use --credit, --debit, --ynab-import, " + "--ynab-transactions, and/or --record-transactions." + ) if __name__ == "__main__": diff --git a/test_import.py b/test_import.py index 2ff5413..14defde 100644 --- a/test_import.py +++ b/test_import.py @@ -1,7 +1,16 @@ """Tests for the YNAB import helpers in main.py.""" +import csv +import tempfile +from pathlib import Path +from types import SimpleNamespace + +from category_suggester import CategoryHistory, normalize_payee from main import ( _wf_to_ynab_millis, _build_import_ids, _convert_wf_to_ynab_transactions, - _classify_for_import, _format_amount_dollars, _preview_import, + _ImportClassification, _apply_category_suggestions, _classify_for_import, + _execute_import, _format_amount_dollars, _load_category_history, + _preview_import, + _write_transactions_csv, ) @@ -141,6 +150,208 @@ def test_preview(): print("test_preview: PASS") +def _write_category_csv(path: Path, rows: list[tuple[str, str, str, str]]) -> None: + with path.open("w", encoding="utf-8", newline="") as csv_file: + writer = csv.writer(csv_file) + writer.writerow(["payee", "category", "inflow", "outflow"]) + writer.writerows(rows) + + +def test_payee_normalization_and_amount_aware_categories(): + amazon_a = "AMAZON MKTPLACE PMTS Amzn.com/billWA" + amazon_b = "AMAZON MKTPL*BY44297W1 Amzn.com/billWA" + normalized_a = normalize_payee(amazon_a) + normalized_b = normalize_payee(amazon_b) + assert "by44297w1" not in normalized_b + assert normalized_a.startswith("amazon") + assert normalized_b.startswith("amazon") + assert normalize_payee("SHEETZ #1234") == normalize_payee("SHEETZ #5678") + + with tempfile.TemporaryDirectory() as temp_dir: + csv_path = Path(temp_dir) / "transactions.csv" + _write_category_csv(csv_path, [ + (amazon_a, "Family", "", "50.00"), + (amazon_a, "Family", "", "80.00"), + ("SPLIT SHOP", "Family", "", "50.00"), + ("SPLIT SHOP", "Dining Out", "", "20.00"), + ("SHEETZ", "Gas", "", "40.00"), + ("SHEETZ", "Gas", "", "40.00"), + ("SHEETZ", "Dining Out", "", "2.00"), + ("SHEETZ", "Dining Out", "", "2.00"), + ("PUBLIX", "Groceries", "", "40.00"), + ("PUBLIX", "Dining Out", "", "2.00"), + ("REFUND SHOP", "Refunds", "12.00", ""), + ("REFUND SHOP", "Shopping", "", "12.00"), + ]) + history = CategoryHistory.from_csv(csv_path) + category_ids = { + "family": ["family-id"], + "gas": ["gas-id"], + "dining out": ["dining-id"], + "groceries": ["grocery-id"], + "refunds": ["refund-id"], + "shopping": ["shopping-id"], + } + + wf_transactions = [ + {"transactionAmount": 50.00, "transactionDate": 1773817200000, + "transactionDescription": amazon_b, "status": "POSTED"}, + {"transactionAmount": 40.00, "transactionDate": 1773817200000, + "transactionDescription": "SHEETZ", "status": "POSTED"}, + {"transactionAmount": 2.00, "transactionDate": 1773817200000, + "transactionDescription": "SHEETZ", "status": "POSTED"}, + {"transactionAmount": 2.00, "transactionDate": 1773817200000, + "transactionDescription": "PUBLIX", "status": "POSTED"}, + {"transactionAmount": 20.00, "transactionDate": 1773817200000, + "transactionDescription": "SHEETZ", "status": "POSTED"}, + {"transactionAmount": 12.00, "transactionDate": 1773817200000, + "transactionDescription": "REFUND SHOP", "status": "POSTED", + "debitCreditType": "CREDIT"}, + ] + ready = _convert_wf_to_ynab_transactions(wf_transactions, "account-id") + _apply_category_suggestions(ready, history, category_ids) + + assert ready[0]["category_id"] == "family-id" + assert ready[1]["category_id"] == "gas-id" + assert ready[2]["category_id"] == "dining-id" + assert ready[3]["category_id"] == "dining-id" + assert "category_id" not in ready[4] + assert ready[4]["_category_applied"] is False + assert ready[5]["category_id"] == "refund-id" + split_patterns = [ + pattern for pattern in history.patterns.values() + if pattern.payee == "split shop" + ] + assert {pattern.category for pattern in split_patterns} == {"Family", "Dining Out"} + assert {pattern.amount_cents for pattern in split_patterns} == {2000, 5000} + print("test_payee_normalization_and_amount_aware_categories: PASS") + + +def test_category_memory_retains_and_merges_patterns(): + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + csv_path = root / "transactions.csv" + memory_path = CategoryHistory.memory_path_for_csv(csv_path) + + _write_category_csv(csv_path, [ + ("OLD MERCHANT", "Family", "", "20.00"), + ("OLD MERCHANT", "Family", "", "20.00"), + ]) + history = CategoryHistory.from_csv(csv_path) + history.save(memory_path) + + _write_category_csv(csv_path, [ + ("NEW MERCHANT", "Dining Out", "", "8.00"), + ]) + history = CategoryHistory.load(memory_path) + latest_snapshot = CategoryHistory.from_csv(csv_path) + history.merge(latest_snapshot) + history.save(memory_path) + + copied_memory = root / "second_machine" / "category_memory.json" + history.save(copied_memory) + copied_history = CategoryHistory.load(copied_memory) + + patterns = copied_history.patterns + assert any(pattern.payee == "old merchant" for pattern in patterns.values()) + assert any(pattern.payee == "new merchant" for pattern in patterns.values()) + assert next( + pattern.count for pattern in patterns.values() + if pattern.payee == "old merchant" + ) == 2 + print("test_category_memory_retains_and_merges_patterns: PASS") + + +def test_split_export_becomes_separate_category_patterns(): + with tempfile.TemporaryDirectory() as temp_dir: + csv_path = Path(temp_dir) / "transactions.csv" + rows_written = _write_transactions_csv([ + { + "id": "transaction-id", + "date": "2026-03-18", + "payee_name": "SPLIT SHOP", + "category_name": "Split", + "amount_milliCurrency": -70000, + "memo": None, + "account_id": "account-id", + "account_name": "Credit Card", + "cleared": "cleared", + "import_id": None, + "subtransactions": [ + {"category_name": "Family", "amount_milliCurrency": -50000}, + {"category_name": "Dining Out", "amount_milliCurrency": -20000}, + ], + }, + ], csv_path) + + with csv_path.open("r", encoding="utf-8", newline="") as csv_file: + rows = list(csv.DictReader(csv_file)) + history = CategoryHistory.from_csv(csv_path) + + assert rows_written == 2 + assert [row["payee"] for row in rows] == ["SPLIT SHOP", "SPLIT SHOP"] + assert {row["category"] for row in rows} == {"Family", "Dining Out"} + assert {pattern.amount_cents for pattern in history.patterns.values()} == {2000, 5000} + print("test_split_export_becomes_separate_category_patterns: PASS") + + +def test_ambiguous_category_name_is_not_assigned(): + with tempfile.TemporaryDirectory() as temp_dir: + csv_path = Path(temp_dir) / "transactions.csv" + _write_category_csv(csv_path, [("COFFEE SHOP", "Dining", "", "5.00")]) + history = CategoryHistory.from_csv(csv_path) + suggestion = history.suggest( + "COFFEE SHOP", + -5000, + {"dining": ["category-one", "category-two"]}, + ) + assert suggestion is not None + assert suggestion.applied is False + assert suggestion.category_id is None + print("test_ambiguous_category_name_is_not_assigned: PASS") + + +def test_missing_category_history_falls_back_empty(): + with tempfile.TemporaryDirectory() as temp_dir: + history, can_update_memory = _load_category_history( + Path(temp_dir) / "missing.csv", + report_empty=False, + ) + assert not history.patterns + assert can_update_memory is True + print("test_missing_category_history_falls_back_empty: PASS") + + +def test_confident_category_is_sent_to_ynab(): + account_id = "11111111-1111-4111-8111-111111111111" + category_id = "22222222-2222-4222-8222-222222222222" + classification = _ImportClassification() + classification.new.append({ + "account_id": account_id, + "date": "2026-03-18", + "amount": -5000, + "payee_name": "COFFEE SHOP", + "category_id": category_id, + "cleared": "cleared", + "flag_color": "blue", + "import_id": "YNAB:-5000:2026-03-18:1", + "approved": False, + }) + + class FakeYNABClient: + def __init__(self): + self.transactions = [] + + def create_transactions(self, transactions): + self.transactions = transactions + return SimpleNamespace(transaction_ids=["created-id"], duplicate_import_ids=[]) + + client = FakeYNABClient() + _execute_import(classification, client) + assert str(client.transactions[0].category_id) == category_id + print("test_confident_category_is_sent_to_ynab: PASS") + + if __name__ == "__main__": test_wf_to_ynab_millis() test_build_import_ids() @@ -148,5 +359,11 @@ if __name__ == "__main__": test_convert_wf_to_ynab_transactions() test_classify_for_import() test_preview() + test_payee_normalization_and_amount_aware_categories() + test_category_memory_retains_and_merges_patterns() + test_split_export_becomes_separate_category_patterns() + test_ambiguous_category_name_is_not_assigned() + test_missing_category_history_falls_back_empty() + test_confident_category_is_sent_to_ynab() print("\nALL TESTS PASSED") diff --git a/ynab_client.py b/ynab_client.py index 86614d1..b8b8346 100644 --- a/ynab_client.py +++ b/ynab_client.py @@ -2,6 +2,7 @@ import yaml from datetime import datetime, timedelta from pathlib import Path from typing import TypedDict +from typing_extensions import NotRequired import ynab @@ -19,6 +20,14 @@ class YNABTransaction(TypedDict): account_name: str cleared: str import_id: str | None + subtransactions: NotRequired[list["YNABSubTransaction"]] + + +class YNABSubTransaction(TypedDict): + """Category and amount data for one line of a split transaction.""" + + category_name: str | None + amount_milliCurrency: int class YNABConfig(TypedDict): @@ -75,6 +84,7 @@ class YNABClient: self, days: int = 30, exclude_cleared_statuses: list[str] | None = None, + include_subtransactions: bool = False, ) -> list[YNABTransaction]: """ Fetch transactions from configured accounts for the past x days. @@ -83,6 +93,7 @@ class YNABClient: days: Number of days to look back (default: 30) exclude_cleared_statuses: Optional list of normalized statuses to skip, for example ["reconciled"] + include_subtransactions: Include category-level amounts for split transactions. Returns: List of transactions with transformed data @@ -149,6 +160,17 @@ class YNABClient: "cleared": cleared_status, "import_id": txn.import_id, } + if include_subtransactions: + subtransactions: list[YNABSubTransaction] = [] + for sub in txn.subtransactions or []: + category_name = sub.category_name + if not category_name and sub.category_id: + category_name = category_map.get(str(sub.category_id)) + subtransactions.append({ + "category_name": category_name, + "amount_milliCurrency": sub.amount, + }) + processed["subtransactions"] = subtransactions transactions.append(processed) except Exception as e: @@ -173,6 +195,17 @@ class YNABClient: return account.balance raise ValueError(f"Account {account_id} not found") + def get_category_ids_by_name(self) -> dict[str, list[str]]: + """Return category IDs grouped by case-insensitive category name.""" + response = self.categories_api.get_categories(self.plan_id) + category_ids: dict[str, list[str]] = {} + for category_group in response.data.category_groups: + for category in category_group.categories: + if category.id and category.name: + key = category.name.strip().casefold() + category_ids.setdefault(key, []).append(str(category.id)) + return category_ids + def get_last_reconciled_date( self, account_ids: list[str] | None = None,