spend_analyzer.py

"""
Spend Analyzer - Financial Intelligence and Analysis Module
"""
import pandas as pd
import numpy as np
from typing import Dict, List, Optional, Tuple
from datetime import datetime, timedelta
from dataclasses import dataclass, asdict
import json
from collections import defaultdict
import logging

@dataclass
class SpendingInsight:
    category: str
    total_amount: float
    transaction_count: int
    average_transaction: float
    percentage_of_total: float
    trend: str  # 'increasing', 'decreasing', 'stable'
    top_merchants: List[str]

@dataclass
class BudgetAlert:
    category: str
    budget_limit: float
    current_spending: float
    percentage_used: float
    alert_level: str  # 'warning', 'critical', 'info'
    days_remaining: int

@dataclass
class FinancialSummary:
    total_income: float
    total_expenses: float
    net_cash_flow: float
    largest_expense: Dict
    most_frequent_category: str
    unusual_transactions: List[Dict]
    monthly_trends: Dict[str, float]

class SpendAnalyzer:
    def __init__(self):
        self.logger = logging.getLogger(__name__)
        self.transactions_df = pd.DataFrame()
        self.budgets = {}
        
    def load_transactions(self, transactions: List) -> None:
        """Load transactions into pandas DataFrame for analysis"""
        try:
            # Convert transactions to DataFrame
            data = []
            for trans in transactions:
                if hasattr(trans, '__dict__'):
                    data.append(asdict(trans))
                else:
                    data.append(trans)
            
            self.transactions_df = pd.DataFrame(data)
            
            if not self.transactions_df.empty:
                # Ensure date column is datetime
                self.transactions_df['date'] = pd.to_datetime(self.transactions_df['date'])
                
                # Sort by date
                self.transactions_df = self.transactions_df.sort_values('date')
                
                # Add derived columns
                self.transactions_df['month'] = self.transactions_df['date'].dt.to_period('M')
                self.transactions_df['week'] = self.transactions_df['date'].dt.to_period('W')
                self.transactions_df['day_of_week'] = self.transactions_df['date'].dt.day_name()
                
                self.logger.info(f"Loaded {len(self.transactions_df)} transactions")
                
        except Exception as e:
            self.logger.error(f"Error loading transactions: {e}")
            raise
    
    def set_budgets(self, budgets: Dict[str, float]) -> None:
        """Set budget limits for categories"""
        self.budgets = budgets
    
    def analyze_spending_by_category(self, months_back: int = 6) -> List[SpendingInsight]:
        """Analyze spending patterns by category"""
        if self.transactions_df.empty:
            return []
        
        # Filter to recent months
        cutoff_date = datetime.now() - timedelta(days=months_back * 30)
        recent_df = self.transactions_df[self.transactions_df['date'] >= cutoff_date]
        
        # Filter only expenses (negative amounts)
        expenses_df = recent_df[recent_df['amount'] < 0].copy()
        expenses_df['amount'] = expenses_df['amount'].abs()  # Make positive for analysis
        
        insights = []
        total_spending = expenses_df['amount'].sum()
        
        if total_spending == 0:
            self.logger.warning("Total spending is zero; no insights can be generated.")
            return insights
        
        # Group by category
        category_stats = expenses_df.groupby('category').agg({
            'amount': ['sum', 'count', 'mean'],
            'description': lambda x: list(x.value_counts().head(3).index)
        }).round(2)
        
        category_stats.columns = ['total', 'count', 'average', 'top_merchants']
        
        for category, stats in category_stats.iterrows():
            # Calculate trend
            trend = self._calculate_trend(expenses_df, category)
            
            insight = SpendingInsight(
                category=category,
                total_amount=stats['total'],
                transaction_count=stats['count'],
                average_transaction=stats['average'],
                percentage_of_total=(stats['total'] / total_spending) * 100,
                trend=trend,
                top_merchants=stats['top_merchants'][:3]
            )
            insights.append(insight)
        
        # Sort by total amount descending
        insights.sort(key=lambda x: x.total_amount, reverse=True)
        return insights
    
    def _calculate_trend(self, df: pd.DataFrame, category: str) -> str:
        """Calculate spending trend for a category"""
        try:
            category_df = df[df['category'] == category]
            monthly_spending = category_df.groupby('month')['amount'].sum()
            
            if len(monthly_spending) < 2:
                return 'stable'
            
            # Calculate trend using linear regression slope
            x = np.arange(len(monthly_spending))
            y = monthly_spending.values
            slope = np.polyfit(x, y, 1)[0]
            
            if slope > 0.1:
                return 'increasing'
            elif slope < -0.1:
                return 'decreasing'
            else:
                return 'stable'
                
        except Exception:
            return 'stable'
    
    def check_budget_alerts(self) -> List[BudgetAlert]:
        """Check for budget alerts and overspending"""
        if self.transactions_df.empty or not self.budgets:
            return []
        
        alerts = []
        current_month = datetime.now().replace(day=1)
        month_df = self.transactions_df[
            (self.transactions_df['date'] >= current_month) &
            (self.transactions_df['amount'] < 0)  # Only expenses
        ].copy()
        
        month_df['amount'] = month_df['amount'].abs()
        
        # Days remaining in month
        import calendar
        days_in_month = calendar.monthrange(current_month.year, current_month.month)[1]
        days_remaining = days_in_month - datetime.now().day
        
        # Check each budget category
        for category, budget_limit in self.budgets.items():
            current_spending = month_df[month_df['category'] == category]['amount'].sum()
            percentage_used = (current_spending / budget_limit) * 100
            
            # Determine alert level
            if percentage_used >= 100:
                alert_level = 'critical'
            elif percentage_used >= 80:
                alert_level = 'warning'
            else:
                alert_level = 'info'
            
            alert = BudgetAlert(
                category=category,
                budget_limit=budget_limit,
                current_spending=current_spending,
                percentage_used=percentage_used,
                alert_level=alert_level,
                days_remaining=days_remaining
            )
            alerts.append(alert)
        
        return alerts
    
    def generate_financial_summary(self) -> FinancialSummary:
        """Generate comprehensive financial summary"""
        if self.transactions_df.empty:
            return FinancialSummary(0, 0, 0, {}, "", [], {})
        
        # Calculate basic metrics
        income_df = self.transactions_df[self.transactions_df['amount'] > 0]
        expense_df = self.transactions_df[self.transactions_df['amount'] < 0]
        
        total_income = income_df['amount'].sum()
        total_expenses = abs(expense_df['amount'].sum())
        net_cash_flow = total_income - total_expenses
        
        # Largest expense
        if not expense_df.empty:
            largest_expense_row = expense_df.loc[expense_df['amount'].idxmin()]
            largest_expense = {
                'amount': abs(largest_expense_row['amount']),
                'description': largest_expense_row['description'],
                'date': largest_expense_row['date'].strftime('%Y-%m-%d'),
                'category': largest_expense_row['category']
            }
        else:
            largest_expense = {}
        
        # Most frequent category
        most_frequent_category = expense_df['category'].mode().iloc[0] if not expense_df.empty else ""
        
        # Unusual transactions (outliers)
        unusual_transactions = self._detect_unusual_transactions()
        
        # Monthly trends
        monthly_trends = self._calculate_monthly_trends()
        
        return FinancialSummary(
            total_income=total_income,
            total_expenses=total_expenses,
            net_cash_flow=net_cash_flow,
            largest_expense=largest_expense,
            most_frequent_category=most_frequent_category,
            unusual_transactions=unusual_transactions,
            monthly_trends=monthly_trends
        )
    
    def _detect_unusual_transactions(self) -> List[Dict]:
        """Detect unusual transactions using statistical methods"""
        if self.transactions_df.empty:
            return []
        
        unusual = []
        
        # Detect amount outliers by category
        for category in self.transactions_df['category'].unique():
            category_df = self.transactions_df[
                (self.transactions_df['category'] == category) & 
                (self.transactions_df['amount'] < 0)
            ].copy()
            
            if len(category_df) < 5:  # Need sufficient data
                continue
            
            amounts = category_df['amount'].abs()
            Q1 = amounts.quantile(0.25)
            Q3 = amounts.quantile(0.75)
            IQR = Q3 - Q1
            
            # Define outliers as values beyond 1.5 * IQR
            lower_bound = Q1 - 1.5 * IQR
            upper_bound = Q3 + 1.5 * IQR
            
            outliers = category_df[(amounts < lower_bound) | (amounts > upper_bound)]
            
            for _, row in outliers.iterrows():
                unusual.append({
                    'date': row['date'].strftime('%Y-%m-%d'),
                    'description': row['description'],
                    'amount': abs(row['amount']),
                    'category': row['category'],
                    'reason': 'Amount significantly higher than usual for this category'
                })
        
        # Detect frequency outliers (multiple transactions same day/merchant)
        daily_merchant = self.transactions_df.groupby([
            self.transactions_df['date'].dt.date, 'description'
        ]).size()
        
        frequent_same_day = daily_merchant[daily_merchant > 3]
        
        for (date, merchant), count in frequent_same_day.items():
            unusual.append({
                'date': str(date),
                'description': merchant,
                'count': count,
                'reason': f'{count} transactions with same merchant on same day'
            })
        
        return unusual[:10]  # Return top 10 unusual transactions
    
    def _calculate_monthly_trends(self) -> Dict[str, float]:
        """Calculate monthly spending trends"""
        if self.transactions_df.empty:
            return {}
        
        # Get last 12 months of expense data
        expense_df = self.transactions_df[self.transactions_df['amount'] < 0].copy()
        expense_df['amount'] = expense_df['amount'].abs()
        
        monthly_spending = expense_df.groupby('month')['amount'].sum()
        
        # Get last 6 months for trend calculation
        recent_months = monthly_spending.tail(6)
        
        trends = {}
        if len(recent_months) >= 2:
            # Overall trend
            x = np.arange(len(recent_months))
            y = recent_months.values
            slope = np.polyfit(x, y, 1)[0]
            trends['overall_trend'] = slope
            
            # Month-over-month change
            if len(recent_months) >= 2:
                current_month = recent_months.iloc[-1]
                previous_month = recent_months.iloc[-2]
                mom_change = ((current_month - previous_month) / previous_month) * 100
                trends['month_over_month_change'] = mom_change
            
            # Average monthly spending
            trends['average_monthly'] = recent_months.mean()
            trends['highest_month'] = recent_months.max()
            trends['lowest_month'] = recent_months.min()
        
        return trends
    
    def predict_future_spending(self, months_ahead: int = 3) -> Dict[str, float]:
        """Predict future spending based on historical trends"""
        if self.transactions_df.empty:
            return {}
        
        # Get historical monthly spending by category
        expense_df = self.transactions_df[self.transactions_df['amount'] < 0].copy()
        expense_df['amount'] = expense_df['amount'].abs()
        
        monthly_category_spending = expense_df.groupby(['month', 'category'])['amount'].sum().unstack(fill_value=0)
        
        predictions = {}
        
        for category in monthly_category_spending.columns:
            category_data = monthly_category_spending[category]
            
            if len(category_data) >= 3:  # Need at least 3 months of data
                # Simple linear trend prediction
                x = np.arange(len(category_data))
                y = category_data.values
                
                # Fit linear model
                coeffs = np.polyfit(x, y, 1)
                slope, intercept = coeffs
                
                # Predict future months
                future_months = []
                for i in range(1, months_ahead + 1):
                    future_x = len(category_data) + i - 1
                    predicted_amount = slope * future_x + intercept
                    future_months.append(max(0, predicted_amount))  # Don't predict negative spending
                
                predictions[category] = {
                    'next_month': future_months[0] if future_months else 0,
                    'total_predicted': sum(future_months),
                    'average_predicted': np.mean(future_months) if future_months else 0
                }
        
        return predictions
    
    def get_spending_recommendations(self) -> List[str]:
        """Generate spending recommendations based on analysis"""
        recommendations = []
        
        if self.transactions_df.empty:
            return ["No transaction data available for analysis"]
        
        # Analyze spending patterns
        insights = self.analyze_spending_by_category()
        budget_alerts = self.check_budget_alerts()
        summary = self.generate_financial_summary()
        
        # Check for overspending categories
        overspending_categories = [alert for alert in budget_alerts if alert.percentage_used > 100]
        if overspending_categories:
            for alert in overspending_categories:
                recommendations.append(
                    f"You've exceeded your {alert.category} budget by "
                    f"${alert.current_spending - alert.budget_limit:.2f} this month. "
                    f"Consider reducing spending in this category."
                )
        
        # Check for high-spending categories
        if insights:
            top_category = insights[0]
            if top_category.percentage_of_total > 40:
                recommendations.append(
                    f"{top_category.category} accounts for {top_category.percentage_of_total:.1f}% "
                    f"of your spending. Consider if this allocation aligns with your priorities."
                )
        
        # Check cash flow
        if summary.net_cash_flow < 0:
            recommendations.append(
                f"Your expenses (${summary.total_expenses:.2f}) exceed your income "
                f"(${summary.total_income:.2f}) by ${abs(summary.net_cash_flow):.2f}. "
                f"Focus on reducing expenses or increasing income."
            )
        
        # Check for increasing trends
        increasing_categories = [i for i in insights if i.trend == 'increasing']
        if increasing_categories:
            top_increasing = increasing_categories[0]
            recommendations.append(
                f"Your {top_increasing.category} spending is trending upward. "
                f"Monitor this category to avoid budget overruns."
            )
        
        # Unusual transaction patterns
        if summary.unusual_transactions:
            recommendations.append(
                f"Found {len(summary.unusual_transactions)} unusual transactions. "
                f"Review these for potential errors or unauthorized charges."
            )
        
        # Positive reinforcement
        decreasing_categories = [i for i in insights if i.trend == 'decreasing']
        if decreasing_categories:
            recommendations.append(
                f"Great job reducing {decreasing_categories[0].category} spending! "
                f"This trend is helping improve your financial health."
            )
        
        if not recommendations:
            recommendations.append("Your spending patterns look healthy. Keep up the good work!")
        
        return recommendations
    
    def export_analysis_data(self) -> Dict:
        """Export all analysis data for Claude API integration"""
        return {
            'spending_insights': [asdict(insight) for insight in self.analyze_spending_by_category()],
            'budget_alerts': [asdict(alert) for alert in self.check_budget_alerts()],
            'financial_summary': asdict(self.generate_financial_summary()),
            'predictions': self.predict_future_spending(),
            'recommendations': self.get_spending_recommendations(),
            'transaction_count': len(self.transactions_df),
            'analysis_date': datetime.now().isoformat()
        }

# Example usage and testing
if __name__ == "__main__":
    # Test the spend analyzer
    analyzer = SpendAnalyzer()
    
    # Sample transaction data for testing
    sample_transactions = [
        {
            'date': datetime.now() - timedelta(days=5),
            'description': 'Amazon Purchase',
            'amount': -45.67,
            'category': 'Shopping'
        },
        {
            'date': datetime.now() - timedelta(days=10),
            'description': 'Grocery Store',
            'amount': -120.50,
            'category': 'Food & Dining'
        },
        {
            'date': datetime.now() - timedelta(days=15),
            'description': 'Salary Deposit',
            'amount': 3000.00,
            'category': 'Income'
        }
    ]
    
    analyzer.load_transactions(sample_transactions)
    analyzer.set_budgets({'Shopping': 100, 'Food & Dining': 200})
    
    insights = analyzer.analyze_spending_by_category()
    print(f"Generated {len(insights)} spending insights")
    
    recommendations = analyzer.get_spending_recommendations()
    print(f"Generated {len(recommendations)} recommendations")