from datetime import datetime, timedelta from collections import defaultdict import numpy as np from sqlalchemy.orm import Session from app.database.models import Account, Transaction def analyze_spending_behavior(db: Session, user_id: str, days: int = 90): """ Analyzes historical transactions to detect behavioral patterns (late-night, impulsive, dopamine, stress). """ accounts = db.query(Account).filter(Account.user_id == user_id).all() account_ids = [acc.id for acc in accounts] if not account_ids: return {"insights": [], "metrics": {}} cutoff = datetime.utcnow() - timedelta(days=days) txns = db.query(Transaction).filter( Transaction.account_id.in_(account_ids), Transaction.timestamp >= cutoff, Transaction.type == "debit" ).all() if not txns: return { "insights": ["No recent debit transactions to analyze. Complete a few purchases to start behavioral profiling."], "metrics": { "late_night_count": 0, "late_night_total": 0.0, "weekend_pct": 0.0, "impulsive_count": 0, "impulsive_total": 0.0, "dopamine_count": 0, "stress_count": 0 } } # Analyze variables late_night_txns = [] weekend_txns = [] impulsive_txns = [] dopamine_txns = [] stress_txns = [] amounts = [t.amount for t in txns] avg_txn = np.mean(amounts) std_txn = np.std(amounts) if len(amounts) > 1 else 0.0 category_totals = defaultdict(float) hourly_counts = defaultdict(int) for t in txns: # Categorize category_totals[t.category or "Other"] += t.amount # Timing hour = t.timestamp.hour hourly_counts[hour] += 1 # Late-night spending (11PM to 4AM) if hour >= 23 or hour < 4: late_night_txns.append(t) # Weekend spending (Friday, Saturday, Sunday) # weekday() is 0=Monday, 4=Friday, 5=Saturday, 6=Sunday day = t.timestamp.weekday() if day in [4, 5, 6]: weekend_txns.append(t) # Impulsive spending (More than average + 1.5 * standard dev, or marked as 'regret') if t.amount > (avg_txn + 1.5 * std_txn) and (t.category in ["Shopping", "Entertainment", "Food"]): impulsive_txns.append(t) # Emotion tags emotion = (t.spending_emotion_label or "").lower() if emotion == "regret": stress_txns.append(t) elif emotion in ["happy", "dopamine"] or (t.category == "Shopping" and t.amount > avg_txn): dopamine_txns.append(t) # Insights construction insights = [] # 1. Late night alert late_night_pct = (len(late_night_txns) / len(txns) * 100) if txns else 0 if late_night_pct > 15: total_late = sum(t.amount for t in late_night_txns) insights.append( f"🌙 High late-night spending: {late_night_pct:.1f}% of transactions occur after 11PM (Total: ${total_late:,.2f}). " "Consider setting a bedtime blocker on your bank card." ) # 2. Weekend overspending weekend_pct = (len(weekend_txns) / len(txns) * 100) if txns else 0 if weekend_pct > 45: weekend_avg = np.mean([t.amount for t in weekend_txns]) if weekend_txns else 0 weekday_txns = [t for t in txns if t not in weekend_txns] weekday_avg = np.mean([t.amount for t in weekday_txns]) if weekday_txns else 0 if weekend_avg > weekday_avg * 1.2: pct_diff = ((weekend_avg - weekday_avg) / weekday_avg) * 100 insights.append( f"🎉 Weekend Spikes: You spend {pct_diff:.1f}% more on weekends than weekdays. " "Mainly driven by dining out and recreational purchases." ) # 3. Dopamine triggers if len(dopamine_txns) > 3: insights.append( f"🛍️ Dopamine Spending: Detected {len(dopamine_txns)} shopping spikes. " "These purchases often occur in bursts, indicating reward-seeking behavior." ) # 4. Stress/Regret Spending if len(stress_txns) > 0: insights.append( f"⚠️ Emotional Spending: You flagged {len(stress_txns)} transactions as 'regret' or 'stress spending'. " "Implementing a 24-hour cooling-off rule for non-essential items over $100 could help." ) # General fallback if no major insights if not insights: insights.append("📊 Spending Discipline: Your transactions exhibit stable and regular timing, with minimal signs of emotional or impulsive spending.") return { "insights": insights, "metrics": { "late_night_count": len(late_night_txns), "late_night_total": round(sum(t.amount for t in late_night_txns), 2), "weekend_pct": round(weekend_pct, 2), "impulsive_count": len(impulsive_txns), "impulsive_total": round(sum(t.amount for t in impulsive_txns), 2), "dopamine_count": len(dopamine_txns), "stress_count": len(stress_txns), "avg_transaction_amount": round(avg_txn, 2) }, "category_breakdown": {cat: round(amt, 2) for cat, amt in category_totals.items()} }