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, category, and currency.""" year: int month: int category: str currency: 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, 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, txn.currency) buckets[key] += txn.amount summaries = [ 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 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 # 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, summary.currency)].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)) for (category, currency), 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 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_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( Insight(message="Spending is stable across tracked months. Keep up the steady habits!") ) 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