Add AI Financial Insights FastAPI service

.history/app/db_20251124161247.py

@@ -0,0 +1,109 @@
+import logging
+import os
+from datetime import datetime
+from functools import lru_cache
+from typing import Dict, Iterable, List, Optional, Set
+
+from bson import ObjectId
+from dateutil.relativedelta import relativedelta
+from pymongo import MongoClient
+
+from .schemas import Transaction
+
+logger = logging.getLogger(__name__)
+
+
+@lru_cache(maxsize=1)
+def get_mongo_client() -> MongoClient:
+    """Return a cached MongoClient instance."""
+    uri = os.getenv("MONGODB_URI")
+    if not uri:
+        raise RuntimeError("MONGODB_URI environment variable is required for MongoDB access.")
+    logger.info("Connecting to MongoDB host from URI")
+    return MongoClient(uri)
+
+
+def _resolve_category_titles(category_ids: Set[ObjectId], db) -> Dict[ObjectId, str]:
+    if not category_ids:
+        return {}
+    cursor = db["categories"].find({"_id": {"$in": list(category_ids)}}, {"title": 1})
+    return {doc["_id"]: doc.get("title", "Uncategorized") for doc in cursor}
+
+
+def _resolve_currency_codes(currency_ids: Set[ObjectId], db) -> Dict[ObjectId, str]:
+    if not currency_ids:
+        return {}
+    cursor = db["currencies"].find({"_id": {"$in": list(currency_ids)}}, {"code": 1})
+    return {doc["_id"]: doc.get("code", "USD") for doc in cursor}
+
+
+def _safe_object_id(identifier: str) -> ObjectId:
+    try:
+        return ObjectId(identifier)
+    except Exception as exc:  # noqa: BLE001
+        raise ValueError(f"Invalid Mongo ObjectId: {identifier}") from exc
+
+
+def fetch_recent_transactions(user_id: str, months: int = 3) -> List[Transaction]:
+    """
+    Fetch up to `months` of recent expense transactions for a user.
+
+    Returns an empty list if no data exists (new user scenario).
+    """
+    if months <= 0:
+        return []
+
+    client = get_mongo_client()
+    db = client.get_default_database() or client["expense"]
+    collection = db["transactions"]
+
+    user_obj_id = _safe_object_id(user_id)
+    start_date = datetime.utcnow() - relativedelta(months=months)
+
+    query = {
+        "user": user_obj_id,
+        "date": {"$gte": start_date},
+        "type": {"$in": ["EXPENSE", "TRANSFER"]},
+    }
+
+    projection = {
+        "date": 1,
+        "amount": 1,
+        "currency": 1,
+        "category": 1,
+    }
+    docs = list(collection.find(query, projection).sort("date", -1))
+    if not docs:
+        logger.info("No MongoDB transactions found for user_id=%s", user_id)
+        return []
+
+    category_ids = {doc["category"] for doc in docs if doc.get("category")}
+    currency_ids = {doc["currency"] for doc in docs if doc.get("currency")}
+
+    category_map = _resolve_category_titles(category_ids, db)
+    currency_map = _resolve_currency_codes(currency_ids, db)
+
+    transactions: List[Transaction] = []
+    for doc in docs:
+        date_value: Optional[datetime] = doc.get("date")
+        if not date_value:
+            continue
+
+        category_name = category_map.get(doc.get("category"), "Uncategorized")
+        currency_code = currency_map.get(doc.get("currency"), "USD")
+
+        try:
+            transactions.append(
+                Transaction(
+                    timestamp=date_value,
+                    category=category_name,
+                    amount=float(doc.get("amount", 0)),
+                    currency=currency_code,
+                )
+            )
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("Skipping malformed transaction doc=%s error=%s", doc.get("_id"), exc)
+
+    logger.info("Fetched %d transactions from MongoDB for user_id=%s", len(transactions), user_id)
+    return transactions
+

.history/app/db_20251124163521.py

@@ -0,0 +1,109 @@
+import logging
+import os
+from datetime import datetime
+from functools import lru_cache
+from typing import Dict, Iterable, List, Optional, Set
+
+from bson import ObjectId
+from dateutil.relativedelta import relativedelta
+from pymongo import MongoClient
+
+from .schemas import Transaction
+
+logger = logging.getLogger(__name__)
+
+
+@lru_cache(maxsize=1)
+def get_mongo_client() -> MongoClient:
+    """Return a cached MongoClient instance."""
+    uri = os.getenv("MONGODB_URI")
+    if not uri:
+        raise RuntimeError("MONGODB_URI environment variable is required for MongoDB access.")
+    logger.info("Connecting to MongoDB host from URI")
+    return MongoClient(uri)
+
+
+def _resolve_category_titles(category_ids: Set[ObjectId], db) -> Dict[ObjectId, str]:
+    if not category_ids:
+        return {}
+    cursor = db["categories"].find({"_id": {"$in": list(category_ids)}}, {"title": 1})
+    return {doc["_id"]: doc.get("title", "Uncategorized") for doc in cursor}
+
+
+def _resolve_currency_codes(currency_ids: Set[ObjectId], db) -> Dict[ObjectId, str]:
+    if not currency_ids:
+        return {}
+    cursor = db["currencies"].find({"_id": {"$in": list(currency_ids)}}, {"code": 1})
+    return {doc["_id"]: doc.get("code", "USD") for doc in cursor}
+
+
+def _safe_object_id(identifier: str) -> ObjectId:
+    try:
+        return ObjectId(identifier)
+    except Exception as exc:  # noqa: BLE001
+        raise ValueError(f"Invalid Mongo ObjectId: {identifier}") from exc
+
+
+def fetch_recent_transactions(user_id: str, months: int = 3) -> List[Transaction]:
+    """
+    Fetch up to `months` of recent expense transactions for a user.
+
+    Returns an empty list if no data exists (new user scenario).
+    """
+    if months <= 0:
+        return []
+
+    client = get_mongo_client()
+    db = client.get_default_database() or client["expense"]
+    collection = db["transactions"]
+
+    user_obj_id = _safe_object_id(user_id)
+    start_date = datetime.utcnow() - relativedelta(months=months)
+
+    query = {
+        "user": user_obj_id,
+        "date": {"$gte": start_date},
+        "type": {"$in": ["EXPENSE", "TRANSFER"]},
+    }
+
+    projection = {
+        "date": 1,
+        "amount": 1,
+        "currency": 1,
+        "category": 1,
+    }
+    docs = list(collection.find(query, projection).sort("date", -1))
+    if not docs:
+        logger.info("No MongoDB transactions found for user_id=%s", user_id)
+        return []
+
+    category_ids = {doc["category"] for doc in docs if doc.get("category")}
+    currency_ids = {doc["currency"] for doc in docs if doc.get("currency")}
+
+    category_map = _resolve_category_titles(category_ids, db)
+    currency_map = _resolve_currency_codes(currency_ids, db)
+
+    transactions: List[Transaction] = []
+    for doc in docs:
+        date_value: Optional[datetime] = doc.get("date")
+        if not date_value:
+            continue
+
+        category_name = category_map.get(doc.get("category"), "Uncategorized")
+        currency_code = currency_map.get(doc.get("currency"), "USD")
+
+        try:
+            transactions.append(
+                Transaction(
+                    timestamp=date_value,
+                    category=category_name,
+                    amount=float(doc.get("amount", 0)),
+                    currency=currency_code,
+                )
+            )
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("Skipping malformed transaction doc=%s error=%s", doc.get("_id"), exc)
+
+    logger.info("Fetched %d transactions from MongoDB for user_id=%s", len(transactions), user_id)
+    return transactions
+

.history/app/db_20251124164204.py

@@ -0,0 +1,109 @@
+import logging
+import os
+from datetime import datetime
+from functools import lru_cache
+from typing import Dict, Iterable, List, Optional, Set
+
+from bson import ObjectId
+from dateutil.relativedelta import relativedelta
+from pymongo import MongoClient
+
+from .schemas import Transaction
+
+logger = logging.getLogger(__name__)
+
+
+@lru_cache(maxsize=1)
+def get_mongo_client() -> MongoClient:
+    """Return a cached MongoClient instance."""
+    uri = os.getenv("MONGODB_URI")
+    if not uri:
+        raise RuntimeError("MONGODB_URI environment variable is required for MongoDB access.")
+    logger.info("Connecting to MongoDB host from URI")
+    return MongoClient(uri)
+
+
+def _resolve_category_titles(category_ids: Set[ObjectId], db) -> Dict[ObjectId, str]:
+    if not category_ids:
+        return {}
+    cursor = db["categories"].find({"_id": {"$in": list(category_ids)}}, {"title": 1})
+    return {doc["_id"]: doc.get("title", "Uncategorized") for doc in cursor}
+
+
+def _resolve_currency_codes(currency_ids: Set[ObjectId], db) -> Dict[ObjectId, str]:
+    if not currency_ids:
+        return {}
+    cursor = db["currencies"].find({"_id": {"$in": list(currency_ids)}}, {"code": 1})
+    return {doc["_id"]: doc.get("code", "USD") for doc in cursor}
+
+
+def _safe_object_id(identifier: str) -> ObjectId:
+    try:
+        return ObjectId(identifier)
+    except Exception as exc:  # noqa: BLE001
+        raise ValueError(f"Invalid Mongo ObjectId: {identifier}") from exc
+
+
+def fetch_recent_transactions(user_id: str, months: int = 3) -> List[Transaction]:
+    """
+    Fetch up to `months` of recent expense transactions for a user.
+
+    Returns an empty list if no data exists (new user scenario).
+    """
+    if months <= 0:
+        return []
+
+    client = get_mongo_client()
+    db = client.get_default_database() or client["expense"]
+    collection = db["transactions"]
+
+    user_obj_id = _safe_object_id(user_id)
+    start_date = datetime.utcnow() - relativedelta(months=months)
+
+    query = {
+        "user": user_obj_id,
+        "date": {"$gte": start_date},
+        "type": {"$in": ["EXPENSE", "TRANSFER"]},
+    }
+
+    projection = {
+        "date": 1,
+        "amount": 1,
+        "currency": 1,
+        "category": 1,
+    }
+    docs = list(collection.find(query, projection).sort("date", -1))
+    if not docs:
+        logger.info("No MongoDB transactions found for user_id=%s", user_id)
+        return []
+
+    category_ids = {doc["category"] for doc in docs if doc.get("category")}
+    currency_ids = {doc["currency"] for doc in docs if doc.get("currency")}
+
+    category_map = _resolve_category_titles(category_ids, db)
+    currency_map = _resolve_currency_codes(currency_ids, db)
+
+    transactions: List[Transaction] = []
+    for doc in docs:
+        date_value: Optional[datetime] = doc.get("date")
+        if not date_value:
+            continue
+
+        category_name = category_map.get(doc.get("category"), "Uncategorized")
+        currency_code = currency_map.get(doc.get("currency"), "USD")
+
+        try:
+            transactions.append(
+                Transaction(
+                    timestamp=date_value,
+                    category=category_name,
+                    amount=float(doc.get("amount", 0)),
+                    currency=currency_code,
+                )
+            )
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("Skipping malformed transaction doc=%s error=%s", doc.get("_id"), exc)
+
+    logger.info("Fetched %d transactions from MongoDB for user_id=%s", len(transactions), user_id)
+    return transactions
+

Dockerfile

@@ -0,0 +1,21 @@
+FROM python:3.11-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PATH="/home/user/.local/bin:$PATH"
+
+RUN useradd -m -u 1000 user
+USER user
+
+WORKDIR /app
+
+COPY --chown=user:user requirements.txt .
+RUN pip install --no-cache-dir --upgrade pip && \
+    pip install --no-cache-dir -r requirements.txt
+
+COPY --chown=user:user . .
+
+EXPOSE 7860
+
+CMD ["python3", "-m", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "7860"]
+

README.md

@@ -0,0 +1,74 @@
+## AI Financial Insights
+
+AI Financial Insights provides personalized spending summaries that mirror the AI-driven budgeting direction outlined for WalletSync, helping users spot changing expense patterns before overspending occurs. The service analyzes up to the three most recent months of transaction history (or gracefully handles brand-new users) and surfaces conversational insights such as “Your grocery expenses increased by 22% this month compared to last month.”
+
+### Features
+- FastAPI backend with `/insights` endpoint that accepts a list of transactions and returns contextual insights per category plus overall month-over-month highlights.
+- Logic considers up to the latest three months of data; if less history is available, it adapts without errors.
+- Dockerfile tailored for Hugging Face Spaces (listening on port `7860`).
+
+### Project structure
+```
+.
+├── app
+│   ├── __init__.py
+│   ├── main.py          # FastAPI entrypoint
+│   ├── schemas.py       # Pydantic models
+│   └── services.py      # Insight generation logic
+├── requirements.txt
+└── Dockerfile
+```
+
+### Local development
+1. Create a virtual environment and install dependencies:
+   ```
+   python -m venv .venv
+   .\.venv\Scripts\activate
+   python -m pip install --upgrade pip
+   python -m pip install -r requirements.txt
+   ```
+2. Export your Mongo connection string (provided by WalletSync) before running the API:
+   ```
+   setx MONGODB_URI "mongodb://expenseuser:Kem_6o%3F%3F@165.227.69.221:27017/expense?authSource=admin"
+   ```
+3. Run the FastAPI server (per user rule, use `python3` to launch uvicorn):
+   ```
+   python3 -m uvicorn app.main:app --host 0.0.0.0 --port 8000
+   ```
+4. Open the docs at `http://localhost:8000/docs`.
+
+### Request example
+```
+POST /insights
+{
+  "user_id": "demo-user",
+  "transactions": [
+    {"timestamp": "2025-09-05T10:00:00Z", "category": "Groceries", "amount": 120.5, "currency": "INR"},
+    {"timestamp": "2025-10-04T11:00:00Z", "category": "Groceries", "amount": 150.0, "currency": "INR"},
+    {"timestamp": "2025-10-15T09:30:00Z", "category": "Travel",    "amount": 2000.0, "currency": "INR"}
+  ]
+}
+```
+
+### Docker (Hugging Face Space)
+Build and test locally:
+```
+docker build -t ai-financial-insights .
+docker run -it -p 7860:7860 ai-financial-insights
+```
+When deploying on Hugging Face, add `MONGODB_URI` as a Space secret so the container can reach your database.
+
+### Deploying to Hugging Face
+1. Log in and clone the Space:
+   ```
+   git clone https://huggingface.co/spaces/LogicGoInfotechSpaces/AI_FINANCIAL_INSIGHTS
+   cd AI_FINANCIAL_INSIGHTS
+   ```
+2. Copy the project files into the cloned repo and commit:
+   ```
+   git add .
+   git commit -m "Add AI Financial Insights FastAPI app"
+   git push
+   ```
+3. The Docker Space automatically builds and serves the FastAPI app on port `7860`. Monitor the Space page for build logs and status.
+

app/__init__.py

@@ -0,0 +1,4 @@
+"""
+AI Financial Insights FastAPI application package.
+"""
+

app/db.py

@@ -0,0 +1,109 @@
+import logging
+import os
+from datetime import datetime
+from functools import lru_cache
+from typing import Dict, Iterable, List, Optional, Set
+
+from bson import ObjectId
+from dateutil.relativedelta import relativedelta
+from pymongo import MongoClient
+
+from .schemas import Transaction
+
+logger = logging.getLogger(__name__)
+
+
+@lru_cache(maxsize=1)
+def get_mongo_client() -> MongoClient:
+    """Return a cached MongoClient instance."""
+    uri = os.getenv("MONGODB_URI")
+    if not uri:
+        raise RuntimeError("MONGODB_URI environment variable is required for MongoDB access.")
+    logger.info("Connecting to MongoDB host from URI")
+    return MongoClient(uri)
+
+
+def _resolve_category_titles(category_ids: Set[ObjectId], db) -> Dict[ObjectId, str]:
+    if not category_ids:
+        return {}
+    cursor = db["categories"].find({"_id": {"$in": list(category_ids)}}, {"title": 1})
+    return {doc["_id"]: doc.get("title", "Uncategorized") for doc in cursor}
+
+
+def _resolve_currency_codes(currency_ids: Set[ObjectId], db) -> Dict[ObjectId, str]:
+    if not currency_ids:
+        return {}
+    cursor = db["currencies"].find({"_id": {"$in": list(currency_ids)}}, {"code": 1})
+    return {doc["_id"]: doc.get("code", "USD") for doc in cursor}
+
+
+def _safe_object_id(identifier: str) -> ObjectId:
+    try:
+        return ObjectId(identifier)
+    except Exception as exc:  # noqa: BLE001
+        raise ValueError(f"Invalid Mongo ObjectId: {identifier}") from exc
+
+
+def fetch_recent_transactions(user_id: str, months: int = 3) -> List[Transaction]:
+    """
+    Fetch up to `months` of recent expense transactions for a user.
+
+    Returns an empty list if no data exists (new user scenario).
+    """
+    if months <= 0:
+        return []
+
+    client = get_mongo_client()
+    db = client.get_default_database() or client["expense"]
+    collection = db["transactions"]
+
+    user_obj_id = _safe_object_id(user_id)
+    start_date = datetime.utcnow() - relativedelta(months=months)
+
+    query = {
+        "user": user_obj_id,
+        "date": {"$gte": start_date},
+        "type": {"$in": ["EXPENSE", "TRANSFER"]},
+    }
+
+    projection = {
+        "date": 1,
+        "amount": 1,
+        "currency": 1,
+        "category": 1,
+    }
+    docs = list(collection.find(query, projection).sort("date", -1))
+    if not docs:
+        logger.info("No MongoDB transactions found for user_id=%s", user_id)
+        return []
+
+    category_ids = {doc["category"] for doc in docs if doc.get("category")}
+    currency_ids = {doc["currency"] for doc in docs if doc.get("currency")}
+
+    category_map = _resolve_category_titles(category_ids, db)
+    currency_map = _resolve_currency_codes(currency_ids, db)
+
+    transactions: List[Transaction] = []
+    for doc in docs:
+        date_value: Optional[datetime] = doc.get("date")
+        if not date_value:
+            continue
+
+        category_name = category_map.get(doc.get("category"), "Uncategorized")
+        currency_code = currency_map.get(doc.get("currency"), "USD")
+
+        try:
+            transactions.append(
+                Transaction(
+                    timestamp=date_value,
+                    category=category_name,
+                    amount=float(doc.get("amount", 0)),
+                    currency=currency_code,
+                )
+            )
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("Skipping malformed transaction doc=%s error=%s", doc.get("_id"), exc)
+
+    logger.info("Fetched %d transactions from MongoDB for user_id=%s", len(transactions), user_id)
+    return transactions
+

app/main.py

@@ -0,0 +1,63 @@
+import logging
+import os
+from typing import Dict, List
+
+from fastapi import FastAPI, HTTPException
+
+from .db import fetch_recent_transactions
+from .schemas import InsightRequest, InsightResponse, Transaction
+from .services import generate_insights
+
+logging.basicConfig(
+    level=os.environ.get("LOG_LEVEL", "INFO"),
+    format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+app = FastAPI(
+    title="AI Financial Insights",
+    description="AI-powered personalized spending insight generator for WalletSync.",
+    version="1.0.0",
+)
+
+
+@app.get("/")
+def root() -> Dict[str, str]:
+    """Simple description endpoint."""
+    return {
+        "message": "AI Financial Insights is ready. POST /insights with transactions to receive insights.",
+        "docs": "/docs",
+    }
+
+
+@app.get("/health")
+def health() -> Dict[str, str]:
+    """Health probe for monitoring."""
+    return {"status": "ok"}
+
+
+def _resolve_transactions(payload: InsightRequest) -> List[Transaction]:
+    if payload.transactions:
+        return payload.transactions
+    if not payload.user_id:
+        raise HTTPException(
+            status_code=400,
+            detail="Provide `transactions` or `user_id` to fetch them from MongoDB.",
+        )
+    try:
+        return fetch_recent_transactions(payload.user_id)
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    except Exception as exc:  # noqa: BLE001
+        logger.exception("Failed to fetch MongoDB data for user_id=%s", payload.user_id)
+        raise HTTPException(status_code=502, detail="Unable to query transaction store.") from exc
+
+
+@app.post("/insights", response_model=InsightResponse)
+def create_insights(payload: InsightRequest) -> InsightResponse:
+    """Generate insights from a transaction payload."""
+    logger.info("Received request for user_id=%s", payload.user_id)
+    transactions = _resolve_transactions(payload)
+    insights, months = generate_insights(transactions)
+    return InsightResponse(user_id=payload.user_id, insights=insights, evaluated_months=months)
+

app/schemas.py

@@ -0,0 +1,50 @@
+from datetime import datetime
+from typing import List, Optional
+
+from pydantic import BaseModel, Field, field_validator
+
+
+class Transaction(BaseModel):
+    """Incoming transaction payload."""
+
+    timestamp: datetime = Field(..., description="Timestamp of the transaction (ISO 8601).")
+    category: str = Field(..., min_length=1, max_length=100, description="Spending category label.")
+    amount: float = Field(..., gt=0, description="Spending amount in the provided currency.")
+    currency: str = Field("USD", min_length=1, max_length=5, description="ISO currency code.")
+
+    @field_validator("category")
+    def normalize_category(cls, value: str) -> str:
+        normalized = value.strip()
+        if not normalized:
+            raise ValueError("category must contain visible characters")
+        return normalized.title()
+
+    @field_validator("currency")
+    def uppercase_currency(cls, value: str) -> str:
+        normalized = value.strip().upper()
+        if not normalized:
+            raise ValueError("currency must contain visible characters")
+        return normalized
+
+
+class InsightRequest(BaseModel):
+    """Request model for insight generation."""
+
+    user_id: Optional[str] = Field(None, description="Optional identifier for the user.")
+    transactions: List[Transaction] = Field(default_factory=list)
+
+
+class Insight(BaseModel):
+    """Single insight message."""
+
+    message: str
+    category: Optional[str] = None
+
+
+class InsightResponse(BaseModel):
+    """Response payload for generated insights."""
+
+    user_id: Optional[str] = None
+    insights: List[Insight] = Field(default_factory=list)
+    evaluated_months: List[str] = Field(default_factory=list, description="ISO year-month strings considered.")
+

app/services.py

@@ -0,0 +1,147 @@
+from __future__ import annotations
+
+import logging
+from collections import defaultdict
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Dict, Iterable, List, Tuple
+
+from dateutil.relativedelta import relativedelta
+
+from .schemas import Insight, Transaction
+
+logger = logging.getLogger(__name__)
+
+CURRENCY_SYMBOLS: Dict[str, str] = {
+    "INR": "₹",
+    "USD": "$",
+    "EUR": "€",
+    "GBP": "£",
+}
+
+
+@dataclass(frozen=True)
+class MonthlySummary:
+    """Aggregate spending for a single month and category."""
+
+    year: int
+    month: int
+    category: str
+    total: float
+
+    @property
+    def iso_month(self) -> str:
+        return f"{self.year:04d}-{self.month:02d}"
+
+
+def _bucket_transactions(transactions: Iterable[Transaction]) -> List[MonthlySummary]:
+    """Aggregate transactions per month and category."""
+    buckets: Dict[Tuple[int, int, str], float] = defaultdict(float)
+    for txn in transactions:
+        key = (txn.timestamp.year, txn.timestamp.month, txn.category)
+        buckets[key] += txn.amount
+
+    summaries = [
+        MonthlySummary(year=year, month=month, category=category, total=round(total, 2))
+        for (year, month, category), total in buckets.items()
+    ]
+    logger.debug("Created %d monthly summaries", len(summaries))
+    return summaries
+
+
+def _format_currency(amount: float, currency: str) -> str:
+    symbol = CURRENCY_SYMBOLS.get(currency.upper(), "")
+    formatted_amount = f"{amount:,.2f}"
+    return f"{symbol}{formatted_amount}" if symbol else f"{formatted_amount} {currency.upper()}"
+
+
+def _month_key(summary: MonthlySummary) -> datetime:
+    return datetime(summary.year, summary.month, 1)
+
+
+def generate_insights(transactions: Iterable[Transaction]) -> Tuple[List[Insight], List[str]]:
+    """Create AI-inspired insights from historical spending."""
+    txns = list(transactions)
+    if not txns:
+        logger.info("No transactions provided, returning onboarding insight")
+        return [Insight(message="Add a few expenses to start seeing personalized insights.")], []
+
+    summaries = _bucket_transactions(txns)
+    if not summaries:
+        return [Insight(message="No spending data yet. Track expenses to unlock insights.")], []
+
+    summaries.sort(key=_month_key, reverse=True)
+    recent_months = summaries[:12]  # guardrail, though we only surface up to 3 months
+
+    grouped: Dict[str, List[MonthlySummary]] = defaultdict(list)
+    for summary in recent_months:
+        grouped[summary.category].append(summary)
+
+    latest_month = max(recent_months, key=_month_key)
+    latest_month_dt = _month_key(latest_month)
+    evaluated_months = []
+
+    insights: List[Insight] = []
+    for offset in range(3):
+        evaluated_months.append((latest_month_dt - relativedelta(months=offset)).strftime("%Y-%m"))
+    evaluated_months = sorted(set(evaluated_months))
+
+    currency = txns[0].currency
+
+    for category, entries in grouped.items():
+        entries.sort(key=_month_key, reverse=True)
+        current = entries[0]
+        history = entries[1:3]
+        if not history:
+            continue
+
+        history_avg = sum(e.total for e in history) / len(history)
+        if history_avg == 0:
+            continue
+
+        change_pct = ((current.total - history_avg) / history_avg) * 100
+        diff_amount = current.total - history_avg
+
+        if abs(change_pct) < 10:
+            continue
+
+        trend = "increased" if change_pct > 0 else "decreased"
+        insights.append(
+            Insight(
+                category=category,
+                message=(
+                    f"Your {category.lower()} spending {trend} by {abs(change_pct):.0f}% this month "
+                    f"versus your prior average, about {_format_currency(abs(diff_amount), currency)} difference."
+                ),
+            )
+        )
+
+    # Additional highlight: biggest month-over-month change regardless of category
+    monthly_totals: Dict[str, float] = defaultdict(float)
+    for summary in summaries:
+        monthly_totals[summary.iso_month] += summary.total
+
+    sorted_months = sorted(monthly_totals.items(), reverse=True)
+    if len(sorted_months) >= 2:
+        latest_label, latest_total = sorted_months[0]
+        prev_label, prev_total = sorted_months[1]
+        delta = latest_total - prev_total
+        if abs(delta) >= 1:
+            descriptor = "more" if delta > 0 else "less"
+            insights.append(
+                Insight(
+                    message=(
+                        f"You spent {_format_currency(abs(delta), currency)} {descriptor} in {latest_label} compared "
+                        f"to {prev_label}. Consider reviewing large outliers."
+                    )
+                )
+            )
+
+    if not insights:
+        insights.append(
+            Insight(message="Spending is stable across tracked months. Keep up the steady habits!")
+        )
+
+    logger.info("Generated %d insights", len(insights))
+    return insights, evaluated_months
+

requirements.txt

@@ -0,0 +1,5 @@
+fastapi==0.115.0
+uvicorn[standard]==0.30.4
+python-dateutil==2.9.0.post0
+pymongo==4.8.0
+

tests/test_main.py

@@ -0,0 +1,49 @@
+from datetime import datetime
+from typing import List
+
+import pytest
+from fastapi import HTTPException
+from fastapi.testclient import TestClient
+
+from app.main import app, _resolve_transactions
+from app.schemas import InsightRequest, Transaction
+
+
+def _txn() -> Transaction:
+    return Transaction(
+        timestamp=datetime.utcnow(),
+        category="Travel",
+        amount=100.0,
+        currency="INR",
+    )
+
+
+def test_resolve_transactions_handles_payload_list(monkeypatch):
+    payload = InsightRequest(user_id="abc", transactions=[_txn()])
+    result = _resolve_transactions(payload)
+    assert len(result) == 1
+
+
+def test_resolve_transactions_requires_user_id_when_no_transactions():
+    payload = InsightRequest(user_id=None, transactions=[])
+    with pytest.raises(HTTPException):
+        _resolve_transactions(payload)
+
+
+def test_resolve_transactions_fetches_from_db(monkeypatch):
+    payload = InsightRequest(user_id="6686280844fc8cbae596aa51", transactions=[])
+
+    def fake_fetch(user_id: str) -> List[Transaction]:
+        assert user_id == payload.user_id
+        return [_txn()]
+
+    monkeypatch.setattr("app.main.fetch_recent_transactions", fake_fetch)
+    result = _resolve_transactions(payload)
+    assert len(result) == 1
+
+
+def test_insights_endpoint_requires_payload_when_empty(monkeypatch):
+    client = TestClient(app)
+    response = client.post("/insights", json={"user_id": None, "transactions": []})
+    assert response.status_code == 400
+

tests/test_services.py

@@ -0,0 +1,32 @@
+from datetime import datetime
+
+from app.schemas import Transaction
+from app.services import generate_insights
+
+
+def _txn(ts: str, category: str, amount: float) -> Transaction:
+    return Transaction(
+        timestamp=datetime.fromisoformat(ts),
+        category=category,
+        amount=amount,
+        currency="INR",
+    )
+
+
+def test_generate_insights_detects_increase():
+    txns = [
+        _txn("2025-09-05T00:00:00", "Groceries", 100),
+        _txn("2025-10-05T00:00:00", "Groceries", 130),
+        _txn("2025-11-05T00:00:00", "Groceries", 200),
+    ]
+    insights, months = generate_insights(txns)
+    assert insights, "Expected at least one insight"
+    assert any("increased" in insight.message for insight in insights)
+    assert months, "Expected evaluated months to be returned"
+
+
+def test_generate_insights_handles_new_user():
+    insights, months = generate_insights([])
+    assert months == []
+    assert insights[0].message.startswith("Add a few expenses")
+