From 5bfedee79f10898e3890444613956a8067b1d11a Mon Sep 17 00:00:00 2001 From: moon <152454724+pabloDarkmoon24@users.noreply.github.com> Date: Mon, 23 Mar 2026 17:04:54 -0500 Subject: [PATCH] fix: solution for issue #73 --- fix_issue_73.py | 3 +++ 1 file changed, 3 insertions(+) create mode 100644 fix_issue_73.py diff --git a/fix_issue_73.py b/fix_issue_73.py new file mode 100644 index 00000000..6b720c0f --- /dev/null +++ b/fix_issue_73.py @@ -0,0 +1,3 @@ +```json +{ + "solution_code": "### FILE: app/services/budget_suggestion.py\n```python\n\"\"\"\nDynamic Budget Suggestion Service\nUses 3-6 months of past spending data to suggest budget limits with confidence scores.\n\"\"\"\n\nfrom __future__ import annotations\n\nimport json\nimport logging\nimport math\nfrom datetime import date, timedelta\nfrom typing import Any\n\nimport numpy as np\nfrom sqlalchemy import text\nfrom sqlalchemy.orm import Session\n\nlogger = logging.getLogger(__name__)\n\n\n# ---------------------------------------------------------------------------\n# Statistical helpers\n# ---------------------------------------------------------------------------\n\ndef _coefficient_of_variation(values: list[float]) -> float:\n \"\"\"CV = std / mean. Returns 0 if mean is 0.\"\"\"\n if not values or len(values) < 2:\n return 0.0\n arr = np.array(values, dtype=float)\n mean = arr.mean()\n if mean == 0:\n return 0.0\n return float(arr.std(ddof=1) / mean)\n\n\ndef _linear_trend_slope(values: list[float]) -> float:\n \"\"\"\n Returns the slope of a simple linear regression over the values.\n Positive slope → spending increasing; negative → decreasing.\n \"\"\"\n n = len(values)\n if n < 2:\n return 0.0\n x = np.arange(n, dtype=float)\n y = np.array(values, dtype=float)\n x_mean, y_mean = x.mean(), y.mean()\n denom = ((x - x_mean) ** 2).sum()\n if denom == 0:\n return 0.0\n return float(((x - x_mean) * (y - y_mean)).sum() / denom)\n\n\ndef _compute_confidence(\n monthly_values: list[float],\n num_months: int,\n) -> float:\n \"\"\"\n Confidence score in [0.0, 1.0] based on:\n - Number of data months (more → higher)\n - Coefficient of variation (lower → higher)\n \"\"\"\n # Month coverage score: 3 months → 0.50, 6 months → 1.0\n month_score = min(num_months / 6.0, 1.0)\n\n # Stability score: CV == 0 → 1.0, CV >= 1.0 → 0.0\n cv = _coefficient_of_variation(monthly_values)\n stability_score = max(0.0, 1.0 - cv)\n\n confidence = 0.5 * month_score + 0.5 * stability_score\n return round(min(max(confidence, 0.0), 1.0), 3)\n\n\n# ---------------------------------------------------------------------------\n# DB query helpers\n# ---------------------------------------------------------------------------\n\ndef _fetch_monthly_category_spending(\n db: Session,\n user_id: int,\n months: int,\n) -> dict[str, list[float]]:\n \"\"\"\n Returns {category_name: [amount_month_oldest, ..., amount_month_latest]}\n for the last `months` calendar months (oldest → newest).\n \"\"\"\n today = date.today()\n # First day of the *current* month\n first_of_current = today.replace(day=1)\n # Start of the window\n # Walk back `months` months\n year = first_of_current.year\n month = first_of_current.month - months\n while month <= 0:\n month += 12\n year -= 1\n window_start = date(year, month, 1)\n\n sql = text(\n \"\"\"\n SELECT\n c.name AS category,\n DATE_TRUNC('month', e.date)::date AS month,\n SUM(e.amount) AS total\n FROM expenses e\n JOIN categories c ON c.id = e.category_id\n WHERE e.user_id = :user_id\n AND e.date >= :window_start\n AND e.date < :window_end\n GROUP BY c.name, DATE_TRUNC('month', e.date)\n ORDER BY c.name, month\n \"\"\"\n )\n rows = db.execute(\n sql,\n {\"user_id\": user_id, \"window_start\": window_start, \"window_end\": first_of_current},\n ).fetchall()\n\n # Build ordered month list (oldest → newest)\n months_list: list[date] = []\n y, m = year, month\n for _ in range(months):\n months_list.append(date(y, m, 1))\n m += 1\n if m > 12:\n m = 1\n y += 1\n\n # category → {month_date: total}\n raw: dict[str, dict[date, float]] = {}\n for row in rows:\n cat = row.category\n raw.setdefault(cat, {})\n raw[cat][row.month] = float(row.total)\n\n # Fill missing months with 0.0\n result: dict[str, list[float]] = {}\n for cat, monthly_map in raw.items():\n result[cat] = [monthly_map.get(m, 0.0) for m in months_list]\n\n return result\n\n\n# ---------------------------------------------------------------------------\n# Core suggestion logic\n# ---------------------------------------------------------------------------\n\ndef generate_budget_suggestions(\n db: Session,\n user_id: int,\n months: int = 6,\n) -> list[dict[str, Any]]:\n \"\"\"\n Generate per-category budget suggestions from the last `months` months.\n\n Parameters\n ----------\n db : SQLAlchemy session\n user_id : authenticated user's ID\n months : look-back window (3-6 months; clamped to that range)\n\n Returns\n -------\n List of dicts, one per category::\n\n [\n {\n \"category\": \"Groceries\",\n \"suggested_limit\": 450.00,\n \"avg_monthly_spend\": 420.00,\n \"trend\": \"increasing\", # increasing / stable / decreasing\n \"confidence\": 0.82,\n \"reasoning\": \"Based on 6 months of data…\"\n },\n ...\n ]\n \"\"\"\n months = max(3, min(months, 6))\n\n monthly_data = _fetch_monthly_category_spending(db, user_id, months)\n\n if not monthly_data:\n return []\n\n suggestions: list[dict[str, Any]] = []\n\n for category, values in monthly_data.items():\n num_nonzero = sum(1 for v in values if v > 0)\n if num_nonzero == 0:\n continue # skip categories with zero spend throughout\n\n arr = np.array(values, dtype=float)\n avg = float(arr.mean())\n p75 = float(np.percentile(arr, 75))\n slope = _linear_trend_slope(values)\n\n # Trend classification\n threshold = avg * 0.05 # 5 % of mean\n if slope > threshold:\n trend = \"increasing\"\n elif slope < -threshold:\n trend = \"decreasing\"\n else:\n trend = \"stable\"\n\n # Suggested limit\n # Base = 75th-percentile spend;\n # add 10 % buffer for increasing trends, subtract 5 % for decreasing.\n if trend == \"increasing\":\n suggested_limit = p75 * 1.10\n elif trend == \"decreasing\":\n suggested_limit = max(avg, p75 * 0.95)\n else:\n suggested_limit = p75\n\n suggested_limit = round(suggested_limit, 2)\n avg_rounded = round(avg, 2)\n\n confidence = _compute_confidence(values, num_months=num_nonzero)\n\n reasoning = (\n f\"Based on {num_nonzero} month(s) of data, your average spend is \"\n f\"${avg_rounded:.2f}/month with a {trend} trend. \"\n f\"The suggested limit of ${suggested_limit:.2f} is set at the \"\n f\"75th-percentile of your monthly spend\"\n )\n if trend == \"increasing\":\n reasoning += \" with a 10 % buffer to account for rising costs.\"\n elif trend == \"decreasing\":\n reasoning += \" adjusted downward to reflect your improving habits.\"\n else:\n reasoning += \" to match your consistent spending pattern.\"\n\n suggestions.append(\n {\n \"category\": category,\n \"suggested_limit\": suggested_limit,\n \"avg_monthly_spend\": avg_rounded,\n \"trend\": trend,\n \"confidence\": confidence,\n \"reasoning\": reasoning,\n }\n )\n\n # Sort by confidence desc, then category name\n suggestions.sort(key=lambda s: (-s[\"confidence\"], s[\"category\"]))\n return suggestions\n```\n\n### FILE: app/routes/insights.py\n```python\n\"\"\"\nInsights routes — monthly summary + budget suggestions.\n\"\"\"\n\nfrom __future__ import annotations\n\nimport json\nimport logging\nfrom functools import wraps\nfrom typing import Any\n\nfrom flask import Blueprint, g, jsonify, request\n\nfrom app.db.session import get_db # adjust import to match project layout\nfrom app.middleware.auth import jwt_required # adjust to match project layout\nfrom app.services.budget_suggestion import generate_budget_suggestions\nfrom app.extensions import redis_client # adjust to match project layout\n\nlogger = logging.getLogger(__name__)\nbp = Blueprint(\"insights\", __name__, url_prefix=\"/insights\")\n\nBUDGET_SUGGESTION_CACHE_TTL = 60 * 60 * 24 # 24 h\n\n\n# ---------------------------------------------------------------------------\n# Helper\n# ---------------------------------------------------------------------------\n\ndef _cache_key(user_id: int, months: int) -> str:\n return f\"insights:{user_id}:budget_suggestion:{months}\"\n\n\n# ---------------------------------------------------------------------------\n# Endpoint\n# ---------------------------------------------------------------------------\n\n@bp.route(\"/budget-suggestion\", methods=[\"GET\"])\n@jwt_required\ndef budget_suggestion() -> Any:\n \"\"\"\n GET /insights/budget-suggestion?months=6\n\n Query params\n ------------\n months : int, optional (default=6, range 3-6)\n Look-back window in months.\n\n Returns\n -------\n 200 OK\n {\n \"user_id\": 42,\n \"months_analyzed\": 6,\n \"suggestions\": [\n {\n \"category\": \"Groceries\",\n \"suggested_limit\": 450.00,\n \"avg_monthly_spend\": 420.00,\n \"trend\": \"increasing\",\n \"confidence \ No newline at end of file