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..e42c9a0 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 that also marks rows missing from the download as deleted); `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`, `ynab_service.py`, `s3_client.py`, `simbakit_service.py` - integrations with Obsidian notes, Mealie recipes, YNAB budgets (API client for the `blueprints/ynab/` mirror), 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 27d5812..1730df5 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: deploy redeploy build up down restart logs migrate migrate-new migrate-history frontend desktop desktop-dev test test-cov reindex reindex-obsidian +.PHONY: deploy redeploy build up down restart logs migrate migrate-new migrate-history frontend desktop desktop-dev 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 1fa4fa5..ecbec34 100644 --- a/app.py +++ b/app.py @@ -32,6 +32,8 @@ from blueprints.users.decorators import jwt_access_required 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 @@ -46,7 +48,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) @@ -112,6 +114,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 ) @@ -121,14 +125,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..de39457 --- /dev/null +++ b/blueprints/ynab/sql_tool.py @@ -0,0 +1,172 @@ +"""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 utils.mcp_tools import admin_only_message + +from . import models +from .sync import ensure_fresh +from .tools import NOT_SYNCED, ynab_tool_errors + +MAX_ROWS = 100 +# Tool results are cut at 4000 chars in blueprints/conversation/logic.py +MAX_OUTPUT_CHARS = 3800 +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.""" + return _check(query)[0] + + +def _check(query: str) -> tuple[str | None, str | None]: + """Validate ``query``; return ``(error, sql_to_run)``. + + The SQL that runs is regenerated from the validated parse tree (minus + comments), so Postgres executes exactly what was checked. + """ + try: + statements = [ + s + for s in sqlglot.parse(query, read="postgres") + # a trailing "; -- comment" parses as a bare Semicolon + if s and not isinstance(s, exp.Semicolon) + ] + except sqlglot.errors.SqlglotError as e: + return f"Could not parse SQL: {e}", None + if len(statements) != 1: + return "Exactly one SQL statement is allowed.", None + root = statements[0] + if not isinstance(root, exp.Query): + return "Only SELECT queries are allowed.", None + 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.", None + + # 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}", None + 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.", None + if table.name not in ALLOWED_TABLES: + return ( + f"Table '{table.name}' is not allowed. " + f"Available: {', '.join(sorted(ALLOWED_TABLES))}", + None, + ) + + 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.", None + return None, root.sql(dialect="postgres", comments=False).rstrip("; ") + + +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(columns: list[str], rows: list[tuple]) -> str: + if not rows: + return "Query returned no rows." + lines = ["\t".join(columns)] + lines += ["\t".join(_format_value(v) for v in row) for row in rows] + text = "\n".join(lines) + if len(text) > MAX_OUTPUT_CHARS: + text = text[:MAX_OUTPUT_CHARS] + "\n...(output truncated; aggregate more)" + elif len(rows) == MAX_ROWS: + text += f"\n(capped at {MAX_ROWS} rows; aggregate or add LIMIT)" + return text + + +@tool +@ynab_tool_errors +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 admin_only_message("YNAB SQL") + error, sql = _check(query) + if error: + return error + if await ensure_fresh() is None: + return NOT_SYNCED + + 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}'" + ) + # Raw records rather than dicts, which would merge same-named + # columns (e.g. c.name and g.name) + _, records = await conn.execute_query( + f"SELECT * FROM ({sql}) AS q LIMIT {MAX_ROWS}" + ) + except Exception as e: + return f"SQL error: {e}" + columns = list(records[0].keys()) if records else [] + return format_rows(columns, [tuple(r.values()) for r in records]) diff --git a/blueprints/ynab/sync.py b/blueprints/ynab/sync.py new file mode 100644 index 0000000..3ddb27e --- /dev/null +++ b/blueprints/ynab/sync.py @@ -0,0 +1,640 @@ +"""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 +import time +from dataclasses import dataclass +from datetime import date, datetime, timedelta, timezone +from decimal import Decimal +from typing import Any, Callable + +from tortoise.models import Model +from tortoise.transactions import in_transaction + +from utils.ynab_service import YNABService + +from .models import ( + YnabAccount, + YnabCategory, + YnabCategoryGroup, + YnabChange, + YnabMonth, + YnabMonthCategory, + YnabPayee, + YnabSubtransaction, + YnabSyncState, + YnabTransaction, +) + +logger = logging.getLogger(__name__) + +FULL_RESYNC_INTERVAL = timedelta(hours=24) +# After a failed sync, tools answer from the mirror without retrying YNAB +FAILURE_BACKOFF_SECONDS = 60 +# Cap on rows passed to a single ``id IN (...)`` lookup +LOOKUP_CHUNK = 2000 + +_sync_lock = asyncio.Lock() +_last_failure = float("-inf") + + +def ynab_configured() -> bool: + return bool(os.getenv("YNAB_ACCESS_TOKEN")) + + +# YNAB accepts these in place of a budget id; they can point at different +# budgets over time, so the mirror resolves them once and keeps the real id +BUDGET_ID_ALIASES = ("last-used", "default") + + +def _configured_budget_id() -> str | None: + """YNAB_BUDGET_ID if it names a specific budget, else None.""" + budget_id = os.getenv("YNAB_BUDGET_ID", "").strip() + return None if budget_id in ("", *BUDGET_ID_ALIASES) else budget_id + + +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"], + } + + +# Filled in from payees/categories/accounts after each sync, not from YNAB +DENORMALIZED_FIELDS = ("account_name", "payee_name", "category_name") + + +@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=DENORMALIZED_FIELDS), + ), + "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 [], + } + + +# Id fields that reference a named entity (account, payee, category, group); +# the change log records these by name +NAMED_ID_FIELDS = ( + "account_id", + "payee_id", + "category_id", + "category_group_id", + "transfer_account_id", +) + + +class _Labeler: + """Resolves ids to names for change log labels and diffs.""" + + def __init__(self, names: dict[str, str]): + self.names = names + + def name(self, entity_id: str | None) -> str | None: + return None if entity_id is None else self.names.get(entity_id, entity_id) + + def __call__(self, entity_type: str, row: dict) -> str: + if entity_type == "transaction": + payee = self.name(row.get("payee_id")) or row.get("import_payee_name") or "" + return f"{row['date']} {payee} {row['amount']:+.2f}" + if entity_type == "subtransaction": + category = self.name(row.get("category_id")) or "" + return f"split line {category} {row['amount']:+.2f}" + if entity_type == "month_category": + return f"{self.name(row['category_id'])} {row['month']:%Y-%m}" + if entity_type == "month": + return row["id"][:7] + return row.get("name") or row["id"] + + def diff_entry(self, field: str, old: Any, new: Any) -> tuple[str, list]: + """Log e.g. ``category_id`` changes as ``category`` names, not ids.""" + if field in NAMED_ID_FIELDS: + return field.removesuffix("_id"), [self.name(old), self.name(new)] + return field, [_jsonable(old), _jsonable(new)] + + +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 _fetch_existing(spec: EntitySpec, ids: list[str], conn, fields) -> dict: + 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", *fields) + ) + existing.update({row["id"]: row for row in rows}) + return existing + + +async def _apply( + spec: EntitySpec, + raw_rows: list[dict], + conn, + server_knowledge: int, + log_changes: bool, + labeler: _Labeler, + complete: bool = False, +) -> int: + """Upsert rows that differ from the mirror; return how many changed. + + ``complete`` means ``raw_rows`` is every live entity of this type (a full + download). Full downloads omit deleted entities, so mirror rows missing + from it are marked deleted here; otherwise deletions that happened right + before a full sync would never reach the mirror. + """ + incoming = {row["id"]: row for row in map(spec.mapper, raw_rows)} + # Only YNAB-sourced columns; denormalized names are filled in afterwards + field_names = tuple(f for f in _fields(spec.model) if f not in DENORMALIZED_FIELDS) + existing = await _fetch_existing(spec, list(incoming), conn, field_names) + + to_write: list[Model] = [] + changes: list[YnabChange] = [] + + def log(entity_id: str, action: str, diff: dict, row: dict) -> None: + changes.append( + YnabChange( + server_knowledge=server_knowledge, + entity_type=spec.entity_type, + entity_id=entity_id, + action=action, + label=labeler(spec.entity_type, row)[:500], + changes=diff, + ) + ) + + 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 + diff = dict( + labeler.diff_entry(f, None, new[f]) + for f in spec.logged_fields + if new[f] is not None + ) + # e.g. a month rollover creates every category's month row with + # nothing assigned; that's not an edit anyone made + if not any(values[1] for values in diff.values()): + continue + log(entity_id, "created", diff, new) + else: + diff = dict( + labeler.diff_entry(f, old[f], new[f]) + for f in spec.logged_fields + if _norm(old[f]) != _norm(new[f]) + ) + if diff: + deleted = diff.get("deleted") == [False, True] + log(entity_id, "deleted" if deleted else "updated", diff, new) + + removed = 0 + if complete: + live_ids = ( + await spec.model.filter(deleted=False) + .using_db(conn) + .values_list("id", flat=True) + ) + missing = [i for i in live_ids if i not in incoming] + if missing: + if log_changes and "deleted" in spec.logged_fields: + gone = await _fetch_existing(spec, missing, conn, field_names) + for entity_id, row in gone.items(): + log(entity_id, "deleted", {"deleted": [False, True]}, row) + for i in range(0, len(missing), LOOKUP_CHUNK): + removed += ( + await spec.model.filter(id__in=missing[i : i + LOOKUP_CHUNK]) + .using_db(conn) + .update(deleted=True) + ) + + 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) + removed + + +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""", +) +# Entity types whose changes can affect the denormalized name columns +NAME_SOURCES = ("accounts", "payees", "categories", "transactions") + +REFERENCE_KEYS = ("accounts", "payees", "category_groups", "categories") +DATA_KEYS = ("months", "month_categories", "transactions", "subtransactions") + + +async def apply_budget( + budget: dict, + server_knowledge: int, + conn, + log_changes: bool, + full: bool = False, +) -> dict[str, int]: + """Write a full or delta budget payload into the mirror. + + With ``full``, entity types present in the payload are treated as + complete and anything missing from them is marked deleted. + """ + entities = _budget_entities(budget) + present = {key for key in SPECS if budget.get(key) is not None} + if budget.get("months") is not None: + present.add("month_categories") + + labeler = _Labeler({}) + if log_changes and any(entities.values()): + labeler.names = await _load_names(conn) + + counts: dict[str, int] = {} + for key in REFERENCE_KEYS + DATA_KEYS: + counts[key] = await _apply( + SPECS[key], + entities[key], + conn, + server_knowledge, + log_changes, + labeler, + complete=full and key in present, + ) + if key in REFERENCE_KEYS: + # Later labels (e.g. a new payee on a new transaction) need these + labeler.names.update({row["id"]: row["name"] for row in entities[key]}) + + if any(counts[key] for key in NAME_SOURCES): + for sql in DENORMALIZE_SQL: + await conn.execute_query(sql) + return counts + + +async def _current_state() -> YnabSyncState | None: + """The sync state for the mirrored budget. + + The mirror holds a single budget, keyed by its real YNAB id. With + ``YNAB_BUDGET_ID`` unset or an alias, the first sync resolves "last-used" + and the mirror stays pinned to that budget, so switching budgets in YNAB + can't mix two budgets' rows. + """ + configured = _configured_budget_id() + states = await YnabSyncState.all().order_by("budget_id") + ids = [s.budget_id for s in states] + if len(ids) > 1 or (configured and ids and ids[0] != configured): + raise ValueError( + f"YNAB mirror holds budget(s) {', '.join(ids)} but YNAB_BUDGET_ID " + f"is {configured or 'unset'}; clear the ynab_* tables to switch budgets" + ) + return states[0] if states else None + + +def _backing_off() -> bool: + return time.monotonic() - _last_failure < FAILURE_BACKOFF_SECONDS + + +def _is_fresh(state: YnabSyncState | None, max_age_seconds: int) -> bool: + return ( + state is not None + and state.last_synced_at is not None + and datetime.now(timezone.utc) - _norm(state.last_synced_at) + <= timedelta(seconds=max_age_seconds) + ) + + +async def sync_ynab(full: bool = False, max_age_seconds: int | None = None) -> dict: + """Sync YNAB into Postgres. Returns per-entity counts of changed rows. + + ``max_age_seconds`` (on-demand syncs from tools) skips the sync if the + mirror is already that fresh or YNAB failed recently. Both are checked + under the lock, so parallel tool calls share one sync or one failure. + + The YNAB request runs before the DB transaction. An asyncio lock + serializes syncs in this process; a Postgres advisory lock plus a + server_knowledge check discards a payload if the other blue/green + container synced in the meantime. + """ + global _last_failure + async with _sync_lock: + state = await _current_state() + if max_age_seconds is not None and ( + _is_fresh(state, max_age_seconds) or _backing_off() + ): + return {"skipped": True} + + now = datetime.now(timezone.utc) + base_knowledge = state.server_knowledge if state else None + first_sync = base_knowledge is None + full = ( + full + or first_sync + or state.last_full_sync_at is None + or now - _norm(state.last_full_sync_at) > FULL_RESYNC_INTERVAL + ) + budget_id = state.budget_id if state else _configured_budget_id() + budget_id = budget_id or "last-used" + + try: + data = await YNABService().fetch_budget( + budget_id, None if full else base_knowledge + ) + budget = data["budget"] + knowledge = data["server_knowledge"] + if state is not None and budget["id"] != state.budget_id: + raise ValueError( + f"YNAB returned budget {budget['id']} for {state.budget_id}" + ) + async with in_transaction() as conn: + await conn.execute_query( + "SELECT pg_advisory_xact_lock(hashtext('ynab_sync'))" + ) + latest = await YnabSyncState.get_or_none( + budget_id=budget["id"] + ).using_db(conn) + if (latest.server_knowledge if latest else None) != base_knowledge: + logger.info("YNAB sync skipped: mirror changed during fetch") + return {"skipped": True} + + counts = await apply_budget( + budget, knowledge, conn, log_changes=not first_sync, full=full + ) + await YnabSyncState.update_or_create( + budget_id=budget["id"], + defaults={ + "budget_name": budget.get("name"), + "server_knowledge": knowledge, + "last_synced_at": now, + "last_error": None, + **({"last_full_sync_at": now} if full else {}), + }, + using_db=conn, + ) + except Exception as e: + _last_failure = time.monotonic() + if state is not None: + await YnabSyncState.filter(budget_id=state.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, and YNAB isn't retried for + ``FAILURE_BACKOFF_SECONDS``. Returns None only if the mirror has never + been populated. + """ + state = await _current_state() + if not _is_fresh(state, max_age_seconds) and not _backing_off(): + try: + await sync_ynab(max_age_seconds=max_age_seconds) + except Exception: + logger.exception("YNAB sync failed; answering from existing mirror") + state = await _current_state() + 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..86b60ee --- /dev/null +++ b/blueprints/ynab/tools.py @@ -0,0 +1,505 @@ +"""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 functools +import json +import logging +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 + +logger = logging.getLogger(__name__) + +NOT_SYNCED = "YNAB data is unavailable: the budget has not been synced yet." + +# Keep in step with TOOL_RESULT_MAX_CHARS in blueprints/conversation/logic.py; +# longer results get truncated there anyway and crowd the 32k context. +MAX_TRANSACTIONS = 40 + +# ynab_recent_changes accepts the names it prints as well as stored types +ENTITY_TYPE_ALIASES = { + "assignment": "month_category", + "transfer": "transaction", + "split": "subtransaction", + "group": "category_group", +} +ENTITY_TYPES = ( + "transaction", + "month", + "subtransaction", + "month_category", + "category", + "category_group", + "payee", + "account", +) + + +def _singular(word: str) -> str: + if word.endswith("ies"): + return word[:-3] + "y" # categories -> category + return word.removesuffix("s") + + +def ynab_tool_errors(fn): + """Return errors to the model as text: LangGraph's ToolNode re-raises + anything else, which would fail the whole chat turn.""" + + @functools.wraps(fn) + async def wrapper(*args, **kwargs): + try: + return await fn(*args, **kwargs) + except Exception as e: + logger.exception(f"{fn.__name__} failed") + return f"YNAB error in {fn.__name__}: {type(e).__name__}: {e}" + + return wrapper + + +# 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 +@ynab_tool_errors +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 +@ynab_tool_errors +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 +@ynab_tool_errors +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 = 20, +) -> 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 20, max 40) + """ + 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), MAX_TRANSACTIONS) + 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 +@ynab_tool_errors +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 +@ynab_tool_errors +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 +@ynab_tool_errors +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, assignment, category, payee, account, month + """ + if await ensure_fresh() is None: + return NOT_SYNCED + entity_type = _singular(entity_type.strip().lower().replace(" ", "_")) + entity_type = ENTITY_TYPE_ALIASES.get(entity_type, entity_type) + if entity_type and entity_type not in ENTITY_TYPES: + return f"Unknown entity_type '{entity_type}'. Use one of: transaction, assignment, category, category_group, payee, account, month." + 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 40""", + 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 40)" if len(rows) == 40 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 745ef1e..2e1a26a 100644 --- a/config/db.py +++ b/config/db.py @@ -20,6 +20,7 @@ TORTOISE_CONFIG = { "blueprints.cron.models", "blueprints.beli.models", "blueprints.artifacts.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/18_20261012180707_add_ynab_mirror.py b/migrations/models/18_20261012180707_add_ynab_mirror.py new file mode 100644 index 0000000..6accb2b --- /dev/null +++ b/migrations/models/18_20261012180707_add_ynab_mirror.py @@ -0,0 +1,259 @@ +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"); + ALTER TABLE "users" ALTER COLUMN "ldap_groups" TYPE JSONB USING "ldap_groups"::JSONB; + ALTER TABLE "beli_restaurants" ALTER COLUMN "tags" TYPE JSONB USING "tags"::JSONB; + ALTER TABLE "beli_restaurants" ALTER COLUMN "cuisine" TYPE JSONB USING "cuisine"::JSONB; + ALTER TABLE "artifacts" ALTER COLUMN "data" TYPE JSONB USING "data"::JSONB; + ALTER TABLE "artifacts" ALTER COLUMN "source_ref" TYPE JSONB USING "source_ref"::JSONB;""" + + +async def downgrade(db: BaseDBAsyncClient) -> str: + return """ + ALTER TABLE "users" ALTER COLUMN "ldap_groups" TYPE JSONB USING "ldap_groups"::JSONB; + ALTER TABLE "artifacts" ALTER COLUMN "data" TYPE JSONB USING "data"::JSONB; + ALTER TABLE "artifacts" ALTER COLUMN "source_ref" TYPE JSONB USING "source_ref"::JSONB; + ALTER TABLE "beli_restaurants" ALTER COLUMN "tags" TYPE JSONB USING "tags"::JSONB; + ALTER TABLE "beli_restaurants" ALTER COLUMN "cuisine" TYPE JSONB USING "cuisine"::JSONB; + DROP TABLE IF EXISTS "ynab_category_groups"; + DROP TABLE IF EXISTS "ynab_month_categories"; + DROP TABLE IF EXISTS "ynab_accounts"; + DROP TABLE IF EXISTS "ynab_sync_state"; + DROP TABLE IF EXISTS "ynab_categories"; + DROP TABLE IF EXISTS "ynab_months"; + DROP TABLE IF EXISTS "ynab_subtransactions"; + DROP TABLE IF EXISTS "ynab_changes"; + DROP TABLE IF EXISTS "ynab_payees"; + DROP TABLE IF EXISTS "ynab_transactions";""" + + +MODELS_STATE = ( + "eJztXQtv4za2/iuEscBOATc3j0lmmrtYwEncadq8MPHc7uxkoKEl2uZGprSilNQo+t8vSU" + "m2HpRs2ZIteghst43EQ8kfKfKc7zz4Z2fqWMimB5cOeUEehT52SOcc/NkhcIrYf0jvd0EH" + "uu7iLr/gw6EtBMxES3EHDqnvQdNnN0fQpohdshA1PexGDyOBbfOLjskaYjJeXAoI/m+ADN" + "8ZI3+CPHbjy1d2GRML/YFo/Kf7bIwwsq3Ue2OLP1tcN/yZK659+nR99bNoyR83NEzHDqZk" + "0dqd+ROHzJsHAbYOuAy/N0YEedBHVuJn8LeMfnZ8KXxjdsH3AjR/VWtxwUIjGNgcjM4/Rg" + "ExOQZAPIn/39t/dirAw6Dm0GLicyz+/Cv8VYvfLK52+KMuf+l9fHNy9oP4lQ71x564KRDp" + "/CUEoQ9DUYHrAkjx7xyUlxPoyaGM22fAZC+6DozxhQWOizkUAxkDtB5qnSn8w7ARGfsT9u" + "fx6WkJjP/X+yiQZK0ElA6b1+Gsv4tuHYf3OKQLCE0P8Z9sQD8P5BW74+MpkoOZlsxAakWi" + "B/F/tBRg9huse2LPoo+gBN/B9W3/cdC7feC/ZErpf20BUW/Q53eOxdVZ5uqbs8xQzDsBv1" + "8PfgH8T/Dv+7t+du7P2w3+3eHvBAPfMYjzakAr8b3GV2NgUgMbuNaaA5uW1AO704GNXj7x" + "wU4gIciusuwlRNZa+aK3qjR8nVc03GC/yKx8h6ssfIfF695hdtkLKPKMavtwQmSDzXgdLD" + "f7FNbferkCM3qW7rwcjDx4PzsewmPyG5oJCK/Ze0BiyrbcSHX7FHXTVtAWVxdv4cHXuU6X" + "nBTs17HfhPzwY+w9Xvau+h2B4RCaz6/Qs4wCMKeIUjhGNA/oRST5828fkT1XcOVYJtXg27" + "DHdq7NRdimPlDo+XjENPMNQelF3SiGBJ81zrGTmC2peZS/NT2eZq9AwmaAFT2bP6lkniyx" + "qhLTaTXjykhOaW1kKW1k+egPiQo5YFflUMbtVTGyylTD/r8GKa0wVine3Pb+9UNKM7y5v/" + "sQN0+oIJc39xfa4tpHxTxvcVEXwWeZWsQ18z4JpjmlKDW6CfEdfzqdT4/9j+eA6zZP5PH6" + "9qJ3DiieDuETGdzf3zA0HEdYE1X1+KNV9PijYj3+KKfH4ynbZYxnNKtiD6WEtmYR1UkFnR" + "4dr4Aka1UIpbiXxjLasw1mLVp2JV4tL6kkqs0QbEm1qJpSIhGtU0NR1PQ0M3T/hiZo1nvQ" + "WhSX2qKS6bK+Tbp9+6s9Rn6j5henPG7R1PFmHYnVlbjbLTO2BO8w5Q2xNrL2wMhiD/ERqW" + "RnJUS0qaVNLYlys7+mlnZu7cXA5ljfnbll2rUYquKXabEyvIFjpknt7/766vLRh76Ucl/c" + "LNX9HGyZBuXtVtP8Og+IWAxTwPsHtjPG5BxMoUsBU/fAN9HTN+BCD06B7wDsU/Dw22UfMG" + "MCs5/lHWSpno07fCJP5NFnk9ICmICHaIaDN8Txges5JqIUCOV29gOgDoDhM1jv0Aesd491" + "yPQ5h6AnMrQD9D9MGhHA1gTHA5BYYIQJphMkWvFXcrheyySxbYMXaGO+vFPxszbQfWk8Uq" + "uyJHOBetS15bpvndzI0fH7Vbi74/fF5B2/V64Fu88mMuJZUgXanKAqGnFjIC9ARX+4mH1d" + "a+hKacnt6Eo1T2JFVKMYh1KlV1sze6H0hgOb0+F2o48IRa+Ah1rOQGnmSX3miY9j1TjqpI" + "yaO20jrh4XUvrqeJKZeYEJ9GYFuktCKgPmcBbp+G2lzmXgXXAeLrnwMsguru96Hz/Lqby4" + "fXLeXnwe9HtZPWYKcaWw17mAkvr24Wq+8jJnec5b/soMKMqWdIME02E1DVsiWouPd7uonq" + "wC6kkxpieSAITI910dUomohjT+bg1E+M+TraWOYyNISr75pGx2RY0iV5rYn6pqPxWW1Pv7" + "m9SSenGd9X58ur3os/VAwMwa4ZDtur4bSMGdTKHJ1K5nJPGgL1lQM7IKztijs1VW1qw9kF" + "hYz7IzNmTkguF/kCmxzYoBzcopCGYjihSzliaG6zkv2Kq2pOYEt6eddmzHhPYG6n0mwmuV" + "ZfW0eFk9zS2rtgVdY+w5gSuJ6fj18f5ODmlGLAPoJ8J+5RcLm34X2Jj6XxuD98vXZkwn/s" + "vLHc9ZH3NGV+UdaMfzHlM12vG8dwM7j/JakYKTRx/WmLDVZiO/PFErGYS2PhbpuLf2zfyV" + "oAj1Y2iaTkA2DZzs8756YVcKQ0LNCbICZosZ9SQ5Psb9qZ/hOEQ25rrqCNubonLBunoIe1" + "IYEJ3y2ZSXRywmHYmbpx9TpMV+HrGorRhoIroD0ZfO/u1D7nYQIRlRtGo+lmQFmXm4COVB" + "JRRZQLwUEA1fsT8BJ4c/WnAGPMTbs5c5ADeYPFMefXI58ZwpvLp4IiPHAy/M0mb/ogh65g" + "SY0IVDbGMfbx4Ooh1SNTmkYlZUBujyFCQpsEow/o0wKWuQUpvyUTv37Z2u5Do5LXGdnOZd" + "JyO2inBzgs1jyf5UjGZWTk1IG5mbzBorBLQkvdxREc5tZz5Y0tjEcoYillGTm1CEi1gp0m" + "voWDOjat2FlJAi+bfb/iwERhN/KokjWAJsLKSBlWdaCS2bPZG9WkXNTSKqCMi6PKPmm9d1" + "JKgVla036+KRjPjeimlraSmduRYDksewcvKautR5N5PElp4kbcpjS2FcRCwmxmAJv5hymq" + "zKM0YynCYc4XHgCV4YcJbv+rb3wC8TJOguWsQ6VughwUHOpTxkcbqRvWPIQiJiejOXTTQQ" + "x/R2E30AinyfPZ52nwhnN+NueFpYQA/AY+C6Dk9tm7Iphl0bxS0ocJEnqjFpjrIlHGU0aa" + "vTQTlBzVTGjAWmrg1nRtVchKycth0StdFcY8ImdhU4U0KqsGvbApOvz3kwr0kBcZCSyWCJ" + "m9NNfvrpZAMoRTXYH4+P3r57+/7k7O171kS8yvzKuxJw8/HGAoN1coxygnoyZiZjYbZRP9" + "ZDijmtXA+qoLttYgszQ5vpIy+SuVuaiJCS22ISwnzjb3EOgg2pbyDPcySx3cVTNi2lyDa/" + "dR5W84X7yRfqwOO9GFhd8UpXvGpfxasMsVFH6LJaWKZjpmYkrK8V1AHEI+vtcd6ZQpA0Th" + "8nkClikNPgLSORMwO3nEcesNvPFHA5XneMLzpURJ6KUl2CdkVJojjPI1fugfPItxxm9g8F" + "XKcNhfney2Smbje8GJVBQ9Y8NJathBFzPGIdBh4C/Nc98zpsvgNoyB4DzGw/NEWsfzvsmA" + "sIrRmI6uysuWaRW8IiC4tGTNp1wrXy0jXodu2ykFqkyq3kChZDEkciB7LvpJArlIlujzI8" + "3EA1r5kwjDEo8AYXApiT+y7R490jM+DEkxFtExIlphDEIvHvEkvxSUYorL9CZzvQi/SOF2" + "nNpOwnk6LjsPYsDqvpoAb1o7By2dASO1qWMV1sSMvztXWJT6XtTA+Z2MXS42WK/e8pIVW8" + "w9uIyC87f335KY+bnsVeH6gdvuM+9j70z0FcifCJ9G971zfnYF46c/cnteuTkZrwE+8gL3" + "gvY+wW+2V1cyIrq6ZBoYgBsZJpWOT0WfEA37n0FqsORlpd6CQ5eOjfXV3ffchrIJ3ozjlw" + "w7Nbnshj/25wDihbKJ/Iz2zV71+dC88Csp7IZe/usn8jLpn8t9pRNdXdbwdMLQk8D0X4rz" + "NM6R62OFQf5w8+uIsmcGaQ+GUGrDjY5ooNyedzYLERmT2R3/v93/ifrwg9879v7+8Gv/AL" + "U7bXTexZOwYnoMhgk1G2W5cGy6XkdMXeTMVe7ryLzbEqalBOUJFtW8fMaUJQx8zpgS1keu" + "PPbjireqB6VlDzvQlMaqB81Y+iy02RNvG9l55DfnWGHQnNG9/qlrG7Jmtk/McZalJXfVK3" + "aj6VTqPKp5wm3qyCXp0R01q1VKt2PWfqVqJtFxKqzNFtY6odOdqRo6wjZ80DnXZylJMyWZ" + "ReQNYwbzOiOiStDXHDZc6HsgjwzbwOu/bKNXDOkc4u1kypJtQ0U/r9DWzVY42a5KqSZ5JI" + "+KrMkSXFnFXumJR6easv8wRi8aBQX/6q2ax19pwN2KwE+BUUoLSUtm0zZwttUiAo14Mq6G" + "5bJxJAFVZxWDJ3i0s5tFXl3Nbk9dCIrR6TokOKV53BuW4UgXnbszhkBEKw1tIDpR1oYqEN" + "xIK2hZv4YjSdWS+dqbmFvTBBNbewpwOrK5fpymVtrlzm8fJJgQc3PnmZM1Mf552phWyjRb" + "sywBSQemnolvB6mUGrm9pzFwSjeNwwoJggSqPJlvC4eRgRy8ja5OJvk438mB9MrlnBnbCC" + "mUGrxK9kZFXhsLZAshTM+FXRLRDfYjbVBpN1e/BWDdLMiCliem+LE5yvxFWXgaSgilP0aJ" + "UgiaPiIImjXJDE9xU93MiJ0Cb2K83EuL3+qOeTkBk6k6HjTRyn0uaTldOIzudkgLnCkwfz" + "18f7u4JpuRDJ4PiJsB/4xcKm3wU2pv7XxhbKhTY6DLDtY0IP+AMbUkg5FuV0cJb5zWiuvI" + "N8uDs2kcEsGVmObvFUzogpOZPr353Uq8MgD8y+6PfvzsEQIfJEfu/dDYzBvTH4yNP4mcXL" + "bFXD91qSuc9m4LMhfmAe9cJqoWmhtWqE7mDu1lwllJqOJ6sMikw8hbYcuLlMltMOhQ4i4V" + "biVwLOVf/y+rZ38+a0e5xx98Rz8m1eEXV8WYXaYufkXECRtXLbfkkfjiV4Fu//cXu9+a+/" + "+Y+Qb64XNpGW1C6udrm4tE96L8Y175OOHAMV3ZdpKe3BTPpXNnRiZsLBW4vfUl9meo60qX" + "hEz/PxKHSu5Rx383vdMpcdjFqteLpOD7AWgekHHrKA7zg2YBOLjQV4I1KSmcQwsMaIKRhc" + "p3sKjg+Pz35gbdgYcglMgD9BIH4ocCFBdv4EnsaeootcbGd96ZY4AP0oKG5VOiVuryZN3Q" + "gl+MzevgqEcXs1ITx7uwKCZ28LAeS3MvYU9mU7XMkcjAXURLCZSivsA69ilMbtd2WUNqZr" + "NWJ+UifwOIuMRlUwTkvVgHSruJVGgH5BHpUWCyqkSBMS2ztD6WiDtaDu86i0Bb2fFrSO6t" + "6Lgc1TXg7ha5aIka1aXTMv+r1wJLIDDKuBl5baALdW7cIVYNPpBDqdYEvpBEXrXQ3oXWa6" + "UxdFyWq+HM3CKvUbAZk4uayti99SONPrewrJx/4A3H26udkVMfyZwGF8TJ+EG07e7pbRwz" + "PW0IgOy2tZheFitqbOuPWmDyxMEzUnZyvwNCdZTXJB0/BbusRww8SXwKAChHF7NSGsv5gf" + "wyB03+RBLM2zT8ltMdO+Mb2wzkx726GVCxcshDSauQC2PJbl8Wtrf+GtMu6aKacEbSg93K" + "s0oDIhtXFIZbtmahxTeXTaPVk5qJLhCz1+iMVaYEqkNaj81TeDVSqvgZ2XjmJPMfF6J2nK" + "e9DVp3ZcfSo0catqGgkpfRSghPHbHUlwGafRFrAE8/vdpTRBlJGL23a+vCYK1iEK4vxqY+" + "w5gVuxlIBUuCn7VwmgNQNTIwMzwZYlqyBaugcthLS1m0KTPQ15BEqCMkvxTIrpTV0TCA0S" + "CGMH2kZV2jUlpAiwTacEh5hAT0q+llq6Gcn9THKtZOImEDHEie1yC3cpnAvhMuNWOUiZsZ" + "o7mlGbjvtoOn7gOn5nif0YNuquakRGloO2JJUxcLTLWRs8TW4o2uBpM6R6b9+zvX0iylcV" + "beqTuLjVst1cNFw1t9h1EbF+dIg9A7YzBs4IvE6gDxA0J4DOiAnC7qxuWFUEDGfAwqMRGy" + "wAx2x8qC/yfqeYH3MhySqut/8n8kQ+3/Uu/k5B7+EaTCAFxAETTH2mvnQBdVhjTAH7HxcS" + "T+XuLM/iD0YW9il4gw7GBwCyjiCleEymiPhg7PAHjjxnCv729vAQ+A74Gy9VeACufUCZzc" + "AEYfgmI+yxd+Kv3kR28wUeF+ZASfUdSfpTpLrsVN0J059+Oj4+OXl3fHhy9v707bt3p+8P" + "53lQ+VtlCVEX1x/455r6rOMFsVgJ4oO0lh80JbidzJq6B2BfEmvy3lCKvBfkGc/svo0sWZ" + "B02Uckk95eRqE6X1XygB4f+7PKVGBGTE1H1Cpk4EkxGXgiObxdoFLN0E0JKQlk/eUEoCnP" + "NClGcSGhppFbf3lQ9tNQpbIgcwE1EWymfPVC3U7DWFIqeCGiCzNkNYS4XkCLDLNb4SwosM" + "tuY0/CErNMeBw0t9r4VlPXMllsVlR2PNXlbNr+Ll7N2aQ98LV64DExHRmLX+ozXgjpkGi2" + "B4jcJRlHWp4GkRDTMIaq84v0mI5SGJNiGkZeVc8YImPNOZkX1pCyKTZGhjPiyhWSzM5CFi" + "gr9p3W3NcepM5eeZCELbIsuyDdqLua4VI10aDzgLwfhSSIo0sAnIqKBgfg27d4Gfv2jftq" + "hEeIu3VYW9um4KnTE64ZZD118i6lWnvWBtVSg6p0TZmvzO8KDSp+SxtUaxlU88CsNTNBFG" + "aM688B0baAtgXaA6POz68BRK2/7pf++gBnCBXpreHN5fqqy9tpol2ZHVoHMTfonWRPI3SE" + "vLieXEVNskBcEcK+aY1Sbz/7tf08BkMx4aNIkYJ9KNNq+YZEUwIr8ic3mCDKo1YhoC5DDC" + "S6+F8RiOpCDxH/73RBg2BOcDzy1jLepJYe9b64tVinktN8FgO3znqeklSSHGggnGwa12Wt" + "YtPOhbQxxivzTp0q3v+4vSLKxLa9/8KMqfh5J2UUgbVpHW2HFOpyMJWwz7QxoY0JbUxUNy" + "ZmxHz0oV/IZy0arGBCsLYGnTfeCa8VOkIqfvMpIa3Ny7T5CKKqZFdGTMlVtJG8/Zbkh+0s" + "OmjL6WGiZuzamZZ5aV1tdsfVZsWQjNjPDHedNcc024Ee1jYMK+JZ9FWM9LSUIpvMNkz1Fm" + "mag+Wc9aASYV2ZrdZ+1FZrmHE03KrhdXH7PY+u05yv5nx3v5FIzhupRFQuRNQMe6i/4Crb" + "3TznpTKVlhTTxdtSiI5sOOZIyXTH4qmZllLkk298cq5FndfAmO/aZ1s/Y95SD5liKGoHmX" + "aQKQEunrqO54fhvZXZcqmwksA2EiYaT7iqsGbllES0ES/EetNUz8+CIkvxblMV0Jygkpg2" + "MkN1wEFncyupJTRwD3nYlJaoiu50y0hfuGjTGqZ3j6rgbliuoZjXfUEerVgEMSGiKF3UxF" + "LIP40qWk/YXE0Aj1ban49K9ucjyf7sEB/JuPSSIogLEV0EMevwbUURxL/+HwqiBhM=" +) 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/raggr-frontend/src/hooks/useChat.ts b/raggr-frontend/src/hooks/useChat.ts index 21074c7..1d61a3d 100644 --- a/raggr-frontend/src/hooks/useChat.ts +++ b/raggr-frontend/src/hooks/useChat.ts @@ -14,9 +14,12 @@ const TOOL_MESSAGES: Record = { web_search: "Searching the web...", get_current_date: "Checking today's date...", ynab_budget_summary: "Checking budget summary...", + ynab_account_balances: "Checking account balances...", ynab_search_transactions: "Looking up transactions...", - ynab_category_spending: "Analyzing category spending...", - ynab_insights: "Generating budget insights...", + ynab_spending_report: "Adding up spending...", + ynab_category_budget: "Checking category budgets...", + ynab_recent_changes: "Checking recent budget changes...", + ynab_sql: "Querying budget data...", obsidian_search_notes: "Searching notes...", obsidian_read_note: "Reading note...", obsidian_create_note: "Saving note...", diff --git a/tests/unit/test_ynab_mirror.py b/tests/unit/test_ynab_mirror.py new file mode 100644 index 0000000..f988fc5 --- /dev/null +++ b/tests/unit/test_ynab_mirror.py @@ -0,0 +1,434 @@ +"""Tests for the YNAB Postgres mirror: delta application, change log, SQL guard.""" + +import asyncio +from decimal import Decimal + +import pytest +import sqlglot +from tortoise import Tortoise +from tortoise.transactions import in_transaction + +from blueprints.ynab.models import YnabChange, YnabMonthCategory, YnabTransaction +from blueprints.ynab.sql_tool import _check, 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, full=False): + async with in_transaction() as conn: + return await apply_budget( + budget, knowledge, conn, log_changes=log_changes, full=full + ) + + +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_keeps_duplicate_column_names(): + rows = [("Groceries", "Bills", Decimal("-54.150")), ("Rent", None, Decimal("10"))] + assert format_rows(["name", "name", "total"], rows) == ( + "name\tname\ttotal\nGroceries\tBills\t-54.150\nRent\t\t10" + ) + assert format_rows([], []) == "Query returned no rows." + + +def test_unbalanced_quote_is_reported_not_raised(): + error = validate_sql("SELECT * FROM ynab_payees WHERE name = 'Trader Joe's'") + assert error and error.startswith("Could not parse SQL") + + +def test_executed_sql_drops_comments_and_semicolon(): + error, sql = _check( + "SELECT payee_name, sum(amount) FROM ynab_transactions GROUP BY 1; -- top" + ) + assert error is None + assert "--" not in sql and not sql.endswith(";") + # The LIMIT wrapper must still parse: a leftover line comment would + # swallow its closing parenthesis + wrapped = sqlglot.parse_one( + f"SELECT * FROM ({sql}) AS q LIMIT 100", read="postgres" + ) + assert wrapped.args["limit"] is not None + + +async def test_full_sync_marks_missing_rows_deleted(db): + await _apply(_budget([_txn(), _txn(id="txn-2")]), 1, log_changes=False, full=True) + # A full download omits deleted entities entirely + counts = await _apply(_budget([_txn(id="txn-2")]), 2, full=True) + + assert counts["transactions"] == 1 + assert (await YnabTransaction.get(id="txn-1")).deleted is True + assert (await YnabTransaction.get(id="txn-2")).deleted is False + change = await YnabChange.get(entity_id="txn-1") + assert change.action == "deleted" and "Costco" in change.label + + +async def test_delta_sync_does_not_treat_missing_rows_as_deleted(db): + await _apply(_budget([_txn(), _txn(id="txn-2")]), 1, log_changes=False) + await _apply(_budget([_txn(id="txn-2")]), 2) + assert (await YnabTransaction.get(id="txn-1")).deleted is False + + +async def test_full_sync_skips_entity_types_absent_from_payload(db): + await _apply(_budget([_txn()]), 1, log_changes=False, full=True) + budget = _budget([]) + del budget["transactions"] + await _apply(budget, 2, full=True) + assert (await YnabTransaction.get(id="txn-1")).deleted is False + + +async def test_new_month_with_nothing_assigned_is_not_logged(db): + await _apply(_budget([]), 1, log_changes=False) + next_month = {**_month(budgeted=0, activity=0), "month": "2026-11-01"} + await _apply(_budget([], next_month), 2) + assert await YnabChange.filter(entity_type="month_category").count() == 0 + + +async def test_recategorization_logs_names_not_ids(db): + household = {**CATEGORY, "id": "cat-2", "name": "Household"} + budget = _budget([_txn()]) + budget["categories"].append(household) + await _apply(budget, 1, log_changes=False) + + budget = _budget([_txn(category_id="cat-2")]) + budget["categories"].append(household) + await _apply(budget, 2) + + change = await YnabChange.get(entity_type="transaction") + assert change.changes == {"category": ["Groceries", "Household"]} + assert (await YnabTransaction.get(id="txn-1")).category_name == "Household" + + +async def test_tool_errors_are_returned_as_text(monkeypatch): + from blueprints.ynab import tools + + async def fresh(): + return object() + + async def broken_query(*args): + raise ConnectionError("db went away") + + monkeypatch.setattr(tools, "ensure_fresh", fresh) + monkeypatch.setattr(tools, "_query", broken_query) + result = await tools.ynab_account_balances.ainvoke({}) + assert "db went away" in result and result.startswith("YNAB error") + + +async def test_recent_changes_accepts_display_names(monkeypatch): + from blueprints.ynab import tools + + seen = [] + + async def fresh(): + return object() + + async def fake_query(sql, *params): + seen.append(params) + return [] + + monkeypatch.setattr(tools, "ensure_fresh", fresh) + monkeypatch.setattr(tools, "_query", fake_query) + for given, stored in [ + ("Assignments", "month_category"), + ("categories", "category"), + ("category groups", "category_group"), + ("month", "month"), + ]: + await tools.ynab_recent_changes.ainvoke({"entity_type": given}) + assert seen[-1][1] == stored + result = await tools.ynab_recent_changes.ainvoke({"entity_type": "bogus"}) + assert result.startswith("Unknown entity_type") + + +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." + ) + + +async def test_subtransaction_parent_id_is_not_renamed(db): + split = { + "id": "sub-1", + "transaction_id": "txn-1", + "amount": -10_000, + "memo": None, + "payee_id": None, + "category_id": "cat-1", + "transfer_account_id": None, + "deleted": False, + } + budget = _budget([_txn()]) + budget["subtransactions"] = [split] + await _apply(budget, 1, log_changes=False) + budget["subtransactions"] = [{**split, "amount": -12_000}] + await _apply(budget, 2) + change = await YnabChange.get(entity_type="subtransaction") + assert change.changes == {"amount": [-10.0, -12.0]} + + +class TestBudgetPinning: + @pytest.fixture(autouse=True) + async def _db(self, db): + yield + + async def _state(self, budget_id): + from blueprints.ynab.models import YnabSyncState + + await YnabSyncState.create(budget_id=budget_id, server_knowledge=1) + + @pytest.mark.parametrize("configured", ["", "last-used", "default"]) + async def test_alias_or_unset_uses_stored_budget(self, monkeypatch, configured): + from blueprints.ynab.sync import _current_state + + monkeypatch.setenv("YNAB_BUDGET_ID", configured) + await self._state("uuid-a") + assert (await _current_state()).budget_id == "uuid-a" + + async def test_different_configured_budget_is_an_error(self, monkeypatch): + from blueprints.ynab.sync import _current_state + + monkeypatch.setenv("YNAB_BUDGET_ID", "uuid-b") + await self._state("uuid-a") + with pytest.raises(ValueError, match="clear the ynab_"): + await _current_state() + + async def test_multiple_stored_budgets_is_an_error(self, monkeypatch): + from blueprints.ynab.sync import _current_state + + monkeypatch.delenv("YNAB_BUDGET_ID", raising=False) + await self._state("uuid-a") + await self._state("uuid-b") + with pytest.raises(ValueError): + await _current_state() + + +async def test_parallel_syncs_share_one_failure(db, monkeypatch): + from blueprints.ynab import sync + + calls = [] + + class FailingYNAB: + async def fetch_budget(self, budget_id, since): + calls.append(budget_id) + raise TimeoutError("ynab down") + + monkeypatch.setattr(sync, "YNABService", FailingYNAB) + monkeypatch.setattr(sync, "_last_failure", float("-inf")) + await asyncio.gather( + *(sync.ensure_fresh() for _ in range(3)), return_exceptions=True + ) + assert len(calls) == 1 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 index afb4245..0ec4cfd 100644 --- a/utils/ynab_service.py +++ b/utils/ynab_service.py @@ -1,405 +1,37 @@ -"""YNAB API service for querying budget data.""" +"""YNAB API client used by the budget mirror (``blueprints.ynab.sync``).""" import os -from datetime import datetime, timedelta -from typing import Any, Optional -from dotenv import load_dotenv -import ynab +import httpx -# Load environment variables -load_dotenv() +YNAB_API_URL = "https://api.ynab.com/v1" class YNABService: - """Service for interacting with YNAB API.""" + """Thin async client for the YNAB endpoints the mirror needs.""" 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) + async def fetch_budget(self, budget_id: str, last_knowledge: int | None) -> dict: + """GET a whole budget, or only what changed since ``last_knowledge``. - # 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. + Returns ``{"budget": {...}, "server_knowledge": int}``. Full responses + omit deleted entities; delta responses include them with + ``deleted: true``. """ - 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, - } + 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 {self.access_token}"}, ) - 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, - } + response.raise_for_status() + return response.json()["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"