From 8f023185429bd4dd6411c2cea231be3bf6ab265f Mon Sep 17 00:00:00 2001 From: Ryan Chen Date: Mon, 12 Oct 2026 17:32:38 -0400 Subject: [PATCH] Mirror YNAB budget into Postgres with delta sync and change log YNAB has no event feed, so keep a local copy in ynab_* tables synced via server_knowledge delta requests: full download on first run, deltas every YNAB_SYNC_INTERVAL seconds and on demand before each tool call, full resync daily. Each sync diffs against the mirror and appends real edits (transactions, assignments, renames) to ynab_changes, the edit history YNAB doesn't keep. Agent YNAB tools now answer from the mirror: budget summary (spending computed from transactions, since YNAB's month activity nets in credit card payment categories), account balances, transaction search with splits expanded, spending reports by category/payee/account/month, category budget, and recent changes. Adds an admin-only read-only ynab_sql tool guarded by a sqlglot allowlist (scope-resolved tables, no dangerous functions), a READ ONLY transaction, a timeout, and a row cap. Replaces utils/ynab_service.py and drops the unused ynab SDK. Co-Authored-By: Claude Opus 5.5 --- .env.example | 2 + CLAUDE.md | 5 +- Makefile | 9 +- app.py | 16 +- blueprints/conversation/agents.py | 211 +------- blueprints/conversation/prompts.py | 8 +- blueprints/ynab/__init__.py | 1 + blueprints/ynab/__main__.py | 22 + blueprints/ynab/models.py | 173 ++++++ blueprints/ynab/sql_tool.py | 151 ++++++ blueprints/ynab/sync.py | 512 ++++++++++++++++++ blueprints/ynab/tools.py | 447 +++++++++++++++ config/db.py | 1 + docker-compose.yml | 1 + docs/TASKS.md | 10 +- .../16_20261012172428_add_ynab_mirror.py | 239 ++++++++ pyproject.toml | 2 +- tests/unit/test_ynab_mirror.py | 249 +++++++++ tests/unit/test_ynab_service.py | 254 --------- utils/ynab_service.py | 405 -------------- uv.lock | 31 +- 21 files changed, 1846 insertions(+), 903 deletions(-) create mode 100644 blueprints/ynab/__init__.py create mode 100644 blueprints/ynab/__main__.py create mode 100644 blueprints/ynab/models.py create mode 100644 blueprints/ynab/sql_tool.py create mode 100644 blueprints/ynab/sync.py create mode 100644 blueprints/ynab/tools.py create mode 100644 migrations/models/16_20261012172428_add_ynab_mirror.py create mode 100644 tests/unit/test_ynab_mirror.py delete mode 100644 tests/unit/test_ynab_service.py delete mode 100644 utils/ynab_service.py diff --git a/.env.example b/.env.example index b6d938e..dfcc9d8 100644 --- a/.env.example +++ b/.env.example @@ -62,6 +62,8 @@ OIDC_USE_DISCOVERY=true YNAB_ACCESS_TOKEN=your-ynab-personal-access-token # Optional: Specify a budget ID, or leave empty to use the default/first budget YNAB_BUDGET_ID= +# Seconds between background YNAB delta syncs (agent tools also sync on demand) +YNAB_SYNC_INTERVAL=900 # Mealie Configuration # Base URL for your Mealie instance (e.g., http://192.168.1.5:9000 or https://mealie.example.com) diff --git a/CLAUDE.md b/CLAUDE.md index a5ac426..6aceebb 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -80,6 +80,7 @@ make redeploy # git pull + scripts/deploy.sh (build new color, wait healthy, - `imessage/` - Twilio webhook receiver for iMessage (HMAC-validated, number allowlist) - `whatsapp/` - Twilio webhook receiver for WhatsApp (TWIML replies, SendBlue send support) - `scheduled_messages/` - Scheduled message delivery with background scheduler + - `ynab/` - Postgres mirror of the YNAB budget (`ynab_*` tables), delta-synced via `server_knowledge` (background loop + on-demand before each tool call, full resync daily); `ynab_changes` is the edit history YNAB itself doesn't keep. Agent tools in `tools.py`; admin-only read-only `ynab_sql` in `sql_tool.py`. Manual sync: `make ynab-sync` / `make ynab-resync` - `config/` - Configuration modules - `db.py` - Database setup - `oidc_config.py` - OIDC authentication configuration @@ -88,7 +89,7 @@ make redeploy # git pull + scripts/deploy.sh (build new color, wait healthy, - `cleaner.py` - PDF cleaning and summarization - `image_process.py` - Image description with LLM - `request.py` - Paperless-NGX API client - - `obsidian_service.py`, `mealie_service.py`, `ynab_service.py`, `s3_client.py`, `simbakit_service.py` - integrations with Obsidian notes, Mealie recipes, YNAB budgets, S3, and SimbaKit + - `obsidian_service.py`, `mealie_service.py`, `s3_client.py`, `simbakit_service.py` - integrations with Obsidian notes, Mealie recipes, YNAB budgets, S3, and SimbaKit - `homebox_mcp.py` - Homebox inventory tools via the `homebox-mcp` stdio MCP server; admin-only (compiled into `app.admin_agent`, picked per user by `get_agent_for`) - `bookstack_mcp.py` - BookStack wiki tools (read/create/update, no deletes) via the `bookstack-mcp` npm stdio server; admin-only like Homebox - `mcp_tools.py` - shared allowlist + admin-gate wrapping for stdio MCP integrations @@ -100,7 +101,7 @@ make redeploy # git pull + scripts/deploy.sh (build new color, wait healthy, - `link_imessage.py` - Link iMessage contacts to users - `deploy.sh` - Blue-green deployment script - `migrations/` - Database migration files -- `tests/` - pytest test suite (`tests/unit/` covers email/crypto, YNAB, Obsidian, SimbaKit, rate limiting, user model) +- `tests/` - pytest test suite (`tests/unit/` covers email/crypto, YNAB mirror, Obsidian, SimbaKit, rate limiting, user model) **simbakit/** (standalone subproject): - Self-hosted PetKit device poller + web dashboard for Simba's smart devices (feeder, litter box, fountain, purifier) diff --git a/Makefile b/Makefile index bad5327..0308ba1 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: deploy redeploy build up down restart logs migrate migrate-new migrate-history frontend test test-cov reindex reindex-obsidian +.PHONY: deploy redeploy build up down restart logs migrate migrate-new migrate-history frontend test test-cov reindex reindex-obsidian ynab-sync ynab-resync # Blue-green: detect the running color (default blue when nothing is running) ACTIVE_COLOR := $(shell docker compose ps --status running --services 2>/dev/null | grep -oE 'raggr-(blue|green)' | head -1 | cut -d- -f2) @@ -50,6 +50,13 @@ reindex-obsidian: asyncio.run(index_obsidian_documents()); \ print('Obsidian reindex complete')" +# YNAB mirror: delta sync, or full re-download +ynab-sync: + $(COMPOSE) exec $(RAGGR) python -m blueprints.ynab + +ynab-resync: + $(COMPOSE) exec $(RAGGR) python -m blueprints.ynab --full + # Database migrations migrate: $(COMPOSE) exec $(RAGGR) aerich upgrade diff --git a/app.py b/app.py index 1558f2a..8203f36 100644 --- a/app.py +++ b/app.py @@ -30,6 +30,8 @@ import blueprints.users.models from config.db import TORTOISE_CONFIG from metrics import render_metrics from blueprints.conversation.agents import build_agent +from blueprints.ynab.sql_tool import ynab_sql +from blueprints.ynab.sync import ynab_configured, ynab_sync_loop from utils.bookstack_mcp import load_bookstack_tools from utils.homebox_mcp import load_homebox_tools @@ -44,7 +46,7 @@ logging.basicConfig( ) # Ensure YNAB and Mealie loggers are visible -logging.getLogger("utils.ynab_service").setLevel(logging.INFO) +logging.getLogger("blueprints.ynab.sync").setLevel(logging.INFO) logging.getLogger("utils.mealie_service").setLevel(logging.INFO) logging.getLogger("blueprints.conversation.agents").setLevel(logging.INFO) @@ -109,6 +111,8 @@ async def lifespan(): await checkpointer.setup() app.main_agent = build_agent(checkpointer) admin_tools = await load_homebox_tools() + await load_bookstack_tools() + if ynab_configured(): + admin_tools.append(ynab_sql) app.admin_agent = ( build_agent(checkpointer, admin_tools) if admin_tools else app.main_agent ) @@ -118,14 +122,16 @@ async def lifespan(): await ensure_beli_nightly_event() - watcher_task = None + background_tasks = [] if os.getenv("OBSIDIAN_CONTINUOUS_SYNC") == "true": - watcher_task = asyncio.create_task(_obsidian_sync_loop()) + background_tasks.append(asyncio.create_task(_obsidian_sync_loop())) + if ynab_configured(): + background_tasks.append(asyncio.create_task(ynab_sync_loop())) yield - if watcher_task is not None: - watcher_task.cancel() + for task in background_tasks: + task.cancel() logging.info("Closing Tortoise ORM connections...") await Tortoise.close_connections() diff --git a/blueprints/conversation/agents.py b/blueprints/conversation/agents.py index c059ef9..e8d40c7 100644 --- a/blueprints/conversation/agents.py +++ b/blueprints/conversation/agents.py @@ -14,11 +14,12 @@ from tavily import AsyncTavilyClient from blueprints.conversation.memory import save_memory from blueprints.conversation.prompts import SIMBA_SYSTEM_PROMPT from blueprints.rag.logic import query_vector_store +from blueprints.ynab.sync import ynab_configured +from blueprints.ynab.tools import YNAB_TOOLS from utils.google_service import GoogleService from utils.mealie_service import MealieService from utils.obsidian_service import ObsidianService from utils.simbakit_service import SimbaKitService -from utils.ynab_service import YNABService # Load environment variables load_dotenv() @@ -51,14 +52,6 @@ model_with_fallback = cast( ) client = AsyncTavilyClient(api_key=os.getenv("TAVILY_API_KEY", "")) -# Initialize YNAB service (will only work if YNAB_ACCESS_TOKEN is set) -try: - ynab_service = YNABService() - ynab_enabled = True -except (ValueError, Exception) as e: - print(f"YNAB service not initialized: {e}") - ynab_enabled = False - # Initialize SimbaKit service (will only work if SIMBAKIT_URL is set) try: simbakit_service = SimbaKitService() @@ -141,194 +134,6 @@ async def simba_search(query: str): return serialized, docs -@tool -def ynab_budget_summary() -> str: - """Get overall YNAB budget health: to-be-budgeted, total budgeted, and available amounts.""" - if not ynab_enabled: - return "YNAB integration is not configured. Please set YNAB_ACCESS_TOKEN environment variable." - - try: - summary = ynab_service.get_budget_summary() - return summary["summary"] - except Exception as e: - return f"Error fetching budget summary: {str(e)}" - - -@tool -def ynab_search_transactions( - start_date: str = "", - end_date: str = "", - category_name: str = "", - payee_name: str = "", -) -> str: - """Search YNAB transactions by date range, category, or payee. - - Args: - start_date: YYYY-MM-DD (default: 30 days ago) - end_date: YYYY-MM-DD (default: today) - category_name: Partial match on category name - payee_name: Partial match on payee/store name - """ - if not ynab_enabled: - return "YNAB integration is not configured. Please set YNAB_ACCESS_TOKEN environment variable." - - try: - result = ynab_service.get_transactions( - start_date=start_date or None, - end_date=end_date or None, - category_name=category_name or None, - payee_name=payee_name or None, - ) - - if result["count"] == 0: - return "No transactions found matching the specified criteria." - - # Format transactions for readability - txn_list = [] - for txn in result["transactions"][:10]: # Limit to 10 for readability - txn_list.append( - f"- {txn['date']}: {txn['payee']} - ${abs(txn['amount']):.2f} ({txn['category'] or 'Uncategorized'})" - ) - - return ( - f"Found {result['count']} transactions from {result['start_date']} to {result['end_date']}. " - f"Total: ${abs(result['total_amount']):.2f}\n\n" - + "\n".join(txn_list) - + ( - f"\n\n(Showing first 10 of {result['count']} transactions)" - if result["count"] > 10 - else "" - ) - ) - except Exception as e: - return f"Error searching transactions: {str(e)}" - - -@tool -def ynab_category_spending(month: str = "") -> str: - """Get YNAB spending breakdown by category, including overspent categories. - - Args: - month: YYYY-MM (default: current month) - """ - if not ynab_enabled: - return "YNAB integration is not configured. Please set YNAB_ACCESS_TOKEN environment variable." - - try: - result = ynab_service.get_category_spending(month=month or None) - - summary = ( - f"Budget spending for {result['month']}:\n" - f"Total budgeted: ${result['total_budgeted']:.2f}\n" - f"Total spent: ${result['total_spent']:.2f}\n" - f"Total available: ${result['total_available']:.2f}\n" - ) - - if result["overspent_categories"]: - summary += ( - f"\nOverspent categories ({len(result['overspent_categories'])}):\n" - ) - for cat in result["overspent_categories"][:5]: - summary += f"- {cat['name']}: Budgeted ${cat['budgeted']:.2f}, Spent ${cat['spent']:.2f}, Over by ${cat['overspent_by']:.2f}\n" - - # All categories with spending activity - spending_cats = [c for c in result["categories"] if c["activity"] < 0] - if spending_cats: - summary += "\nSpending by category:\n" - for cat in spending_cats: - summary += f"- {cat['category']}: ${abs(cat['activity']):.2f} spent (budgeted: ${cat['budgeted']:.2f}, available: ${cat['available']:.2f})\n" - - return summary - except Exception as e: - return f"Error fetching category spending: {str(e)}" - - -@tool -def ynab_category_allocations(month: str = "") -> str: - """Get all YNAB category allocations grouped by group, with assigned and available amounts. - - Args: - month: YYYY-MM (default: current month) - """ - if not ynab_enabled: - return "YNAB integration is not configured. Please set YNAB_ACCESS_TOKEN environment variable." - - try: - result = ynab_service.get_category_allocations(month=month or None) - - lines = [ - f"Category allocations for {result['month']} " - f"(to be budgeted: ${result['to_be_budgeted']:.2f}):\n" - ] - - for group_name in result["group_order"]: - categories = result["groups"].get(group_name, []) - group_assigned = sum(c["assigned"] for c in categories) - group_available = sum(c["available"] for c in categories) - if group_assigned == 0 and group_available == 0: - continue - - lines.append( - f"{group_name} [assigned ${group_assigned:.2f} / available ${group_available:.2f}]" - ) - for cat in categories: - if cat["assigned"] == 0 and cat["available"] == 0: - continue - avail = cat["available"] - avail_str = f"+${avail:.2f}" if avail >= 0 else f"-${abs(avail):.2f}" - lines.append( - f" {cat['name']}: assigned ${cat['assigned']:.2f}, available {avail_str}" - ) - lines.append("") - - lines.append( - f"Totals — assigned: ${result['total_assigned']:.2f}, available: ${result['total_available']:.2f}" - ) - return "\n".join(lines) - except Exception as e: - return f"Error fetching category allocations: {str(e)}" - - -@tool -def ynab_insights(months_back: int = 3) -> str: - """Get YNAB spending trends, frequently overspent categories, and budget recommendations. - - Args: - months_back: Months to analyze (default 3, max 6) - """ - if not ynab_enabled: - return "YNAB integration is not configured. Please set YNAB_ACCESS_TOKEN environment variable." - - try: - # Limit to reasonable range - months_back = min(max(1, months_back), 6) - result = ynab_service.get_spending_insights(months_back=months_back) - - if "error" in result: - return result["error"] - - summary = ( - f"Spending insights for the last {months_back} months:\n\n" - f"Average monthly spending: ${result['average_monthly_spending']:.2f}\n" - f"Current month spending: ${result['current_month_spending']:.2f}\n" - f"Spending trend: {result['spending_trend']}\n" - ) - - if result["frequently_overspent_categories"]: - summary += "\nFrequently overspent categories:\n" - for cat in result["frequently_overspent_categories"][:5]: - summary += f"- {cat['category']}: overspent in {cat['months_overspent']} of {months_back} months\n" - - if result["recommendations"]: - summary += "\nRecommendations:\n" - for rec in result["recommendations"]: - summary += f"- {rec}\n" - - return summary - except Exception as e: - return f"Error generating insights: {str(e)}" - - @tool async def mealie_todays_meals() -> str: """Get today's planned meals from Mealie.""" @@ -978,16 +783,8 @@ if simbakit_enabled: simbakit_drinking_activity, ] ) -if ynab_enabled: - tools.extend( - [ - ynab_budget_summary, - ynab_search_transactions, - ynab_category_spending, - ynab_category_allocations, - ynab_insights, - ] - ) +if ynab_configured(): + tools.extend(YNAB_TOOLS) if mealie_enabled: tools.extend( [ diff --git a/blueprints/conversation/prompts.py b/blueprints/conversation/prompts.py index c83d92e..570f55d 100644 --- a/blueprints/conversation/prompts.py +++ b/blueprints/conversation/prompts.py @@ -34,10 +34,12 @@ IMPORTANT: When users ask factual questions about Simba's health, medical histor BUDGET & FINANCE (YNAB Integration): You have access to Ryan's budget data through YNAB (You Need A Budget). When users ask about financial matters, use the appropriate YNAB tools: -- Use ynab_budget_summary for overall budget health and status questions +- Use ynab_budget_summary for a month's overview (Ready to Assign, income, spent, overspending, net worth) +- Use ynab_account_balances for account and credit card balances - Use ynab_search_transactions to find specific purchases or spending at particular stores -- Use ynab_category_spending to analyze spending by category for a month -- Use ynab_insights to provide spending trends, patterns, and recommendations +- Use ynab_spending_report for totals by category, payee, or account, and group_by='month' for trends +- Use ynab_category_budget for what's assigned/available per category in a month +- Use ynab_recent_changes for what was added, edited, or re-assigned recently Always use these tools when asked about budgets, spending, transactions, or financial health. MEAL PLANNING (Mealie Integration): diff --git a/blueprints/ynab/__init__.py b/blueprints/ynab/__init__.py new file mode 100644 index 0000000..c25dbc9 --- /dev/null +++ b/blueprints/ynab/__init__.py @@ -0,0 +1 @@ +"""YNAB budget mirror: Postgres models, delta sync, and agent tools.""" diff --git a/blueprints/ynab/__main__.py b/blueprints/ynab/__main__.py new file mode 100644 index 0000000..a76b551 --- /dev/null +++ b/blueprints/ynab/__main__.py @@ -0,0 +1,22 @@ +"""Run one YNAB sync: ``python -m blueprints.ynab [--full]``.""" + +import asyncio +import sys + +from tortoise import Tortoise + +from config.db import TORTOISE_CONFIG + +from .sync import sync_ynab + + +async def main(full: bool) -> None: + await Tortoise.init(config=TORTOISE_CONFIG) + try: + print(await sync_ynab(full=full)) + finally: + await Tortoise.close_connections() + + +if __name__ == "__main__": + asyncio.run(main(full="--full" in sys.argv)) diff --git a/blueprints/ynab/models.py b/blueprints/ynab/models.py new file mode 100644 index 0000000..80f295c --- /dev/null +++ b/blueprints/ynab/models.py @@ -0,0 +1,173 @@ +"""Local Postgres mirror of the YNAB budget. + +Rows are keyed by YNAB's own ids and kept current by ``blueprints.ynab.sync`` +using YNAB delta requests. Money is stored in dollars with 3 decimal places, +which is exactly YNAB's milliunits. Deleted YNAB entities are kept with +``deleted=True`` rather than removed. +""" + +from tortoise import fields +from tortoise.models import Model + + +def money(**kwargs): + return fields.DecimalField(max_digits=15, decimal_places=3, **kwargs) + + +class YnabAccount(Model): + id = fields.CharField(max_length=36, primary_key=True) + name = fields.CharField(max_length=255) + type = fields.CharField(max_length=50) + on_budget = fields.BooleanField() + closed = fields.BooleanField() + note = fields.TextField(null=True) + balance = money() + cleared_balance = money() + uncleared_balance = money() + last_reconciled_at = fields.DatetimeField(null=True) + deleted = fields.BooleanField(default=False) + + class Meta: + table = "ynab_accounts" + + +class YnabPayee(Model): + id = fields.CharField(max_length=36, primary_key=True) + name = fields.CharField(max_length=500) + transfer_account_id = fields.CharField(max_length=36, null=True) + deleted = fields.BooleanField(default=False) + + class Meta: + table = "ynab_payees" + + +class YnabCategoryGroup(Model): + id = fields.CharField(max_length=36, primary_key=True) + name = fields.CharField(max_length=255) + hidden = fields.BooleanField() + internal = fields.BooleanField(default=False) + deleted = fields.BooleanField(default=False) + + class Meta: + table = "ynab_category_groups" + + +class YnabCategory(Model): + id = fields.CharField(max_length=36, primary_key=True) + category_group_id = fields.CharField(max_length=36, db_index=True) + name = fields.CharField(max_length=255) + hidden = fields.BooleanField() + internal = fields.BooleanField(default=False) + note = fields.TextField(null=True) + goal_type = fields.CharField(max_length=20, null=True) + goal_target = money(null=True) + goal_target_month = fields.DateField(null=True) + deleted = fields.BooleanField(default=False) + + class Meta: + table = "ynab_categories" + + +class YnabMonth(Model): + # ISO month start, e.g. "2026-10-01" + id = fields.CharField(max_length=10, primary_key=True) + month = fields.DateField(db_index=True) + note = fields.TextField(null=True) + income = money() + budgeted = money() + activity = money() + to_be_budgeted = money() + age_of_money = fields.IntField(null=True) + deleted = fields.BooleanField(default=False) + + class Meta: + table = "ynab_months" + + +class YnabMonthCategory(Model): + """Per-month category amounts. ``budgeted`` is what YNAB calls "Assigned".""" + + # ":" + id = fields.CharField(max_length=47, primary_key=True) + month = fields.DateField(db_index=True) + category_id = fields.CharField(max_length=36, db_index=True) + budgeted = money() + activity = money() + balance = money() + deleted = fields.BooleanField(default=False) + + class Meta: + table = "ynab_month_categories" + + +class YnabTransaction(Model): + id = fields.CharField(max_length=64, primary_key=True) + date = fields.DateField(db_index=True) + amount = money() + memo = fields.TextField(null=True) + cleared = fields.CharField(max_length=20) + approved = fields.BooleanField() + flag_color = fields.CharField(max_length=20, null=True) + account_id = fields.CharField(max_length=36, db_index=True) + payee_id = fields.CharField(max_length=36, null=True, db_index=True) + category_id = fields.CharField(max_length=36, null=True, db_index=True) + transfer_account_id = fields.CharField(max_length=36, null=True) + import_payee_name = fields.CharField(max_length=500, null=True) + # Denormalized from payees/categories/accounts after each sync for easy querying + account_name = fields.CharField(max_length=255, null=True) + payee_name = fields.CharField(max_length=500, null=True) + category_name = fields.CharField(max_length=255, null=True) + deleted = fields.BooleanField(default=False) + + class Meta: + table = "ynab_transactions" + + +class YnabSubtransaction(Model): + """Lines of a split transaction; the parent's category is "Split".""" + + id = fields.CharField(max_length=64, primary_key=True) + transaction_id = fields.CharField(max_length=64, db_index=True) + amount = money() + memo = fields.TextField(null=True) + payee_id = fields.CharField(max_length=36, null=True) + category_id = fields.CharField(max_length=36, null=True, db_index=True) + transfer_account_id = fields.CharField(max_length=36, null=True) + deleted = fields.BooleanField(default=False) + + class Meta: + table = "ynab_subtransactions" + + +class YnabChange(Model): + """Append-only log of what each sync changed, built by diffing against the mirror. + + YNAB's API has no history, so this is the only record of edits (e.g. an + assignment going from $400 to $500). It starts at the first sync. + """ + + id = fields.BigIntField(primary_key=True) + synced_at = fields.DatetimeField(auto_now_add=True, db_index=True) + server_knowledge = fields.BigIntField() + entity_type = fields.CharField(max_length=30, db_index=True) + entity_id = fields.CharField(max_length=64, db_index=True) + # created / updated / deleted + action = fields.CharField(max_length=10) + label = fields.CharField(max_length=500) + # {field: [old, new]}; for "created", old is null + changes = fields.JSONField() + + class Meta: + table = "ynab_changes" + + +class YnabSyncState(Model): + budget_id = fields.CharField(max_length=64, primary_key=True) + budget_name = fields.CharField(max_length=255, null=True) + server_knowledge = fields.BigIntField(null=True) + last_synced_at = fields.DatetimeField(null=True) + last_full_sync_at = fields.DatetimeField(null=True) + last_error = fields.TextField(null=True) + + class Meta: + table = "ynab_sync_state" diff --git a/blueprints/ynab/sql_tool.py b/blueprints/ynab/sql_tool.py new file mode 100644 index 0000000..58f49cc --- /dev/null +++ b/blueprints/ynab/sql_tool.py @@ -0,0 +1,151 @@ +"""Admin-only read-only SQL over the YNAB mirror tables. + +Defense in depth: the query is parsed and must be a single SELECT touching +only ``ynab_*`` tables with no dangerous functions, then runs in a READ ONLY +transaction with a statement timeout and a row cap. +""" + +import re +from decimal import Decimal + +import sqlglot +from langchain_core.runnables import RunnableConfig +from langchain.tools import tool +from sqlglot import exp +from sqlglot.optimizer.scope import traverse_scope +from tortoise.transactions import in_transaction + +from . import models +from .sync import ensure_fresh + +MAX_ROWS = 200 +MAX_OUTPUT_CHARS = 12000 +STATEMENT_TIMEOUT = "10s" + +ALLOWED_TABLES = frozenset( + m._meta.db_table + for m in vars(models).values() + if isinstance(m, type) and issubclass(m, models.Model) and m is not models.Model +) + +FORBIDDEN_NODES = ( + exp.Insert, + exp.Update, + exp.Delete, + exp.Merge, + exp.Create, + exp.Drop, + exp.Alter, + exp.Command, + exp.Into, +) + +FORBIDDEN_FUNCTIONS = re.compile( + r"^(pg_|lo_|dblink|set_config|current_setting|" + r"(query|table|cursor|schema|database)_to_xml|txid_|nextval|setval)", + re.IGNORECASE, +) + + +def validate_sql(query: str) -> str | None: + """Return an error message if ``query`` isn't allowed, else None.""" + try: + statements = [s for s in sqlglot.parse(query, read="postgres") if s] + except sqlglot.errors.ParseError as e: + return f"Could not parse SQL: {e}" + if len(statements) != 1: + return "Exactly one SQL statement is allowed." + root = statements[0] + if not isinstance(root, exp.Query): + return "Only SELECT queries are allowed." + if root.find(*FORBIDDEN_NODES) or any( + s.args.get("locks") for s in root.find_all(exp.Select) + ): + return "Only read-only SELECT queries are allowed." + + # A name matching a CTE isn't necessarily a CTE reference (e.g. inside the + # CTE's own body, or outside the scope that defines it), so resolve each + # FROM source through sqlglot's scopes. Any table node that isn't a CTE + # name at all is checked too, in case scope analysis misses a construct. + try: + scopes = traverse_scope(root) + except Exception as e: + return f"Could not analyze SQL: {e}" + cte_names = {cte.alias_or_name for cte in root.find_all(exp.CTE)} + tables = [ + source + for scope in scopes + for source in scope.sources.values() + if isinstance(source, exp.Table) + ] + tables += [t for t in root.find_all(exp.Table) if t.name not in cte_names] + for table in tables: + if isinstance(table.this, exp.Func): # e.g. generate_series(...) + continue + if table.catalog or table.db not in ("", "public"): + return f"Schema-qualified table '{table.sql()}' is not allowed." + if table.name not in ALLOWED_TABLES: + return ( + f"Table '{table.name}' is not allowed. " + f"Available: {', '.join(sorted(ALLOWED_TABLES))}" + ) + + for func in root.find_all(exp.Func): + name = func.name if isinstance(func, exp.Anonymous) else func.sql_name() + if FORBIDDEN_FUNCTIONS.match(name or ""): + return f"Function '{name}' is not allowed." + return None + + +def _format_value(value) -> str: + if value is None: + return "" + if isinstance(value, Decimal): + value = ( + format(value.normalize(), "f") if value == value.to_integral() else value + ) + return str(value).replace("\t", " ").replace("\n", " ") + + +def format_rows(rows: list[dict]) -> str: + if not rows: + return "Query returned no rows." + header = list(rows[0]) + lines = ["\t".join(header)] + lines += ["\t".join(_format_value(r[c]) for c in header) for r in rows] + text = "\n".join(lines) + if len(text) > MAX_OUTPUT_CHARS: + text = text[:MAX_OUTPUT_CHARS] + "\n...(output truncated)" + if len(rows) == MAX_ROWS: + text += f"\n(capped at {MAX_ROWS} rows; aggregate or add LIMIT)" + return text + + +@tool +async def ynab_sql(query: str, config: RunnableConfig) -> str: + """Run a read-only Postgres SELECT over the YNAB mirror for questions the other YNAB tools can't answer. Tables: ynab_transactions (date, amount, payee_name, category_name, account_name, memo, transfer_account_id, deleted), ynab_subtransactions (split lines), ynab_accounts, ynab_categories, ynab_category_groups, ynab_payees, ynab_months, ynab_month_categories (month, category_id, budgeted=assigned, activity, balance=available), ynab_changes (edit history: synced_at, entity_type, action, label, changes JSON). Amounts are dollars; outflows negative. Filter NOT deleted. + + Args: + query: A single PostgreSQL SELECT statement + """ + if not (config or {}).get("configurable", {}).get("is_admin"): + return "YNAB SQL access is restricted to admin users." + error = validate_sql(query) + if error: + return error + if await ensure_fresh() is None: + return "YNAB data is unavailable: the budget has not been synced yet." + + query = query.strip().rstrip(";") + try: + async with in_transaction() as conn: + await conn.execute_query("SET TRANSACTION READ ONLY") + await conn.execute_query( + f"SET LOCAL statement_timeout = '{STATEMENT_TIMEOUT}'" + ) + rows = await conn.execute_query_dict( + f"SELECT * FROM ({query}) AS q LIMIT {MAX_ROWS}" + ) + except Exception as e: + return f"SQL error: {e}" + return format_rows(rows) diff --git a/blueprints/ynab/sync.py b/blueprints/ynab/sync.py new file mode 100644 index 0000000..88bf15f --- /dev/null +++ b/blueprints/ynab/sync.py @@ -0,0 +1,512 @@ +"""Sync the YNAB budget into the local Postgres mirror. + +YNAB has no event feed, but every budget response carries a +``server_knowledge`` counter; passing it back as ``last_knowledge_of_server`` +returns only entities changed since then. The first sync downloads the whole +budget, later syncs are deltas, and a full resync runs once a day as a safety +net (deltas can't express things like month rollovers). + +Each sync diffs incoming entities against the mirror and appends what changed +to ``ynab_changes``, which is the only history of edits we get. +""" + +import asyncio +import logging +import os +from dataclasses import dataclass +from datetime import date, datetime, timedelta, timezone +from decimal import Decimal +from typing import Any, Callable + +import httpx +from tortoise.models import Model +from tortoise.transactions import in_transaction + +from .models import ( + YnabAccount, + YnabCategory, + YnabCategoryGroup, + YnabChange, + YnabMonth, + YnabMonthCategory, + YnabPayee, + YnabSubtransaction, + YnabSyncState, + YnabTransaction, +) + +logger = logging.getLogger(__name__) + +YNAB_API_URL = "https://api.ynab.com/v1" +FULL_RESYNC_INTERVAL = timedelta(hours=24) +# Cap on rows passed to a single ``id IN (...)`` lookup +LOOKUP_CHUNK = 2000 + +_sync_lock = asyncio.Lock() + + +def ynab_configured() -> bool: + return bool(os.getenv("YNAB_ACCESS_TOKEN")) + + +def _budget_id() -> str: + return os.getenv("YNAB_BUDGET_ID") or "last-used" + + +def _money(milliunits: int | None) -> Decimal | None: + return None if milliunits is None else Decimal(milliunits).scaleb(-3) + + +def _date(value: str | None) -> date | None: + return date.fromisoformat(value) if value else None + + +def _datetime(value: str | None) -> datetime | None: + if not value: + return None + return datetime.fromisoformat(value.replace("Z", "+00:00")) + + +def _norm(value: Any) -> Any: + """Normalize values so DB rows and freshly mapped rows compare equal.""" + if isinstance(value, datetime) and value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value + + +def _jsonable(value: Any) -> Any: + if isinstance(value, Decimal): + return float(value) + if isinstance(value, (date, datetime)): + return value.isoformat() + return value + + +def _map_account(a: dict) -> dict: + return { + "id": a["id"], + "name": a["name"], + "type": a["type"], + "on_budget": a["on_budget"], + "closed": a["closed"], + "note": a.get("note"), + "balance": _money(a["balance"]), + "cleared_balance": _money(a["cleared_balance"]), + "uncleared_balance": _money(a["uncleared_balance"]), + "last_reconciled_at": _datetime(a.get("last_reconciled_at")), + "deleted": a["deleted"], + } + + +def _map_payee(p: dict) -> dict: + return { + "id": p["id"], + "name": p["name"], + "transfer_account_id": p.get("transfer_account_id"), + "deleted": p["deleted"], + } + + +def _map_category_group(g: dict) -> dict: + return { + "id": g["id"], + "name": g["name"], + "hidden": g["hidden"], + "internal": g.get("internal") or False, + "deleted": g["deleted"], + } + + +def _map_category(c: dict) -> dict: + return { + "id": c["id"], + "category_group_id": c["category_group_id"], + "name": c["name"], + "hidden": c["hidden"], + "internal": c.get("internal") or False, + "note": c.get("note"), + "goal_type": c.get("goal_type"), + "goal_target": _money(c.get("goal_target")), + "goal_target_month": _date(c.get("goal_target_month")), + "deleted": c["deleted"], + } + + +def _map_month(m: dict) -> dict: + return { + "id": m["month"], + "month": _date(m["month"]), + "note": m.get("note"), + "income": _money(m["income"]), + "budgeted": _money(m["budgeted"]), + "activity": _money(m["activity"]), + "to_be_budgeted": _money(m["to_be_budgeted"]), + "age_of_money": m.get("age_of_money"), + "deleted": m["deleted"], + } + + +def _map_month_category(c: dict) -> dict: + return { + "id": f"{c['month']}:{c['id']}", + "month": _date(c["month"]), + "category_id": c["id"], + "budgeted": _money(c["budgeted"]), + "activity": _money(c["activity"]), + "balance": _money(c["balance"]), + "deleted": c["deleted"], + } + + +def _map_transaction(t: dict) -> dict: + return { + "id": t["id"], + "date": _date(t["date"]), + "amount": _money(t["amount"]), + "memo": t.get("memo"), + "cleared": t["cleared"], + "approved": t["approved"], + "flag_color": t.get("flag_color"), + "account_id": t["account_id"], + "payee_id": t.get("payee_id"), + "category_id": t.get("category_id"), + "transfer_account_id": t.get("transfer_account_id"), + "import_payee_name": t.get("import_payee_name"), + "deleted": t["deleted"], + } + + +def _map_subtransaction(s: dict) -> dict: + return { + "id": s["id"], + "transaction_id": s["transaction_id"], + "amount": _money(s["amount"]), + "memo": s.get("memo"), + "payee_id": s.get("payee_id"), + "category_id": s.get("category_id"), + "transfer_account_id": s.get("transfer_account_id"), + "deleted": s["deleted"], + } + + +@dataclass +class EntitySpec: + entity_type: str + model: type[Model] + mapper: Callable[[dict], dict] + # Fields worth recording in the change log. Derived fields (balances, + # activity) change with every transaction and would drown out real edits. + logged_fields: tuple[str, ...] + + +def _fields(model: type[Model], exclude: tuple[str, ...] = ()) -> tuple[str, ...]: + return tuple(f for f in model._meta.fields_map if f != "id" and f not in exclude) + + +SPECS = { + "accounts": EntitySpec( + "account", + YnabAccount, + _map_account, + ("name", "type", "on_budget", "closed", "note", "deleted"), + ), + "payees": EntitySpec("payee", YnabPayee, _map_payee, ("name", "deleted")), + "category_groups": EntitySpec( + "category_group", + YnabCategoryGroup, + _map_category_group, + ("name", "hidden", "deleted"), + ), + "categories": EntitySpec( + "category", YnabCategory, _map_category, _fields(YnabCategory) + ), + "months": EntitySpec("month", YnabMonth, _map_month, ("note",)), + "month_categories": EntitySpec( + "month_category", YnabMonthCategory, _map_month_category, ("budgeted",) + ), + "transactions": EntitySpec( + "transaction", + YnabTransaction, + _map_transaction, + _fields( + YnabTransaction, exclude=("account_name", "payee_name", "category_name") + ), + ), + "subtransactions": EntitySpec( + "subtransaction", + YnabSubtransaction, + _map_subtransaction, + _fields(YnabSubtransaction), + ), +} + + +def _budget_entities(budget: dict) -> dict[str, list[dict]]: + """Split a budget payload into raw entity lists keyed like ``SPECS``.""" + month_categories = [ + {**c, "month": m["month"]} + for m in budget.get("months") or [] + for c in m.get("categories") or [] + ] + return { + "accounts": budget.get("accounts") or [], + "payees": budget.get("payees") or [], + "category_groups": budget.get("category_groups") or [], + "categories": budget.get("categories") or [], + "months": budget.get("months") or [], + "month_categories": month_categories, + "transactions": budget.get("transactions") or [], + "subtransactions": budget.get("subtransactions") or [], + } + + +class _Labeler: + """Builds human-readable change log labels from ids.""" + + def __init__(self, names: dict[str, str]): + self.names = names + + def __call__(self, entity_type: str, row: dict) -> str: + n = self.names.get + if entity_type == "transaction": + payee = n(row.get("payee_id") or "") or row.get("import_payee_name") or "" + return f"{row['date']} {payee} {row['amount']:+.2f}".strip() + if entity_type == "subtransaction": + category = n(row.get("category_id") or "", "") + return f"split line {category} {row['amount']:+.2f}".strip() + if entity_type == "month_category": + return f"{n(row['category_id'], row['category_id'])} {row['month']:%Y-%m}" + if entity_type == "month": + return row["id"][:7] + return row.get("name") or row["id"] + + +async def _load_names(conn) -> dict[str, str]: + names: dict[str, str] = {} + for model in (YnabAccount, YnabPayee, YnabCategory, YnabCategoryGroup): + for row in await model.all().using_db(conn).values("id", "name"): + names[row["id"]] = row["name"] + return names + + +async def _apply( + spec: EntitySpec, + raw_rows: list[dict], + conn, + server_knowledge: int, + log_changes: bool, + labeler: _Labeler, +) -> int: + """Upsert rows that differ from the mirror; return how many changed.""" + if not raw_rows: + return 0 + + incoming = {row["id"]: row for row in map(spec.mapper, raw_rows)} + # Only YNAB-sourced columns; denormalized names are filled in afterwards + sample = next(iter(incoming.values())) + field_names = tuple(f for f in _fields(spec.model) if f in sample) + + ids = list(incoming) + existing: dict[str, dict] = {} + for i in range(0, len(ids), LOOKUP_CHUNK): + rows = ( + await spec.model.filter(id__in=ids[i : i + LOOKUP_CHUNK]) + .using_db(conn) + .values("id", *field_names) + ) + existing.update({row["id"]: row for row in rows}) + + to_write: list[Model] = [] + changes: list[YnabChange] = [] + for entity_id, new in incoming.items(): + old = existing.get(entity_id) + if old is not None and all(_norm(old[f]) == _norm(new[f]) for f in field_names): + continue + to_write.append(spec.model(**new)) + + if not log_changes or not spec.logged_fields: + continue + if old is None: + # Created and deleted between two syncs: nothing worth logging + if new.get("deleted"): + continue + action = "created" + diff = { + f: [None, _jsonable(new[f])] + for f in spec.logged_fields + if new[f] is not None + } + else: + diff = { + f: [_jsonable(old[f]), _jsonable(new[f])] + for f in spec.logged_fields + if _norm(old[f]) != _norm(new[f]) + } + if not diff: + continue + action = "deleted" if diff.get("deleted") == [False, True] else "updated" + changes.append( + YnabChange( + server_knowledge=server_knowledge, + entity_type=spec.entity_type, + entity_id=entity_id, + action=action, + label=labeler(spec.entity_type, new)[:500], + changes=diff, + ) + ) + + if to_write: + await spec.model.bulk_create( + to_write, + batch_size=500, + on_conflict=["id"], + update_fields=list(field_names), + using_db=conn, + ) + if changes: + await YnabChange.bulk_create(changes, batch_size=500, using_db=conn) + return len(to_write) + + +DENORMALIZE_SQL = ( + """UPDATE ynab_transactions AS t SET payee_name = p.name FROM ynab_payees p + WHERE t.payee_id = p.id AND t.payee_name IS DISTINCT FROM p.name""", + """UPDATE ynab_transactions AS t SET category_name = c.name FROM ynab_categories c + WHERE t.category_id = c.id AND t.category_name IS DISTINCT FROM c.name""", + """UPDATE ynab_transactions AS t SET account_name = a.name FROM ynab_accounts a + WHERE t.account_id = a.id AND t.account_name IS DISTINCT FROM a.name""", + """UPDATE ynab_transactions SET payee_name = NULL + WHERE payee_id IS NULL AND payee_name IS NOT NULL""", + """UPDATE ynab_transactions SET category_name = NULL + WHERE category_id IS NULL AND category_name IS NOT NULL""", +) + + +async def apply_budget( + budget: dict, server_knowledge: int, conn, log_changes: bool +) -> dict[str, int]: + """Write a (full or delta) budget payload into the mirror.""" + entities = _budget_entities(budget) + counts: dict[str, int] = {} + + # Reference data first so transaction labels can resolve names + labeler = _Labeler({}) + for key in ("accounts", "payees", "category_groups", "categories"): + counts[key] = await _apply( + SPECS[key], entities[key], conn, server_knowledge, log_changes, labeler + ) + labeler.names = await _load_names(conn) + for key in ("months", "month_categories", "transactions", "subtransactions"): + counts[key] = await _apply( + SPECS[key], entities[key], conn, server_knowledge, log_changes, labeler + ) + + if any(counts.values()): + for sql in DENORMALIZE_SQL: + await conn.execute_query(sql) + return counts + + +async def fetch_budget(last_knowledge: int | None) -> dict: + """GET the budget from YNAB, as a delta when ``last_knowledge`` is set.""" + params = {} + if last_knowledge is not None: + params["last_knowledge_of_server"] = last_knowledge + async with httpx.AsyncClient(timeout=httpx.Timeout(60.0, connect=10.0)) as client: + response = await client.get( + f"{YNAB_API_URL}/budgets/{_budget_id()}", + params=params, + headers={"Authorization": f"Bearer {os.environ['YNAB_ACCESS_TOKEN']}"}, + ) + response.raise_for_status() + return response.json()["data"] + + +async def sync_ynab(full: bool = False) -> dict: + """Sync YNAB into Postgres. Returns per-entity counts of changed rows. + + Safe to call concurrently: an in-process lock plus a Postgres advisory + lock serialize syncs across both blue/green containers. + """ + budget_id = _budget_id() + async with _sync_lock: + try: + async with in_transaction() as conn: + await conn.execute_query( + "SELECT pg_advisory_xact_lock(hashtext('ynab_sync'))" + ) + state, _ = await YnabSyncState.get_or_create( + budget_id=budget_id, using_db=conn + ) + first_sync = state.server_knowledge is None + now = datetime.now(timezone.utc) + full = ( + full + or first_sync + or ( + state.last_full_sync_at is None + or now - _norm(state.last_full_sync_at) > FULL_RESYNC_INTERVAL + ) + ) + + data = await fetch_budget(None if full else state.server_knowledge) + knowledge = data["server_knowledge"] + counts = await apply_budget( + data["budget"], knowledge, conn, log_changes=not first_sync + ) + + state.budget_name = data["budget"].get("name") or state.budget_name + state.server_knowledge = knowledge + state.last_synced_at = now + if full: + state.last_full_sync_at = now + state.last_error = None + await state.save(using_db=conn) + except Exception as e: + await YnabSyncState.filter(budget_id=budget_id).update( + last_error=f"{type(e).__name__}: {e}"[:2000] + ) + raise + + changed = {k: v for k, v in counts.items() if v} + if changed: + logger.info(f"YNAB {'full' if full else 'delta'} sync @{knowledge}: {changed}") + return {"full": full, "server_knowledge": knowledge, "changed": changed} + + +async def ensure_fresh(max_age_seconds: int = 120) -> YnabSyncState | None: + """Sync if the mirror is older than ``max_age_seconds``. + + Used by agent tools before answering. A failed sync is logged and the + (possibly stale) mirror is used anyway; returns None only if the mirror + has never been populated. + """ + state = await YnabSyncState.get_or_none(budget_id=_budget_id()) + stale = ( + state is None + or state.last_synced_at is None + or datetime.now(timezone.utc) - _norm(state.last_synced_at) + > timedelta(seconds=max_age_seconds) + ) + if stale: + try: + await sync_ynab() + except Exception: + logger.exception("YNAB sync failed; answering from existing mirror") + state = await YnabSyncState.get_or_none(budget_id=_budget_id()) + if state is None or state.server_knowledge is None: + return None + return state + + +async def ynab_sync_loop() -> None: + """Background task: delta-sync YNAB every ``YNAB_SYNC_INTERVAL`` seconds.""" + interval = int(os.getenv("YNAB_SYNC_INTERVAL", "900")) + logger.info(f"YNAB sync loop started (interval={interval}s)") + while True: + try: + await sync_ynab() + except Exception: + logger.exception("YNAB sync error") + await asyncio.sleep(interval) diff --git a/blueprints/ynab/tools.py b/blueprints/ynab/tools.py new file mode 100644 index 0000000..5499aed --- /dev/null +++ b/blueprints/ynab/tools.py @@ -0,0 +1,447 @@ +"""LangChain agent tools that answer YNAB questions from the local mirror. + +Every tool calls ``ensure_fresh`` first, which runs a cheap delta sync when +the mirror is more than a couple of minutes old. +""" + +import json +from datetime import date, datetime, timedelta +from decimal import Decimal +from zoneinfo import ZoneInfo + +from langchain.tools import tool +from tortoise import connections + +from .sync import ensure_fresh + +NOT_SYNCED = "YNAB data is unavailable: the budget has not been synced yet." + +# Every transaction line, with split transactions expanded into their +# subtransactions so category/payee filters see the real allocation. +LINES_CTE = """ +WITH lines AS ( + SELECT t.id, t.date, t.amount, t.memo, t.cleared, t.approved, + t.account_id, t.account_name, t.payee_name, t.category_id, + t.category_name, t.transfer_account_id + FROM ynab_transactions t + WHERE NOT t.deleted AND NOT EXISTS ( + SELECT 1 FROM ynab_subtransactions s + WHERE s.transaction_id = t.id AND NOT s.deleted + ) + UNION ALL + SELECT s.id, t.date, s.amount, COALESCE(s.memo, t.memo), t.cleared, t.approved, + t.account_id, t.account_name, COALESCE(sp.name, t.payee_name), + s.category_id, sc.name, s.transfer_account_id + FROM ynab_subtransactions s + JOIN ynab_transactions t ON t.id = s.transaction_id + LEFT JOIN ynab_payees sp ON sp.id = s.payee_id + LEFT JOIN ynab_categories sc ON sc.id = s.category_id + WHERE NOT s.deleted AND NOT t.deleted +) +""" + +# Spending = outflows (net of refunds) on budget accounts, excluding transfers +# (credit card payments, moves to savings) and income. +SPENDING_FILTER = """ + a.on_budget AND l.transfer_account_id IS NULL + AND NOT COALESCE(c.internal, false) +""" + +GROUP_BY_EXPR = { + "category": "COALESCE(l.category_name, 'Uncategorized')", + "category_group": "COALESCE(g.name, 'Uncategorized')", + "payee": "COALESCE(l.payee_name, 'Unknown payee')", + "account": "l.account_name", + "month": "to_char(l.date, 'YYYY-MM')", +} + + +def _today() -> date: + return datetime.now(ZoneInfo("America/New_York")).date() + + +def _usd(value: Decimal | float | None) -> str: + value = Decimal(value or 0) + sign = "-" if value < 0 else "" + return f"{sign}${abs(value):,.2f}" + + +def _parse_date(value: str, default: date | None) -> date | None: + return date.fromisoformat(value) if value else default + + +def _month_start(month: str) -> date: + if not month: + return _today().replace(day=1) + return date.fromisoformat(month[:7] + "-01") + + +async def _query(sql: str, *params) -> list[dict]: + return await connections.get("default").execute_query_dict(sql, list(params)) + + +@tool +async def ynab_budget_summary(month: str = "") -> str: + """Get YNAB budget overview for a month: Ready to Assign, income, assigned, spent, age of money, overspent categories, and net worth. + + Args: + month: YYYY-MM (default: current month) + """ + state = await ensure_fresh() + if state is None: + return NOT_SYNCED + try: + start = _month_start(month) + except ValueError: + return "Month must be YYYY-MM." + + months = await _query("SELECT * FROM ynab_months WHERE month = $1", start) + if not months: + return f"No YNAB budget data for {start:%Y-%m}." + m = months[0] + + overspent = await _query( + """SELECT c.name, mc.balance FROM ynab_month_categories mc + JOIN ynab_categories c ON c.id = mc.category_id + WHERE mc.month = $1 AND mc.balance < 0 AND NOT mc.deleted + AND NOT c.deleted AND NOT c.internal + ORDER BY mc.balance""", + start, + ) + # YNAB's month "activity" nets in credit card payment categories, which + # go up whenever a card is used, so compute spending from transactions. + next_month = (start.replace(day=28) + timedelta(days=4)).replace(day=1) + spent = await _query( + f"""{LINES_CTE} + SELECT COALESCE(-SUM(l.amount), 0) AS spent FROM lines l + JOIN ynab_accounts a ON a.id = l.account_id + LEFT JOIN ynab_categories c ON c.id = l.category_id + WHERE {SPENDING_FILTER} AND l.date >= $1 AND l.date < $2""", + start, + next_month, + ) + accounts = await _query( + """SELECT on_budget, SUM(balance) AS total FROM ynab_accounts + WHERE NOT deleted AND NOT closed GROUP BY on_budget""" + ) + by_budget = {r["on_budget"]: r["total"] for r in accounts} + + lines = [ + f"YNAB budget '{state.budget_name}' for {start:%Y-%m}:", + f"Ready to Assign: {_usd(m['to_be_budgeted'])}", + f"Income: {_usd(m['income'])}", + f"Assigned: {_usd(m['budgeted'])}", + f"Spent: {_usd(spent[0]['spent'])}", + ] + if m["age_of_money"] is not None: + lines.append(f"Age of money: {m['age_of_money']} days") + if overspent: + lines.append( + "Overspent: " + + ", ".join(f"{r['name']} ({_usd(r['balance'])})" for r in overspent) + ) + else: + lines.append("No overspent categories.") + lines.append( + f"Net worth: {_usd(sum(by_budget.values(), Decimal(0)))} " + f"(budget accounts {_usd(by_budget.get(True))}, " + f"tracking accounts {_usd(by_budget.get(False))})" + ) + return "\n".join(lines) + + +@tool +async def ynab_account_balances(include_closed: bool = False) -> str: + """List YNAB accounts (checking, savings, credit cards, investments) with current balances. + + Args: + include_closed: Also list closed accounts + """ + if await ensure_fresh() is None: + return NOT_SYNCED + rows = await _query( + """SELECT name, type, on_budget, closed, balance, cleared_balance + FROM ynab_accounts WHERE NOT deleted AND (NOT closed OR $1) + ORDER BY on_budget DESC, type, name""", + include_closed, + ) + if not rows: + return "No YNAB accounts found." + lines = [] + for r in rows: + tags = [r["type"]] + ([] if r["on_budget"] else ["tracking"]) + if r["closed"]: + tags.append("closed") + line = f"- {r['name']} [{', '.join(tags)}]: {_usd(r['balance'])}" + if r["cleared_balance"] != r["balance"]: + line += f" (cleared {_usd(r['cleared_balance'])})" + lines.append(line) + total = sum((r["balance"] for r in rows if not r["closed"]), Decimal(0)) + return "\n".join(lines) + f"\n\nNet worth (open accounts): {_usd(total)}" + + +@tool +async def ynab_search_transactions( + start_date: str = "", + end_date: str = "", + payee: str = "", + category: str = "", + account: str = "", + memo: str = "", + min_amount: float = 0, + max_amount: float = 0, + limit: int = 25, +) -> str: + """Search YNAB transactions. Split transactions are expanded into their category lines. + + Args: + start_date: YYYY-MM-DD (default: 30 days ago if no other filter, else all time) + end_date: YYYY-MM-DD (default: today) + payee: Partial match on payee/store name + category: Partial match on category name + account: Partial match on account name + memo: Partial match on memo + min_amount: Minimum absolute dollar amount + max_amount: Maximum absolute dollar amount + limit: Max transactions to list (default 25, max 100) + """ + if await ensure_fresh() is None: + return NOT_SYNCED + + has_filter = any([payee, category, account, memo, min_amount, max_amount]) + try: + default_start = ( + None if has_filter else date.fromordinal(_today().toordinal() - 30) + ) + start = _parse_date(start_date, default_start) + end = _parse_date(end_date, _today()) + except ValueError: + return "Dates must be YYYY-MM-DD." + + conditions, params = ["l.date <= $1"], [end] + + def add(condition: str, value) -> None: + params.append(value) + conditions.append(condition.replace("?", f"${len(params)}")) + + if start: + add("l.date >= ?", start) + if payee: + add("l.payee_name ILIKE ?", f"%{payee}%") + if category: + add("l.category_name ILIKE ?", f"%{category}%") + if account: + add("l.account_name ILIKE ?", f"%{account}%") + if memo: + add("l.memo ILIKE ?", f"%{memo}%") + if min_amount: + add("abs(l.amount) >= ?", Decimal(str(min_amount))) + if max_amount: + add("abs(l.amount) <= ?", Decimal(str(max_amount))) + + where = " AND ".join(conditions) + totals = await _query( + f"""{LINES_CTE} + SELECT count(*) AS n, + COALESCE(SUM(amount) FILTER (WHERE amount < 0), 0) AS outflow, + COALESCE(SUM(amount) FILTER (WHERE amount > 0), 0) AS inflow + FROM lines l WHERE {where}""", + *params, + ) + count = totals[0]["n"] + range_text = f"{start or 'the beginning'} to {end}" + if not count: + return f"No transactions found from {range_text}." + + limit = min(max(1, limit), 100) + rows = await _query( + f"""{LINES_CTE} + SELECT * FROM lines l WHERE {where} + ORDER BY l.date DESC, abs(l.amount) DESC LIMIT {limit}""", + *params, + ) + lines = [ + f"Found {count} transactions from {range_text}. " + f"Outflows {_usd(totals[0]['outflow'])}, inflows {_usd(totals[0]['inflow'])}." + ] + for r in rows: + line = ( + f"- {r['date']}: {r['payee_name'] or 'Unknown payee'} {_usd(r['amount'])} " + f"({r['category_name'] or 'Uncategorized'}; {r['account_name']})" + ) + if r["memo"]: + line += f" — {r['memo']}" + if not r["approved"]: + line += " [unapproved]" + lines.append(line) + if count > len(rows): + lines.append(f"(Showing {len(rows)} most recent of {count})") + return "\n".join(lines) + + +@tool +async def ynab_spending_report( + group_by: str = "category", + start_date: str = "", + end_date: str = "", + category: str = "", + payee: str = "", + top: int = 15, +) -> str: + """Total YNAB spending grouped by category, category_group, payee, account, or month. Use group_by='month' for trends over time. + + Args: + group_by: category, category_group, payee, account, or month + start_date: YYYY-MM-DD (default: first of current month) + end_date: YYYY-MM-DD (default: today) + category: Partial match on category name + payee: Partial match on payee name + top: Max groups to list (default 15) + """ + if await ensure_fresh() is None: + return NOT_SYNCED + key = GROUP_BY_EXPR.get(group_by) + if key is None: + return f"group_by must be one of: {', '.join(GROUP_BY_EXPR)}." + try: + start = _parse_date(start_date, _today().replace(day=1)) + end = _parse_date(end_date, _today()) + except ValueError: + return "Dates must be YYYY-MM-DD." + + conditions, params = [SPENDING_FILTER, "l.date >= $1", "l.date <= $2"], [start, end] + if category: + params.append(f"%{category}%") + conditions.append(f"l.category_name ILIKE ${len(params)}") + if payee: + params.append(f"%{payee}%") + conditions.append(f"l.payee_name ILIKE ${len(params)}") + + order = "key" if group_by == "month" else "spent DESC" + rows = await _query( + f"""{LINES_CTE} + SELECT {key} AS key, -SUM(l.amount) AS spent, count(*) AS n + FROM lines l + JOIN ynab_accounts a ON a.id = l.account_id + LEFT JOIN ynab_categories c ON c.id = l.category_id + LEFT JOIN ynab_category_groups g ON g.id = c.category_group_id + WHERE {' AND '.join(conditions)} + GROUP BY 1 ORDER BY {order}""", + *params, + ) + if not rows: + return f"No spending found from {start} to {end}." + + total = sum((r["spent"] for r in rows), Decimal(0)) + top = max(1, top) + shown = rows if group_by == "month" else rows[:top] + lines = [f"Spending by {group_by} from {start} to {end}: total {_usd(total)}"] + lines += [f"- {r['key']}: {_usd(r['spent'])} ({r['n']} txns)" for r in shown] + if len(rows) > len(shown): + rest = sum((r["spent"] for r in rows[len(shown) :]), Decimal(0)) + lines.append(f"- {len(rows) - len(shown)} others: {_usd(rest)}") + if group_by == "month" and len(rows) > 1: + lines.append(f"Monthly average: {_usd(total / len(rows))}") + return "\n".join(lines) + + +@tool +async def ynab_category_budget(month: str = "", category_group: str = "") -> str: + """Get YNAB categories for a month grouped by category group: assigned, activity (negative = spent), available, and goal target. Flags overspending. + + Args: + month: YYYY-MM (default: current month) + category_group: Partial match to show only one group + """ + if await ensure_fresh() is None: + return NOT_SYNCED + try: + start = _month_start(month) + except ValueError: + return "Month must be YYYY-MM." + + rows = await _query( + """SELECT g.name AS group_name, c.name, c.goal_target, + mc.budgeted, mc.activity, mc.balance + FROM ynab_month_categories mc + JOIN ynab_categories c ON c.id = mc.category_id + JOIN ynab_category_groups g ON g.id = c.category_group_id + WHERE mc.month = $1 AND NOT mc.deleted + AND NOT c.deleted AND NOT c.hidden AND NOT c.internal + AND NOT g.deleted AND NOT g.hidden + AND g.name ILIKE $2 + ORDER BY g.name, c.name""", + start, + f"%{category_group}%", + ) + if not rows: + return f"No YNAB category data for {start:%Y-%m}." + + lines = [f"Categories for {start:%Y-%m}:"] + current = None + for r in rows: + if not (r["budgeted"] or r["activity"] or r["balance"]): + continue + if r["group_name"] != current: + current = r["group_name"] + lines.append(f"\n{current}") + line = ( + f" {r['name']}: assigned {_usd(r['budgeted'])}, " + f"activity {_usd(r['activity'])}, available {_usd(r['balance'])}" + ) + if r["goal_target"]: + line += f", goal {_usd(r['goal_target'])}" + if r["balance"] < 0: + line += " [OVERSPENT]" + lines.append(line) + return "\n".join(lines) + + +@tool +async def ynab_recent_changes(days: int = 7, entity_type: str = "") -> str: + """Show what changed in YNAB recently: new/edited/deleted transactions, assignment changes, renamed categories or payees. + + Args: + days: Look back this many days (default 7) + entity_type: Optional filter: transaction, month_category (assignments), category, payee, account + """ + if await ensure_fresh() is None: + return NOT_SYNCED + rows = await _query( + """SELECT synced_at, entity_type, action, label, changes FROM ynab_changes + WHERE synced_at >= now() - make_interval(days => $1) + AND ($2 = '' OR entity_type = $2) + ORDER BY id DESC LIMIT 50""", + max(1, days), + entity_type, + ) + if not rows: + return f"No YNAB changes recorded in the last {days} days." + + lines = [] + for r in rows: + when = r["synced_at"].astimezone(ZoneInfo("America/New_York")) + changes = r["changes"] + if isinstance(changes, str): + changes = json.loads(changes) + if r["action"] == "updated": + detail = "; ".join( + f"{field}: {old} → {new}" for field, (old, new) in changes.items() + ) + else: + detail = "" + entity = r["entity_type"].replace("month_category", "assignment") + line = f"- {when:%Y-%m-%d %H:%M} {r['action']} {entity} '{r['label']}'" + lines.append(line + (f": {detail}" if detail else "")) + note = " (newest 50)" if len(rows) == 50 else "" + return f"YNAB changes in the last {days} days{note}:\n" + "\n".join(lines) + + +YNAB_TOOLS = [ + ynab_budget_summary, + ynab_account_balances, + ynab_search_transactions, + ynab_spending_report, + ynab_category_budget, + ynab_recent_changes, +] diff --git a/config/db.py b/config/db.py index 73c2be0..63dc614 100644 --- a/config/db.py +++ b/config/db.py @@ -19,6 +19,7 @@ TORTOISE_CONFIG = { "blueprints.scheduled_messages.models", "blueprints.cron.models", "blueprints.beli.models", + "blueprints.ynab.models", "aerich.models", ], "default_connection": "default", diff --git a/docker-compose.yml b/docker-compose.yml index d18f5bc..ff46df1 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -24,6 +24,7 @@ x-raggr-common: &raggr-common - TAVILY_API_KEY=${TAVILIY_API_KEY} - YNAB_ACCESS_TOKEN=${YNAB_ACCESS_TOKEN} - YNAB_BUDGET_ID=${YNAB_BUDGET_ID} + - YNAB_SYNC_INTERVAL=${YNAB_SYNC_INTERVAL:-900} - MEALIE_BASE_URL=${MEALIE_BASE_URL} - MEALIE_API_TOKEN=${MEALIE_API_TOKEN} - TWILIO_ACCOUNT_SID=${TWILIO_ACCOUNT_SID} diff --git a/docs/TASKS.md b/docs/TASKS.md index 9d4c370..0729ac8 100644 --- a/docs/TASKS.md +++ b/docs/TASKS.md @@ -36,11 +36,11 @@ ## Backlog -- [ ] YNAB API client module -- [ ] YNAB data models (Budget, Transaction, Category, Account) -- [ ] Database schema for financial data -- [ ] YNAB sync background job/scheduler -- [ ] Financial insights LangChain tools +- [x] YNAB API client module +- [x] YNAB data models (Budget, Transaction, Category, Account) +- [x] Database schema for financial data +- [x] YNAB sync background job/scheduler +- [x] Financial insights LangChain tools - [ ] Admin UI for YNAB configuration - [ ] Frontend components for money insights display diff --git a/migrations/models/16_20261012172428_add_ynab_mirror.py b/migrations/models/16_20261012172428_add_ynab_mirror.py new file mode 100644 index 0000000..a2b6927 --- /dev/null +++ b/migrations/models/16_20261012172428_add_ynab_mirror.py @@ -0,0 +1,239 @@ +from tortoise import BaseDBAsyncClient + +RUN_IN_TRANSACTION = True + + +async def upgrade(db: BaseDBAsyncClient) -> str: + return """ + CREATE TABLE IF NOT EXISTS "ynab_accounts" ( + "id" VARCHAR(36) NOT NULL PRIMARY KEY, + "name" VARCHAR(255) NOT NULL, + "type" VARCHAR(50) NOT NULL, + "on_budget" BOOL NOT NULL, + "closed" BOOL NOT NULL, + "note" TEXT, + "balance" DECIMAL(15,3) NOT NULL, + "cleared_balance" DECIMAL(15,3) NOT NULL, + "uncleared_balance" DECIMAL(15,3) NOT NULL, + "last_reconciled_at" TIMESTAMPTZ, + "deleted" BOOL NOT NULL DEFAULT False +); + CREATE TABLE IF NOT EXISTS "ynab_categories" ( + "id" VARCHAR(36) NOT NULL PRIMARY KEY, + "category_group_id" VARCHAR(36) NOT NULL, + "name" VARCHAR(255) NOT NULL, + "hidden" BOOL NOT NULL, + "internal" BOOL NOT NULL DEFAULT False, + "note" TEXT, + "goal_type" VARCHAR(20), + "goal_target" DECIMAL(15,3), + "goal_target_month" DATE, + "deleted" BOOL NOT NULL DEFAULT False +); +CREATE INDEX IF NOT EXISTS "idx_ynab_catego_categor_2fc4cc" ON "ynab_categories" ("category_group_id"); + CREATE TABLE IF NOT EXISTS "ynab_category_groups" ( + "id" VARCHAR(36) NOT NULL PRIMARY KEY, + "name" VARCHAR(255) NOT NULL, + "hidden" BOOL NOT NULL, + "internal" BOOL NOT NULL DEFAULT False, + "deleted" BOOL NOT NULL DEFAULT False +); + CREATE TABLE IF NOT EXISTS "ynab_changes" ( + "id" BIGSERIAL NOT NULL PRIMARY KEY, + "synced_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + "server_knowledge" BIGINT NOT NULL, + "entity_type" VARCHAR(30) NOT NULL, + "entity_id" VARCHAR(64) NOT NULL, + "action" VARCHAR(10) NOT NULL, + "label" VARCHAR(500) NOT NULL, + "changes" JSONB NOT NULL +); +CREATE INDEX IF NOT EXISTS "idx_ynab_change_synced__77ae75" ON "ynab_changes" ("synced_at"); +CREATE INDEX IF NOT EXISTS "idx_ynab_change_entity__c12b5b" ON "ynab_changes" ("entity_type"); +CREATE INDEX IF NOT EXISTS "idx_ynab_change_entity__9fb442" ON "ynab_changes" ("entity_id"); +COMMENT ON TABLE "ynab_changes" IS 'Append-only log of what each sync changed, built by diffing against the mirror.'; + CREATE TABLE IF NOT EXISTS "ynab_months" ( + "id" VARCHAR(10) NOT NULL PRIMARY KEY, + "month" DATE NOT NULL, + "note" TEXT, + "income" DECIMAL(15,3) NOT NULL, + "budgeted" DECIMAL(15,3) NOT NULL, + "activity" DECIMAL(15,3) NOT NULL, + "to_be_budgeted" DECIMAL(15,3) NOT NULL, + "age_of_money" INT, + "deleted" BOOL NOT NULL DEFAULT False +); +CREATE INDEX IF NOT EXISTS "idx_ynab_months_month_483dcb" ON "ynab_months" ("month"); + CREATE TABLE IF NOT EXISTS "ynab_month_categories" ( + "id" VARCHAR(47) NOT NULL PRIMARY KEY, + "month" DATE NOT NULL, + "category_id" VARCHAR(36) NOT NULL, + "budgeted" DECIMAL(15,3) NOT NULL, + "activity" DECIMAL(15,3) NOT NULL, + "balance" DECIMAL(15,3) NOT NULL, + "deleted" BOOL NOT NULL DEFAULT False +); +CREATE INDEX IF NOT EXISTS "idx_ynab_month__month_f13a79" ON "ynab_month_categories" ("month"); +CREATE INDEX IF NOT EXISTS "idx_ynab_month__categor_a62299" ON "ynab_month_categories" ("category_id"); +COMMENT ON TABLE "ynab_month_categories" IS 'Per-month category amounts. ``budgeted`` is what YNAB calls \"Assigned\".'; + CREATE TABLE IF NOT EXISTS "ynab_payees" ( + "id" VARCHAR(36) NOT NULL PRIMARY KEY, + "name" VARCHAR(500) NOT NULL, + "transfer_account_id" VARCHAR(36), + "deleted" BOOL NOT NULL DEFAULT False +); + CREATE TABLE IF NOT EXISTS "ynab_subtransactions" ( + "id" VARCHAR(64) NOT NULL PRIMARY KEY, + "transaction_id" VARCHAR(64) NOT NULL, + "amount" DECIMAL(15,3) NOT NULL, + "memo" TEXT, + "payee_id" VARCHAR(36), + "category_id" VARCHAR(36), + "transfer_account_id" VARCHAR(36), + "deleted" BOOL NOT NULL DEFAULT False +); +CREATE INDEX IF NOT EXISTS "idx_ynab_subtra_transac_0571ba" ON "ynab_subtransactions" ("transaction_id"); +CREATE INDEX IF NOT EXISTS "idx_ynab_subtra_categor_c16c94" ON "ynab_subtransactions" ("category_id"); +COMMENT ON TABLE "ynab_subtransactions" IS 'Lines of a split transaction; the parent''s category is \"Split\".'; + CREATE TABLE IF NOT EXISTS "ynab_sync_state" ( + "budget_id" VARCHAR(64) NOT NULL PRIMARY KEY, + "budget_name" VARCHAR(255), + "server_knowledge" BIGINT, + "last_synced_at" TIMESTAMPTZ, + "last_full_sync_at" TIMESTAMPTZ, + "last_error" TEXT +); + CREATE TABLE IF NOT EXISTS "ynab_transactions" ( + "id" VARCHAR(64) NOT NULL PRIMARY KEY, + "date" DATE NOT NULL, + "amount" DECIMAL(15,3) NOT NULL, + "memo" TEXT, + "cleared" VARCHAR(20) NOT NULL, + "approved" BOOL NOT NULL, + "flag_color" VARCHAR(20), + "account_id" VARCHAR(36) NOT NULL, + "payee_id" VARCHAR(36), + "category_id" VARCHAR(36), + "transfer_account_id" VARCHAR(36), + "import_payee_name" VARCHAR(500), + "account_name" VARCHAR(255), + "payee_name" VARCHAR(500), + "category_name" VARCHAR(255), + "deleted" BOOL NOT NULL DEFAULT False +); +CREATE INDEX IF NOT EXISTS "idx_ynab_transa_date_a71ead" ON "ynab_transactions" ("date"); +CREATE INDEX IF NOT EXISTS "idx_ynab_transa_account_414dfe" ON "ynab_transactions" ("account_id"); +CREATE INDEX IF NOT EXISTS "idx_ynab_transa_payee_i_4146bc" ON "ynab_transactions" ("payee_id"); +CREATE INDEX IF NOT EXISTS "idx_ynab_transa_categor_232d47" ON "ynab_transactions" ("category_id");""" + + +async def downgrade(db: BaseDBAsyncClient) -> str: + return """ + DROP TABLE IF EXISTS "ynab_months"; + DROP TABLE IF EXISTS "ynab_categories"; + DROP TABLE IF EXISTS "ynab_transactions"; + DROP TABLE IF EXISTS "ynab_payees"; + DROP TABLE IF EXISTS "ynab_subtransactions"; + DROP TABLE IF EXISTS "ynab_accounts"; + DROP TABLE IF EXISTS "ynab_sync_state"; + DROP TABLE IF EXISTS "ynab_category_groups"; + DROP TABLE IF EXISTS "ynab_changes"; + DROP TABLE IF EXISTS "ynab_month_categories";""" + + +MODELS_STATE = ( + "eJztXQtv27YW/iuEcYF1gBvk3S734gJO4nbZ8igS925dU6i0RNtaZErTI6kx7L9fkpJsPS" + "jZlCVbdAmsay3xUNLHI55zvnNI/d2Z2gayvL0LGz8j14O+aePOGfi7g+EUkX9wz3dBBzrO" + "4iw94MOhxQT0REt2Bg4934W6T06OoOUhcshAnu6aTnQxHFgWPWjrpKGJx4tDATb/CpDm22" + "PkT5BLTnz+Qg6b2EDfkBf/dJ60kYksI3XfpkGvzY5r/sxhxz5+vLp8x1rSyw013baCKV60" + "dmb+xMbz5kFgGntUhp4bI4xc6CMj8Rj0LqPHjg+Fd0wO+G6A5rdqLA4YaAQDi4LR+c8owD" + "rFALAr0f8d/7cjAA+BmkJrYp9i8fc/4VMtnpkd7dBLXfzcu391dPoje0rb88cuO8kQ6fzD" + "BKEPQ1GG6wJI9ncOyosJdPlQxu0zYJIbrQJjfGCB40KHYiBjgKqh1pnCb5qF8NifkJ+HJy" + "clMP6vd8+QJK0YlDbR61Drb6NTh+E5CukCQt1F9JE16OeBvCRnfHOK+GCmJTOQGpHoXvyP" + "lgJMnsG4w9YseglK8B1c3fQfBr2bD/RJpp73l8Ug6g369MwhOzrLHH11mhmKeSfgt6vBz4" + "D+BH/c3fazuj9vN/ijQ+8JBr6tYftFg0bifY2PxsCkBjZwjIoDm5ZUA7vVgY1uPvHCTiDG" + "yBKZ9hIilWa+6K6Ehq/zgoZr2IvMzLe/ysS3Xzzv7WenvcBDriZmhxMiaxjjKliu9ypUN7" + "3UgRk9cS0vBSMP3jvbReYY/4pmDMIrch8Q6zyTG7luH6Nu2gra4ujiLlz4MvfpkkpBno48" + "E/LDl7H3cNG77HcYhkOoP71A19AKwJwiz4Nj5OUBPY8k3/16j6y5g8vHMukG34Q9tnNuLs" + "KWYWUf2gmMUujlT00Pp9kjEJPnNqJr0yuVoLMklkiAuFpIoSUHUoUWUocWPvrGcZwG5Cgf" + "yri9LKFFmUPU/32Q8oViQ/rqpvf7jyl/6Pru9n3cPGF4L67vzlWcsYvuaD7O8BwEn3jOAP" + "VH+ziY5lyB1OgmxLf86nQ+PvTvzwC16I/44ermvHcGPHM6hI94cHd3TdCwbeZDi3qvB6t4" + "rwfF3utBzns1p8TKaE9oJhIFpIQ2FgfUSYCcHByugCRpVQglO5fGMrLZGomRDEuITcpLSo" + "lqM7RS0i0Sc0o4onV6KJIGXHqG5F4z8Mpy5q1FcWkExlEX0UisybiDRrg3aGq7sw4n3Eic" + "LY0yWJg5pQ1NFV3sQHRBLuIjLBRgJERUjKFijO8qxlC5jJ0Y2FwuY2ssfLsmQ1lo+BZ7gW" + "vw8E17f0V+33KPT3l68nt6dBxFy1SSMrL4ehvgFBzoeS+2y9HMcxNDd8YHMymVAXM488No" + "qnUMTQl459TvTVp3Atn51W3v/hPfdY7bJ/X2/NOg38ugi6bQFKoqmAvUo6LL3/g6FfRgfz" + "VStoyVzdGyLxPoe2RK13AwHRZR4nwwOaK1kImbRfVoFVCPijE94jDdEckqDilHVEEav7ca" + "wvTxeHOpbVsI4pJ3PimbnVGjFEkT9knU+xGYUu/urlNT6vlVlm34eHPeJ/MBg5k0MkPv8u" + "p2wAV3MoU6cbueEIeqXTKhZmQl1NiD01Vm1mzQmZhYT7Maa5uGrnnB8E+kcwiAYkCzchKC" + "2YgjRULyiea49rNpiE2pOcHNeacdy9ahtYZ7n0klrjKtnhRPqye5adUyoKONXTtwOHVcvz" + "zc3fIhzYhlAP2IyVN+Nkzd7wLL9PwvjcH7+UszoRN98nKiN8vpZnxV2oEieneSD1RE744O" + "7LyaM0e0LSuFza3Qqqcets1Bfr4ONl2eskj6VscinWdun+avBEXoH0NdtwMqthYgfdpXL+" + "xKYkg8fYKMgMRiWj015A9xf3IWkCehGSLLpL7qyLTWReWcdPUh7EkyQJrMbbBXqMNJbvRj" + "YrA4u8Fe5dXSGx3WHYj0m/ztQ0q2A4gNENVE7HUyKK0i84gf8YNvExMHHOh6yADspgBr+G" + "L6E3C0/9qAM+Ai2p7czB64NvGTB3wbXExcewovzx/xyHbBM4kvyV8egq4+ATp04NC0TJ/M" + "2uzOVBqmgi7XnIaJuUAeoMsrPLnASsFzN8IfVKBi1mVhtp7ROlkpYXBSkjA4yScMRmQWoU" + "400WOOiSpGMysnJ6SN6CaJQQoBLVm9Y8sI56br62j0LBqXxzJyRuSSRODxY5dyK0PbmGmi" + "y9pSQpIsb9j0a8EwmvhTTvZ8CbCxkAKWX8/LvGxyRXJrgp4bR1QSkNWeL4plrUqfo2+OSX" + "qrMLBpSTkHVpKBXMlYRyynYHF0WkrVR8eA5DEULpGWlzDuZkql00rSpmrpFMZFxGJiDJbw" + "i6lUwao8YyRDacKROQ5cRg0DyvJd3fQ+0MMYMbrLK2IdBXpIcJBzKRcZlG4k9xiykAjr7s" + "whigbiStZuog/gId8nl/e6j5iym3E3RJ39wNsDD4Hj2K7vgSlRMdOxUNzCAw5y2WJ3xVG2" + "hKOMlFacDsoJKqYyZixMz7HgTBOtwM/KqdghsfWEo02IYovAmRKShV3bFJh0fs6DeYULiI" + "OUTAZLsznf5KefjtaAkm229frw4PjN8duj0+O3pAm7lfmRNyXg5qtsGQZVVtbkBJUyZpSx" + "cI1NP/ZDijmtXA+yoLtpYsskgTbxR545ultafp+S22Dp/dzwt7jy3oKeryHXtTkVzcUqm5" + "aSxMyrfRUUX6jKbdXAqn0V1L4K7d1XIUNs1FGwKxeW6ZqpGda1kJWrAYgH0tvDvDOJIGmc" + "Pk4gU8Qgp8FbRiJnBm45jzwgp588QOWA49p00vFY5SntBDHaFSWJ4jyPLNwD5ZFvKMzkjw" + "eoTxsKU9tLZKZONzxIOtNJX8iYl8aSmTBijkekw8BFgD7dE7kRWubqhewxMEnsh6aI9G+F" + "HVMB5jUDtvklaa5Y5JawyCyiYUpbpVwrL12Db9euCKlFrtxKqWA2JHElcsB7Twq5Qp7o5i" + "jD/TVc85oJwxiDgmxwIYA5ue8SPdo90gNKPGmRmeA4MYUgFol/l1iyVzJCofoMne1ATdJb" + "nqQVk7KbTIqqw9qxOqymixrkr8LKrQHmxNG8dcLFgTR/lbLa2FLqONNFuumY3E3Mi/PvKS" + "FZssObqMgv+6jj8o/orPuBx/pA7VCL+9B73z8D8f57j7h/07u6PgPzDSO3//lHtf9+E3ni" + "LawL3skau4W9FA8nsrJyBhSSBBArhYZFSZ8Vv482l97gXnuRVxcmSfY+9G8vr27f5z2QTn" + "TmDDgIG2Q8H/FD/3ZwBjwyUT7id2TW71+escwCMh7xRe/2on/NDun0Wa1oD9HtmwPilgSu" + "iyL8qwxTuocNDtX9/MJ7t5ECZwaJHibA2phY4ksyJJ/OgEFGZPaIf+v3f6U/XxB6or9v7m" + "4HP9MDU2LrJtasHYMTeEgjysiz1qXFcik5tU9tZp9amryLwzERNygnKInZVjVzihBUNXNq" + "YAuZ3vi1G85Ev1eZFVR8bwKTGihf+avocirSJr73wrXxL/aww6F541PdMnZXJ420P+2hIn" + "XlJ3VF11OpZVT5JaeJOxPwqzNiyqvmetWOa08dIdp2ISGLjm4aU5XIUYkcaRM5FT9jtJUP" + "GEmzitINcIXwNiOqStLaUDdclnwoqwBfL+uw7axcA1/3UauLFVOqCDXFlH5/Ayv6MZ8mua" + "rklzg4fFXmQx3FnFXu4yD18laf5wuI2YVCf/mLYrOq2Jw12KwE+AIOUFpKxbaZL+qss0FQ" + "rgdZ0N20T8SAKtzFYYnuFm/l0FaXc1PK66IRmT0mRZ/mXVWDc91IAvOmtThkBEKwKvmB3A" + "4UsdAGYkHFwk28MYrOrJfOVNzCToSgilvY0YFVO5epncvavHOZS7dPCly49veGKTN1P+9M" + "LmQb3bQrA0wBqZeGbgmvlxm0uqk9Z0EwsssNA8/EyPMiZUtk3FwTYUPLxuTst05Gfkw/x6" + "1Ywa2wgplBE+JXMrKycFgbIFkKNH5VdAvEN7iaag1l3Ry8okWaGTFJQu9NcYLzmVh0GkgK" + "yqiiB6sUSRwUF0kc5Iokvq/q4Ua+CK2bvpAmxu3VSz1XQhLoTIa2O7FtIeOTlVOIznUyMK" + "nDkwfzl4e72wK1XIhkcPyIyQN+Nkzd7wLL9PwvjU2UC290GJiWb2Jvj16wIYeUYlFOB2eZ" + "34znSjvIl7ubOtJIJMNbo1usyhkxKTW5fusk3z4M/MLs837/9gwMEcKP+Lfe7UAb3GmDe7" + "qMn0S8JFbVfLclK/eJBj5p7AHzqBfuFpoWqrRH6BZ0t+ZdQj3ddnk7gyLdnEKLD9xcJstp" + "h0J7kXAr8SsB57J/cXXTu3510j3MpHtinTzOO6K2z9uhtjg5OReQZK7cdF7Sh2MOnsX2P2" + "6vjH914z9Cvl6tbCItqVJc7UpxqZz0ToxrPicdJQYE05dpKZXBTOZX1kxiZsrBW4vf0lxm" + "WkfatHnEJwyHJV9sT57uliXuZqSh2PfahbJ26yTZSr4mW2N6YrMf2z46XSHWO8rOpYtYj5" + "5SO0k0zLsxDAQgjNvLCWH9azYJBsPAIJNAHsTScsqU3AYLKhuzxXUWVFq2J1yfuhBSaOZ4" + "ClGaQrEUhSzFEFqQu4drKW+WkFqbOWuXpsbU2cFJ92hl7ozgC126V1klMDnSClR66+vByp" + "VXwM5XCJGr6Ga1DdP5PahFRlteZBRGuKKeRkJK7fjMYVm2RxJcxNVSBSzB/Hx3KU0QFV6Z" + "bfuMkCIKqhAFcRmdNnbtwBGsGOUKNxX/SgG0YmBqZGAmpmHwFoqX2qCFkIp2U2iSqyEXQ8" + "5mHKV4JsWUUVcEQoMEwtiGliZKu6aEJAG26cqvEBPocsnX0kg3I7mbtUxCIW4CEY19mIcf" + "4S6FcyFcFtxKBykJVnM7cKvQcRdDx/fUx+8siR/DRt1Vg8goclCRpDQBjko5q4CnSYOiAp" + "42Q6ps+47Z9glbpVRk1CfxGqZl1pw1XM2Kd3oO/VzpaxtbM2DZY2CPwMsE+gBBfQK8GdZB" + "2J3RDYvHwXAGDHM0IoMF4JiMj+cDYuXB1KS7me11MqNZd/+P+BF/uu2d/+CB3ocrMIEewD" + "aYmJ5P3Jcu8GzS2PQA+Y8KsavSdJZr0Asjw/Q98ArtjfcAJB1BzzPHeIqwD8Y2veDItafg" + "X8f7+8C3wb/oitQ9cOUDj8QMRBCGdzIyXXJP9NbZ09bs7Zyb48LVQFx/h7MKKHJdturuhG" + "uAfjo8PDp6c7h/dPr25PjNm5O3+/PFQPlTZauCzq/e09c19VrHE2KxE0QHqdqHo5OCm6kt" + "r3sAdre03EPuM3K1J3LeQgZvNWjZS8STrrSwbhtO7wbfquQ+jL7pz4SpwIyYnImoVcjAo2" + "Iy8IjzjR6GiligmxKSEsjT4xWAPD0uBJKeSgMJdf5X5YpRXEjIGeTWvwqcPFrR57mK6oOG" + "rfggV+Wy6iZ2KVm422kYS3aEWIhsa1FoY5RBbcs/WxSY3bBkQUFcdhNnEpaEZSzjoLjVxk" + "1NXdNkcVghnHiqK9m0eSsulmxSGfhaM/Am1m0ei1+aM14IqZJoYgPY2iUeR1q+DCIhpmAM" + "Xedn7m5spTAmxRSMxCGwtSHSKupkXlhBSlRsjDR7RJ0rxNHOQhYoK/adbq2kMkidncogsV" + "hk2eqCdKPuaoGL6EKDzgfkvmaSIK4uAXDKdjTYA1+/xtPY1680V8MyQjStQ9palgceOz2W" + "mkHGYyefUqq1ZxVQLQ2oSueU+cz8pjCgoqdUQFUpoJoXZlVcCSIxY1z/GhAVC6hYoD0wqv" + "X5NYCo/Nfd8l8/wBlCRX5reHK5v+rQdopol8ZCqyLmBrOT5GrYGyE33k9O0JMsEJeEsG/a" + "o1TmZ7fMz0MwZAofVYoU2KFMq+UGyUsJrMifXNPPatGqVQg8hyAGEl38mxWiOtBF2P/BW9" + "AgJiU4HmhrHm9SS4/KLm6s1qnYLiYGrsp8npKUkhxooJxsGu/LKhLTzoVUMEamPzS1RbL/" + "cXtJnIlNZ/9ZGCP4eidlJIG1aR9tixTqcjCliM9UMKGCCRVMiAcTM6w/+NAv5LMWDVYIIU" + "hbzZs33gqvFSZCRD8SnBRS3jzPm48gEiW7MmJSzqKNrNtvyfqwrVUHbXh5GNsztvJKy7y0" + "2m12y7vNsiEZkccMrU7FMc12oIa1DcOK6Cp6kSA9LSWJkdlEqN4iT3OwnLMeCBHWwmy1yq" + "O22sOMq+FWLa+L2+94dZ3ifBXnu31DwvneiBBRuRCRs+yh/g1XiXVz7WdhKi0ppjZvSyE6" + "suCYIsXzHYtVMy0lySvfuHJWos5rYMy3nbOtnzFvaYZMMhRVgkwlyKQA15w6tuuH5b3CbD" + "lXWEpgGykTjRVOFNasnJSINpKFqKamSj8LNlmKrY0ooDlBKTFtRENVwUFn/SipJTRwD7mm" + "zt2iKjrTLSN94aJNa5jeHdoFd83tGop53WfkeoKbICZEJKWLmpgK6ash4vWEzeUE8GAl+3" + "xQYp8POPbZxj7iceklmyAuRNQmiNmEbys2Qfzn/zygDEE=" +) diff --git a/pyproject.toml b/pyproject.toml index 3fe1a42..90a1c58 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,7 +36,6 @@ dependencies = [ "aiosqlite>=0.20.0", "jq>=1.10.0", "tavily-python>=0.7.17", - "ynab>=1.3.0", "aioimaplib>=2.0.1", "html2text>=2025.4.15", "ollama>=0.6.1", @@ -47,6 +46,7 @@ dependencies = [ "google-auth-oauthlib>=1.0.0", "google-auth-httplib2>=0.2.0", "langchain-mcp-adapters>=0.3.2", + "sqlglot>=30.22.0", ] [project.optional-dependencies] diff --git a/tests/unit/test_ynab_mirror.py b/tests/unit/test_ynab_mirror.py new file mode 100644 index 0000000..8d9bc4f --- /dev/null +++ b/tests/unit/test_ynab_mirror.py @@ -0,0 +1,249 @@ +"""Tests for the YNAB Postgres mirror: delta application, change log, SQL guard.""" + +from decimal import Decimal + +import pytest +from tortoise import Tortoise +from tortoise.transactions import in_transaction + +from blueprints.ynab.models import YnabChange, YnabMonthCategory, YnabTransaction +from blueprints.ynab.sql_tool import format_rows, validate_sql +from blueprints.ynab.sync import _money, apply_budget + +ACCOUNT = { + "id": "acct-1", + "name": "Checking", + "type": "checking", + "on_budget": True, + "closed": False, + "note": None, + "balance": 1_000_000, + "cleared_balance": 1_000_000, + "uncleared_balance": 0, + "last_reconciled_at": None, + "deleted": False, +} +PAYEE = { + "id": "payee-1", + "name": "Costco", + "transfer_account_id": None, + "deleted": False, +} +GROUP = {"id": "grp-1", "name": "Bills", "hidden": False, "deleted": False} +CATEGORY = { + "id": "cat-1", + "category_group_id": "grp-1", + "name": "Groceries", + "hidden": False, + "note": None, + "goal_type": None, + "goal_target": 0, + "goal_target_month": None, + "deleted": False, +} + + +def _txn(**overrides): + txn = { + "id": "txn-1", + "date": "2026-10-03", + "amount": -54_150, + "memo": None, + "cleared": "cleared", + "approved": True, + "flag_color": None, + "account_id": "acct-1", + "payee_id": "payee-1", + "category_id": "cat-1", + "transfer_account_id": None, + "import_payee_name": None, + "deleted": False, + } + return {**txn, **overrides} + + +def _month(budgeted=400_000, activity=-54_150): + return { + "month": "2026-10-01", + "note": "", + "income": 0, + "budgeted": budgeted, + "activity": activity, + "to_be_budgeted": 0, + "age_of_money": 30, + "deleted": False, + "categories": [ + { + **CATEGORY, + "budgeted": budgeted, + "activity": activity, + "balance": budgeted + activity, + } + ], + } + + +def _budget(transactions, month=None): + return { + "accounts": [ACCOUNT], + "payees": [PAYEE], + "category_groups": [GROUP], + "categories": [CATEGORY], + "months": [month or _month()], + "transactions": transactions, + "subtransactions": [], + } + + +@pytest.fixture +async def db(): + await Tortoise.init( + db_url="sqlite://:memory:", modules={"models": ["blueprints.ynab.models"]} + ) + await Tortoise.generate_schemas() + yield + await Tortoise.close_connections() + + +async def _apply(budget, knowledge, log_changes=True): + async with in_transaction() as conn: + return await apply_budget(budget, knowledge, conn, log_changes=log_changes) + + +def test_money_is_exact(): + assert _money(-54_150) == Decimal("-54.15") + assert _money(1) == Decimal("0.001") + assert _money(None) is None + + +async def test_initial_sync_writes_rows_without_logging(db): + counts = await _apply(_budget([_txn()]), 1, log_changes=False) + + assert counts["transactions"] == 1 + txn = await YnabTransaction.get(id="txn-1") + assert txn.amount == Decimal("-54.15") + # Names are denormalized onto the transaction + assert (txn.payee_name, txn.category_name, txn.account_name) == ( + "Costco", + "Groceries", + "Checking", + ) + assert await YnabChange.all().count() == 0 + + +async def test_unchanged_delta_is_a_noop(db): + await _apply(_budget([_txn()]), 1, log_changes=False) + counts = await _apply(_budget([_txn()]), 2) + + assert not any(counts.values()) + assert await YnabChange.all().count() == 0 + + +async def test_edit_logs_field_diff(db): + await _apply(_budget([_txn()]), 1, log_changes=False) + await _apply(_budget([_txn(amount=-60_000, memo="bulk")]), 2) + + change = await YnabChange.get(entity_type="transaction") + assert change.action == "updated" + assert change.server_knowledge == 2 + assert change.changes == {"amount": [-54.15, -60.0], "memo": [None, "bulk"]} + assert "Costco" in change.label + + +async def test_assignment_change_is_logged_but_activity_is_not(db): + await _apply(_budget([]), 1, log_changes=False) + # Only activity moves: derived, so not logged + await _apply(_budget([], _month(activity=-80_000)), 2) + assert await YnabChange.filter(entity_type="month_category").count() == 0 + + await _apply(_budget([], _month(budgeted=500_000, activity=-80_000)), 3) + change = await YnabChange.get(entity_type="month_category") + assert change.changes == {"budgeted": [400.0, 500.0]} + assert change.label == "Groceries 2026-10" + mc = await YnabMonthCategory.get(id="2026-10-01:cat-1") + assert mc.balance == Decimal("420") + + +async def test_new_and_deleted_transactions(db): + await _apply(_budget([_txn()]), 1, log_changes=False) + await _apply(_budget([_txn(deleted=True), _txn(id="txn-2")]), 2) + + actions = { + c.entity_id: c.action + for c in await YnabChange.filter(entity_type="transaction") + } + assert actions == {"txn-1": "deleted", "txn-2": "created"} + assert (await YnabTransaction.get(id="txn-1")).deleted is True + + +async def test_created_then_deleted_between_syncs_is_not_logged(db): + await _apply(_budget([]), 1, log_changes=False) + await _apply(_budget([_txn(id="txn-9", deleted=True)]), 2) + assert await YnabChange.filter(entity_type="transaction").count() == 0 + + +class TestValidateSql: + @pytest.mark.parametrize( + "query", + [ + "SELECT payee_name, sum(amount) FROM ynab_transactions WHERE NOT deleted GROUP BY 1", + "WITH x AS (SELECT * FROM ynab_payees) SELECT * FROM x", + "SELECT date_trunc('month', date) m, count(*) FROM ynab_transactions GROUP BY 1;", + "SELECT * FROM generate_series(1, 3) g", + "WITH RECURSIVE r AS (SELECT 1 n UNION ALL SELECT n + 1 FROM r WHERE n < 3) SELECT * FROM r", + "WITH ynab_payees AS (SELECT * FROM ynab_payees WHERE NOT deleted) SELECT * FROM ynab_payees", + "SELECT * FROM ynab_transactions UNION ALL SELECT * FROM ynab_transactions", + ], + ) + def test_allows_read_queries(self, query): + assert validate_sql(query) is None + + @pytest.mark.parametrize( + "query, error", + [ + ("DELETE FROM ynab_payees", "Only SELECT"), + ("SELECT 1; SELECT 2", "Exactly one"), + ("SELECT * FROM users", "not allowed"), + ("SELECT * FROM public.conversations", "not allowed"), + ("SELECT * FROM pg_catalog.pg_user", "not allowed"), + ("SELECT * INTO evil FROM ynab_payees", "read-only"), + ("SELECT * FROM ynab_payees FOR UPDATE", "read-only"), + ( + "WITH x AS (DELETE FROM ynab_payees RETURNING *) SELECT * FROM x", + "read-only", + ), + ("SELECT pg_read_file('/etc/passwd')", "pg_read_file"), + ("SELECT (SELECT lo_import('/etc/passwd'))", "lo_import"), + ("SELECT current_setting('is_superuser')", "current_setting"), + # CTE named like a real table: its body still reads the real table + ("WITH users AS (SELECT * FROM users) SELECT * FROM users", "users"), + # CTE defined in a subquery doesn't cover a reference outside it + ("SELECT * FROM (WITH users AS (SELECT 1) SELECT 1) x, users", "users"), + ("SELECT * FROM ynab_payees WHERE id IN (SELECT id FROM users)", "users"), + ], + ) + def test_rejects(self, query, error): + result = validate_sql(query) + assert result is not None and error in result + + +def test_format_rows(): + rows = [ + {"payee": "Costco", "total": Decimal("-54.150")}, + {"payee": None, "total": Decimal("10.000")}, + ] + assert format_rows(rows) == "payee\ttotal\nCostco\t-54.150\n\t10" + assert format_rows([]) == "Query returned no rows." + + +async def test_budget_summary_rejects_bad_month(monkeypatch): + from blueprints.ynab import tools + + async def fresh(): + return object() + + monkeypatch.setattr(tools, "ensure_fresh", fresh) + for month in ("October", "2026-13"): + assert await tools.ynab_budget_summary.ainvoke({"month": month}) == ( + "Month must be YYYY-MM." + ) diff --git a/tests/unit/test_ynab_service.py b/tests/unit/test_ynab_service.py deleted file mode 100644 index da19b7b..0000000 --- a/tests/unit/test_ynab_service.py +++ /dev/null @@ -1,254 +0,0 @@ -"""Tests for YNAB service data formatting and filtering logic.""" - -import os -from unittest.mock import MagicMock, patch - -import pytest - - -def _mock_category( - name, budgeted, activity, balance, deleted=False, hidden=False, goal_type=None -): - cat = MagicMock() - cat.name = name - cat.budgeted = budgeted - cat.activity = activity - cat.balance = balance - cat.deleted = deleted - cat.hidden = hidden - cat.goal_type = goal_type - return cat - - -def _mock_transaction( - var_date, payee_name, category_name, amount, memo="", deleted=False, approved=True -): - txn = MagicMock() - txn.var_date = var_date - txn.payee_name = payee_name - txn.category_name = category_name - txn.amount = amount - txn.memo = memo - txn.deleted = deleted - txn.approved = approved - return txn - - -@pytest.fixture -def ynab_service(): - """Create a YNABService with mocked API client.""" - with patch.dict( - os.environ, {"YNAB_ACCESS_TOKEN": "fake-token", "YNAB_BUDGET_ID": "test-budget"} - ): - with patch("utils.ynab_service.ynab") as mock_ynab: - # Mock the configuration and API client chain - mock_ynab.Configuration.return_value = MagicMock() - mock_ynab.ApiClient.return_value = MagicMock() - mock_ynab.PlansApi.return_value = MagicMock() - mock_ynab.TransactionsApi.return_value = MagicMock() - mock_ynab.MonthsApi.return_value = MagicMock() - mock_ynab.CategoriesApi.return_value = MagicMock() - - from utils.ynab_service import YNABService - - service = YNABService() - yield service - - -class TestGetBudgetSummary: - def test_calculates_totals(self, ynab_service): - categories = [ - _mock_category("Groceries", 500_000, -350_000, 150_000), - _mock_category("Rent", 1_500_000, -1_500_000, 0), - ] - - mock_month = MagicMock() - mock_month.to_be_budgeted = 200_000 - - mock_budget = MagicMock() - mock_budget.name = "My Budget" - mock_budget.months = [mock_month] - mock_budget.categories = categories - mock_budget.currency_format = MagicMock(iso_code="USD") - - mock_response = MagicMock() - mock_response.data.budget = mock_budget - ynab_service.plans_api.get_plan_by_id.return_value = mock_response - - result = ynab_service.get_budget_summary() - - assert result["budget_name"] == "My Budget" - assert result["to_be_budgeted"] == 200.0 - assert result["total_budgeted"] == 2000.0 # (500k + 1500k) / 1000 - assert result["total_activity"] == -1850.0 - assert result["currency_format"] == "USD" - - def test_skips_deleted_and_hidden(self, ynab_service): - categories = [ - _mock_category("Active", 100_000, -50_000, 50_000), - _mock_category("Deleted", 999_000, -999_000, 0, deleted=True), - _mock_category("Hidden", 999_000, -999_000, 0, hidden=True), - ] - - mock_month = MagicMock() - mock_month.to_be_budgeted = 0 - mock_budget = MagicMock() - mock_budget.name = "Budget" - mock_budget.months = [mock_month] - mock_budget.categories = categories - mock_budget.currency_format = None - - mock_response = MagicMock() - mock_response.data.budget = mock_budget - ynab_service.plans_api.get_plan_by_id.return_value = mock_response - - result = ynab_service.get_budget_summary() - assert result["total_budgeted"] == 100.0 - assert result["currency_format"] == "USD" # Default fallback - - -class TestGetTransactions: - def test_filters_by_date_range(self, ynab_service): - transactions = [ - _mock_transaction("2026-01-05", "Store", "Groceries", -25_000), - _mock_transaction("2026-01-15", "Gas", "Transport", -40_000), - _mock_transaction( - "2026-02-01", "Store", "Groceries", -30_000 - ), # Out of range - ] - - mock_response = MagicMock() - mock_response.data.transactions = transactions - ynab_service.transactions_api.get_transactions.return_value = mock_response - - result = ynab_service.get_transactions( - start_date="2026-01-01", end_date="2026-01-31" - ) - - assert result["count"] == 2 - assert result["total_amount"] == -65.0 - - def test_filters_by_category(self, ynab_service): - transactions = [ - _mock_transaction("2026-01-05", "Store", "Groceries", -25_000), - _mock_transaction("2026-01-06", "Gas", "Transport", -40_000), - ] - - mock_response = MagicMock() - mock_response.data.transactions = transactions - ynab_service.transactions_api.get_transactions.return_value = mock_response - - result = ynab_service.get_transactions( - start_date="2026-01-01", - end_date="2026-01-31", - category_name="groceries", # Case insensitive - ) - - assert result["count"] == 1 - assert result["transactions"][0]["category"] == "Groceries" - - def test_filters_by_payee(self, ynab_service): - transactions = [ - _mock_transaction("2026-01-05", "Whole Foods", "Groceries", -25_000), - _mock_transaction("2026-01-06", "Shell Gas", "Transport", -40_000), - ] - - mock_response = MagicMock() - mock_response.data.transactions = transactions - ynab_service.transactions_api.get_transactions.return_value = mock_response - - result = ynab_service.get_transactions( - start_date="2026-01-01", - end_date="2026-01-31", - payee_name="whole", - ) - - assert result["count"] == 1 - assert result["transactions"][0]["payee"] == "Whole Foods" - - def test_skips_deleted(self, ynab_service): - transactions = [ - _mock_transaction("2026-01-05", "Store", "Groceries", -25_000), - _mock_transaction("2026-01-06", "Deleted", "Other", -10_000, deleted=True), - ] - - mock_response = MagicMock() - mock_response.data.transactions = transactions - ynab_service.transactions_api.get_transactions.return_value = mock_response - - result = ynab_service.get_transactions( - start_date="2026-01-01", end_date="2026-01-31" - ) - - assert result["count"] == 1 - - def test_converts_milliunits(self, ynab_service): - transactions = [ - _mock_transaction("2026-01-05", "Store", "Groceries", -12_340), - ] - - mock_response = MagicMock() - mock_response.data.transactions = transactions - ynab_service.transactions_api.get_transactions.return_value = mock_response - - result = ynab_service.get_transactions( - start_date="2026-01-01", end_date="2026-01-31" - ) - - assert result["transactions"][0]["amount"] == -12.34 - - def test_sorts_by_date_descending(self, ynab_service): - transactions = [ - _mock_transaction("2026-01-01", "A", "Cat", -10_000), - _mock_transaction("2026-01-15", "B", "Cat", -20_000), - _mock_transaction("2026-01-10", "C", "Cat", -30_000), - ] - - mock_response = MagicMock() - mock_response.data.transactions = transactions - ynab_service.transactions_api.get_transactions.return_value = mock_response - - result = ynab_service.get_transactions( - start_date="2026-01-01", end_date="2026-01-31" - ) - - dates = [t["date"] for t in result["transactions"]] - assert dates == sorted(dates, reverse=True) - - -class TestGetCategorySpending: - def test_month_format_normalization(self, ynab_service): - """Passing YYYY-MM should be normalized to YYYY-MM-01.""" - categories = [_mock_category("Food", 100_000, -50_000, 50_000)] - - mock_month = MagicMock() - mock_month.categories = categories - mock_month.to_be_budgeted = 0 - - mock_response = MagicMock() - mock_response.data.month = mock_month - ynab_service.months_api.get_plan_month.return_value = mock_response - - result = ynab_service.get_category_spending("2026-03") - - assert result["month"] == "2026-03" - - def test_identifies_overspent(self, ynab_service): - categories = [ - _mock_category("Dining", 200_000, -300_000, -100_000), # Overspent - _mock_category("Groceries", 500_000, -400_000, 100_000), # Fine - ] - - mock_month = MagicMock() - mock_month.categories = categories - mock_month.to_be_budgeted = 0 - - mock_response = MagicMock() - mock_response.data.month = mock_month - ynab_service.months_api.get_plan_month.return_value = mock_response - - result = ynab_service.get_category_spending("2026-03") - - assert len(result["overspent_categories"]) == 1 - assert result["overspent_categories"][0]["name"] == "Dining" - assert result["overspent_categories"][0]["overspent_by"] == 100.0 diff --git a/utils/ynab_service.py b/utils/ynab_service.py deleted file mode 100644 index afb4245..0000000 --- a/utils/ynab_service.py +++ /dev/null @@ -1,405 +0,0 @@ -"""YNAB API service for querying budget data.""" - -import os -from datetime import datetime, timedelta -from typing import Any, Optional - -from dotenv import load_dotenv -import ynab - -# Load environment variables -load_dotenv() - - -class YNABService: - """Service for interacting with YNAB API.""" - - def __init__(self): - """Initialize YNAB API client.""" - self.access_token = os.getenv("YNAB_ACCESS_TOKEN", "") - self.budget_id = os.getenv("YNAB_BUDGET_ID", "") - - if not self.access_token: - raise ValueError("YNAB_ACCESS_TOKEN environment variable is required") - - # Configure API client - configuration = ynab.Configuration(access_token=self.access_token) - self.api_client = ynab.ApiClient(configuration) - - # Initialize API endpoints - self.plans_api = ynab.PlansApi(self.api_client) - self.transactions_api = ynab.TransactionsApi(self.api_client) - self.months_api = ynab.MonthsApi(self.api_client) - self.categories_api = ynab.CategoriesApi(self.api_client) - - # Get budget ID if not provided, fall back to last-used - if not self.budget_id: - self.budget_id = "last-used" - - def get_budget_summary(self) -> dict[str, Any]: - """Get overall budget summary and health status. - - Returns: - Dictionary containing budget summary with to-be-budgeted amount, - total budgeted, total activity, and overall budget health. - """ - budget_response = self.plans_api.get_plan_by_id(self.budget_id) - budget_data = budget_response.data.budget - - # Calculate totals from categories - to_be_budgeted = ( - budget_data.months[0].to_be_budgeted / 1000 if budget_data.months else 0 - ) - - total_budgeted = 0 - total_activity = 0 - total_available = 0 - - for category in budget_data.categories or []: - if category.deleted or category.hidden: - continue - total_budgeted += category.budgeted / 1000 - total_activity += category.activity / 1000 - total_available += category.balance / 1000 - - return { - "budget_name": budget_data.name, - "to_be_budgeted": round(to_be_budgeted, 2), - "total_budgeted": round(total_budgeted, 2), - "total_activity": round(total_activity, 2), - "total_available": round(total_available, 2), - "currency_format": budget_data.currency_format.iso_code - if budget_data.currency_format - else "USD", - "summary": f"Budget '{budget_data.name}' has ${abs(to_be_budgeted):.2f} {'to be budgeted' if to_be_budgeted > 0 else 'overbudgeted'}. " - f"Total budgeted: ${total_budgeted:.2f}, Total spent: ${abs(total_activity):.2f}, " - f"Total available: ${total_available:.2f}.", - } - - def get_transactions( - self, - start_date: Optional[str] = None, - end_date: Optional[str] = None, - category_name: Optional[str] = None, - payee_name: Optional[str] = None, - ) -> dict[str, Any]: - """Get transactions filtered by date range, category, or payee. - - Args: - start_date: Start date in YYYY-MM-DD format (defaults to 30 days ago) - end_date: End date in YYYY-MM-DD format (defaults to today) - category_name: Filter by category name (case-insensitive partial match) - payee_name: Filter by payee name (case-insensitive partial match) - - Returns: - Dictionary containing matching transactions and summary statistics. - """ - # Set default date range if not provided - if not start_date: - start_date = (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d") - if not end_date: - end_date = datetime.now().strftime("%Y-%m-%d") - - # Get transactions (SDK v2 requires datetime.date, not string) - since_date_obj = datetime.strptime(start_date, "%Y-%m-%d").date() - transactions_response = self.transactions_api.get_transactions( - self.budget_id, since_date=since_date_obj - ) - - transactions = transactions_response.data.transactions or [] - - # Filter by date range, category, and payee - filtered_transactions = [] - total_amount = 0 - - for txn in transactions: - # Skip if deleted or before start date or after end date - if txn.deleted: - continue - txn_date = str(txn.var_date) - if txn_date < start_date or txn_date > end_date: - continue - - # Filter by category if specified - if category_name and txn.category_name: - if category_name.lower() not in txn.category_name.lower(): - continue - - # Filter by payee if specified - if payee_name and txn.payee_name: - if payee_name.lower() not in txn.payee_name.lower(): - continue - - amount = txn.amount / 1000 # Convert milliunits to dollars - filtered_transactions.append( - { - "date": str(txn.var_date), - "payee": txn.payee_name, - "category": txn.category_name, - "memo": txn.memo, - "amount": round(amount, 2), - "approved": txn.approved, - } - ) - total_amount += amount - - # Sort by date (most recent first) - filtered_transactions.sort(key=lambda x: x["date"], reverse=True) - - return { - "transactions": filtered_transactions, - "count": len(filtered_transactions), - "total_amount": round(total_amount, 2), - "start_date": start_date, - "end_date": end_date, - "filters": {"category": category_name, "payee": payee_name}, - } - - def get_category_spending(self, month: Optional[str] = None) -> dict[str, Any]: - """Get spending breakdown by category for a specific month. - - Args: - month: Month in YYYY-MM format (defaults to current month) - - Returns: - Dictionary containing spending by category and summary. - """ - if not month: - month = datetime.now().strftime("%Y-%m-01") - else: - # Ensure format is YYYY-MM-01 - if len(month) == 7: # YYYY-MM - month = f"{month}-01" - - # Get budget month (SDK v2 requires datetime.date, not string) - month_date_obj = datetime.strptime(month, "%Y-%m-%d").date() - month_response = self.months_api.get_plan_month(self.budget_id, month_date_obj) - - month_data = month_response.data.month - - categories_spending = [] - total_budgeted = 0 - total_activity = 0 - total_available = 0 - overspent_categories = [] - - for category in month_data.categories or []: - if category.deleted or category.hidden: - continue - - budgeted = category.budgeted / 1000 - activity = category.activity / 1000 - available = category.balance / 1000 - - total_budgeted += budgeted - total_activity += activity - total_available += available - - # Track overspent categories - if available < 0: - overspent_categories.append( - { - "name": category.name, - "budgeted": round(budgeted, 2), - "spent": round(abs(activity), 2), - "overspent_by": round(abs(available), 2), - } - ) - - # Only include categories with activity - if activity != 0: - categories_spending.append( - { - "category": category.name, - "budgeted": round(budgeted, 2), - "activity": round(activity, 2), - "available": round(available, 2), - "goal_type": category.goal_type - if hasattr(category, "goal_type") - else None, - } - ) - - # Sort by absolute activity (highest spending first) - categories_spending.sort(key=lambda x: abs(x["activity"]), reverse=True) - - return { - "month": month[:7], # Return YYYY-MM format - "categories": categories_spending, - "total_budgeted": round(total_budgeted, 2), - "total_spent": round(abs(total_activity), 2), - "total_available": round(total_available, 2), - "overspent_categories": overspent_categories, - "to_be_budgeted": round(month_data.to_be_budgeted / 1000, 2) - if month_data.to_be_budgeted - else 0, - } - - def get_category_allocations(self, month: Optional[str] = None) -> dict[str, Any]: - """Get all category allocations (assigned and available) grouped by category group. - - Args: - month: Month in YYYY-MM format (defaults to current month) - - Returns: - Dictionary with all categories grouped, showing assigned and available amounts. - """ - if not month: - month = datetime.now().strftime("%Y-%m-01") - else: - if len(month) == 7: - month = f"{month}-01" - - month_date_obj = datetime.strptime(month, "%Y-%m-%d").date() - - # Get category groups so we know which group each category belongs to - categories_response = self.categories_api.get_categories(self.budget_id) - category_group_map: dict[str, str] = {} - group_order: list[str] = [] - for group in categories_response.data.category_groups or []: - if group.deleted or group.hidden: - continue - if group.name not in group_order: - group_order.append(group.name) - for cat in group.categories or []: - category_group_map[cat.id] = group.name - - # Get month data for assigned/available amounts - month_response = self.months_api.get_plan_month(self.budget_id, month_date_obj) - month_data = month_response.data.month - - groups: dict[str, list] = {g: [] for g in group_order} - total_assigned = 0.0 - total_available = 0.0 - - for category in month_data.categories or []: - if category.deleted or category.hidden: - continue - - assigned = category.budgeted / 1000 - activity = category.activity / 1000 - available = category.balance / 1000 - - group_name = category_group_map.get(category.id, "Other") - if group_name not in groups: - groups[group_name] = [] - - groups[group_name].append( - { - "name": category.name, - "assigned": round(assigned, 2), - "activity": round(activity, 2), - "available": round(available, 2), - } - ) - total_assigned += assigned - total_available += available - - return { - "month": month[:7], - "group_order": group_order, - "groups": groups, - "total_assigned": round(total_assigned, 2), - "total_available": round(total_available, 2), - "to_be_budgeted": round(month_data.to_be_budgeted / 1000, 2) - if month_data.to_be_budgeted - else 0, - } - - def get_spending_insights(self, months_back: int = 3) -> dict[str, Any]: - """Generate insights about spending patterns and budget health. - - Args: - months_back: Number of months to analyze (default 3) - - Returns: - Dictionary containing spending insights, trends, and recommendations. - """ - current_month = datetime.now() - monthly_data = [] - - # Collect data for the last N months - for i in range(months_back): - month_date = current_month - timedelta(days=30 * i) - month_str = month_date.strftime("%Y-%m-01") - - try: - month_spending = self.get_category_spending(month_str) - monthly_data.append( - { - "month": month_str[:7], - "total_spent": month_spending["total_spent"], - "total_budgeted": month_spending["total_budgeted"], - "overspent_categories": month_spending["overspent_categories"], - } - ) - except Exception: - # Skip months that don't have data yet - continue - - if not monthly_data: - return {"error": "No spending data available for analysis"} - - # Calculate average spending - avg_spending = sum(m["total_spent"] for m in monthly_data) / len(monthly_data) - current_spending = monthly_data[0]["total_spent"] if monthly_data else 0 - - # Identify spending trend - if len(monthly_data) >= 2: - recent_avg = sum(m["total_spent"] for m in monthly_data[:2]) / 2 - older_avg = sum(m["total_spent"] for m in monthly_data[1:]) / ( - len(monthly_data) - 1 - ) - trend = ( - "increasing" - if recent_avg > older_avg * 1.1 - else "decreasing" - if recent_avg < older_avg * 0.9 - else "stable" - ) - else: - trend = "insufficient data" - - # Find frequently overspent categories - overspent_frequency = {} - for month in monthly_data: - for cat in month["overspent_categories"]: - cat_name = cat["name"] - if cat_name not in overspent_frequency: - overspent_frequency[cat_name] = 0 - overspent_frequency[cat_name] += 1 - - frequently_overspent = [ - {"category": cat, "months_overspent": count} - for cat, count in overspent_frequency.items() - if count > 1 - ] - frequently_overspent.sort(key=lambda x: x["months_overspent"], reverse=True) - - # Generate recommendations - recommendations = [] - if current_spending > avg_spending * 1.2: - recommendations.append( - f"Current month spending (${current_spending:.2f}) is significantly higher than your {months_back}-month average (${avg_spending:.2f})" - ) - - if frequently_overspent: - top_overspent = frequently_overspent[0] - recommendations.append( - f"'{top_overspent['category']}' has been overspent in {top_overspent['months_overspent']} of the last {months_back} months" - ) - - if trend == "increasing": - recommendations.append( - "Your spending trend is increasing. Consider reviewing your budget allocations." - ) - - return { - "analysis_period": f"Last {months_back} months", - "average_monthly_spending": round(avg_spending, 2), - "current_month_spending": round(current_spending, 2), - "spending_trend": trend, - "frequently_overspent_categories": frequently_overspent, - "recommendations": recommendations, - "monthly_breakdown": monthly_data, - } diff --git a/uv.lock b/uv.lock index cb0ebf9..8fa9f44 100644 --- a/uv.lock +++ b/uv.lock @@ -2627,12 +2627,12 @@ dependencies = [ { name = "python-dotenv" }, { name = "quart" }, { name = "quart-jwt-extended" }, + { name = "sqlglot" }, { name = "tavily-python" }, { name = "tomlkit" }, { name = "tortoise-orm" }, { name = "tortoise-orm-stubs" }, { name = "twilio" }, - { name = "ynab" }, ] [package.optional-dependencies] @@ -2683,12 +2683,12 @@ requires-dist = [ { name = "python-dotenv", specifier = ">=1.0.0" }, { name = "quart", specifier = ">=0.20.0" }, { name = "quart-jwt-extended", specifier = ">=0.1.0" }, + { name = "sqlglot", specifier = ">=30.22.0" }, { name = "tavily-python", specifier = ">=0.7.17" }, { name = "tomlkit", specifier = ">=0.13.3" }, { name = "tortoise-orm", specifier = ">=0.25.1,<1.0.0" }, { name = "tortoise-orm-stubs", specifier = ">=1.0.2" }, { name = "twilio", specifier = ">=9.10.2" }, - { name = "ynab", specifier = ">=1.3.0" }, ] provides-extras = ["test"] @@ -2955,6 +2955,15 @@ asyncio = [ { name = "greenlet" }, ] +[[package]] +name = "sqlglot" +version = "30.22.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/94/e0/db58fbf2527426758dc1e862ce538736978e100e4e78fc9657e9661826ee/sqlglot-30.22.0.tar.gz", hash = "sha256:ec4b83ca8236ea8867f574a382dc15ce35b071c977fecfcc66482d9a3f500661", size = 6088770, upload-time = "2026-10-09T16:09:01.04Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b4/4c/b8474b02b572d9c7a2903e364335d566d52b6128b834b92a7cdfe5597823/sqlglot-30.22.0-py3-none-any.whl", hash = "sha256:90aa461490fcd95d14ec3842a97506ae20f6d3e9313307ad31be793d479cca65", size = 777816, upload-time = "2026-10-09T16:08:59.07Z" }, +] + [[package]] name = "sqlite-vec" version = "0.1.9" @@ -3531,24 +3540,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/73/ae/b48f95715333080afb75a4504487cbe142cae1268afc482d06692d605ae6/yarl-1.22.0-py3-none-any.whl", hash = "sha256:1380560bdba02b6b6c90de54133c81c9f2a453dee9912fe58c1dcced1edb7cff", size = 46814, upload-time = "2025-10-06T14:12:53.872Z" }, ] -[[package]] -name = "ynab" -version = "1.9.0" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "certifi" }, - { name = "pydantic", version = "2.11.9", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.14'" }, - { name = "pydantic", version = "2.14.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.14'" }, - { name = "python-dateutil" }, - { name = "typing-extensions", version = "4.15.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.14'" }, - { name = "typing-extensions", version = "4.16.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.14'" }, - { name = "urllib3" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/9a/3e/36599ae876db3e1d32e393ab0934547df75bab70373c14ca5805246f99bc/ynab-1.9.0.tar.gz", hash = "sha256:fa50bdff641b3a273661e9f6e8a210f5ad98991a998dc09dec0a8122d734d1c6", size = 64898, upload-time = "2025-10-06T19:14:32.707Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/b2/9c/0ccd11bcdf7522fcb2823fcd7ffbb48e3164d72caaf3f920c7b068347175/ynab-1.9.0-py3-none-any.whl", hash = "sha256:72ac0219605b4280149684ecd0fec3bd75d938772d65cdeea9b3e66a1b2f470d", size = 208674, upload-time = "2025-10-06T19:14:31.719Z" }, -] - [[package]] name = "zstandard" version = "0.25.0"