rohitdash08 · pabloDarkmoon24 · Mar 23, 2026
@@ -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