Update app/services.py

app/services.py

@@ -22,11 +22,12 @@ CURRENCY_SYMBOLS: Dict[str, str] = {
 
 @dataclass(frozen=True)
 class MonthlySummary:
-    """Aggregate spending for a single month and category."""
+    """Aggregate spending for a single month, category, and currency."""
 
     year: int
     month: int
     category: str
+    currency: str
     total: float
 
     @property
@@ -35,15 +36,15 @@ class MonthlySummary:
 
 
 def _bucket_transactions(transactions: Iterable[Transaction]) -> List[MonthlySummary]:
-    """Aggregate transactions per month and category."""
-    buckets: Dict[Tuple[int, int, str], float] = defaultdict(float)
+    """Aggregate transactions per month, category, and currency."""
+    buckets: Dict[Tuple[int, int, str, str], float] = defaultdict(float)
     for txn in transactions:
-        key = (txn.timestamp.year, txn.timestamp.month, txn.category)
+        key = (txn.timestamp.year, txn.timestamp.month, txn.category, txn.currency)
         buckets[key] += txn.amount
 
     summaries = [
-        MonthlySummary(year=year, month=month, category=category, total=round(total, 2))
-        for (year, month, category), total in buckets.items()
+        MonthlySummary(year=year, month=month, category=category, currency=currency, total=round(total, 2))
+        for (year, month, category, currency), total in buckets.items()
     ]
     logger.debug("Created %d monthly summaries", len(summaries))
     return summaries
@@ -73,9 +74,10 @@ def generate_insights(transactions: Iterable[Transaction]) -> Tuple[List[Insight
     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)
+    # Group by category and currency to handle multi-currency scenarios
+    grouped: Dict[Tuple[str, str], List[MonthlySummary]] = defaultdict(list)
     for summary in recent_months:
-        grouped[summary.category].append(summary)
+        grouped[(summary.category, summary.currency)].append(summary)
 
     latest_month = max(recent_months, key=_month_key)
     latest_month_dt = _month_key(latest_month)
@@ -86,9 +88,7 @@ def generate_insights(transactions: Iterable[Transaction]) -> Tuple[List[Insight
         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():
+    for (category, currency), entries in grouped.items():
         entries.sort(key=_month_key, reverse=True)
         current = entries[0]
         history = entries[1:3]
@@ -116,26 +116,30 @@ def generate_insights(transactions: Iterable[Transaction]) -> Tuple[List[Insight
             )
         )
 
-    # Additional highlight: biggest month-over-month change regardless of category
-    monthly_totals: Dict[str, float] = defaultdict(float)
+    # Additional highlight: biggest month-over-month change per currency
+    # Group monthly totals by currency to avoid mixing currencies
+    monthly_totals_by_currency: Dict[str, Dict[str, float]] = defaultdict(lambda: 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."
+        monthly_totals_by_currency[summary.currency][summary.iso_month] += summary.total
+
+    # Generate insights per currency
+    for currency, monthly_totals in monthly_totals_by_currency.items():
+        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"
+                currency_note = f" (in {currency})" if len(monthly_totals_by_currency) > 1 else ""
+                insights.append(
+                    Insight(
+                        message=(
+                            f"You spent {_format_currency(abs(delta), currency)} {descriptor} in {latest_label} compared "
+                            f"to {prev_label}{currency_note}. Consider reviewing large outliers."
+                        )
                     )
                 )
-            )
 
     if not insights:
         insights.append(
@@ -145,3 +149,155 @@ def generate_insights(transactions: Iterable[Transaction]) -> Tuple[List[Insight
     logger.info("Generated %d insights", len(insights))
     return insights, evaluated_months
 
+
+
+
+
+
+# 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
+