Spaces:
Build error
Build error
| """ | |
| Smart Budget Planner for BankBot | |
| Categorizes spending and provides budgeting insights | |
| """ | |
| import json | |
| import os | |
| import numpy as np | |
| import pandas as pd | |
| from datetime import datetime, timedelta | |
| from collections import defaultdict | |
| import uuid | |
| BUDGET_FILE = "budgets.json" | |
| # Category keywords for automatic categorization | |
| CATEGORY_KEYWORDS = { | |
| "Food & Dining": ["restaurant", "food", "cafe", "pizza", "burger", "biryani", "zomato", "swiggy", "coffee", "tea", "meal"], | |
| "Shopping": ["shop", "store", "mall", "amazon", "flipkart", "ebay", "retail", "boutique", "apparel", "clothes"], | |
| "Travel": ["uber", "taxi", "bus", "flight", "train", "travel", "hotel", "airline", "booking", "transport"], | |
| "Entertainment": ["movie", "cinema", "game", "netflix", "spotify", "music", "ticket", "concert", "show"], | |
| "Bills & Utilities": ["electricity", "water", "gas", "internet", "mobile", "phone", "bill", "subscription"], | |
| "Healthcare": ["hospital", "doctor", "pharmacy", "medical", "health", "clinic", "medicine"], | |
| "Groceries": ["grocery", "supermarket", "vegetables", "fruits", "milk", "wheat", "bazar"], | |
| "Fitness": ["gym", "yoga", "fitness", "sports", "training", "coach"], | |
| "Insurance": ["insurance", "premium", "policy"], | |
| "Education": ["school", "college", "course", "book", "tuition", "fees"], | |
| "Loan & EMI": ["loan", "emi", "mortgage", "credit"], | |
| "Transfer": ["transfer", "sent", "payment"] | |
| } | |
| class BudgetPlanner: | |
| """Smart budget planning and expense tracking""" | |
| def __init__(self): | |
| self.budgets = self.load_budgets() | |
| def load_budgets(self): | |
| """Load saved budgets from file""" | |
| if os.path.exists(BUDGET_FILE): | |
| try: | |
| with open(BUDGET_FILE, "r", encoding="utf-8") as f: | |
| return json.load(f) | |
| except Exception as e: | |
| print(f"Error loading budgets: {e}") | |
| return {} | |
| return {} | |
| def save_budgets(self): | |
| """Save budgets to file""" | |
| try: | |
| with open(BUDGET_FILE, "w", encoding="utf-8") as f: | |
| json.dump(self.budgets, f, indent=4, ensure_ascii=False) | |
| except Exception as e: | |
| print(f"Error saving budgets: {e}") | |
| def categorize_transaction(self, description, amount=0): | |
| """ | |
| Automatically categorize a transaction based on description | |
| Returns: Category name | |
| """ | |
| description_lower = description.lower() | |
| # Check against keywords | |
| for category, keywords in CATEGORY_KEYWORDS.items(): | |
| if any(keyword in description_lower for keyword in keywords): | |
| return category | |
| # Default category | |
| return "Other" | |
| def set_budget_limit(self, username, category, limit): | |
| """Set budget limit for a spending category""" | |
| if username not in self.budgets: | |
| self.budgets[username] = {} | |
| self.budgets[username][category] = { | |
| "limit": limit, | |
| "created_at": datetime.now().isoformat(), | |
| "alerts": [] | |
| } | |
| self.save_budgets() | |
| def analyze_spending(self, username, transactions, period_days=30): | |
| """ | |
| Analyze spending by category for a given period | |
| Returns: Categorized spending data | |
| """ | |
| if not transactions: | |
| return {} | |
| # Filter transactions from last N days | |
| cutoff_date = datetime.now() - timedelta(days=period_days) | |
| recent_txns = [] | |
| for txn in transactions: | |
| try: | |
| txn_date = datetime.fromisoformat(txn.get('date', '')) | |
| if txn_date > cutoff_date and txn.get('type') == 'debit': | |
| recent_txns.append(txn) | |
| except: | |
| pass | |
| # Categorize transactions | |
| spending_by_category = defaultdict(float) | |
| categorized_txns = defaultdict(list) | |
| for txn in recent_txns: | |
| category = self.categorize_transaction( | |
| txn.get('description', txn.get('details', '')), | |
| float(txn.get('amount', 0)) | |
| ) | |
| amount = float(txn.get('amount', 0)) | |
| spending_by_category[category] += amount | |
| categorized_txns[category].append({ | |
| 'date': txn.get('date'), | |
| 'amount': amount, | |
| 'details': txn.get('details', '') | |
| }) | |
| return { | |
| "period_days": period_days, | |
| "spending_by_category": dict(spending_by_category), | |
| "categorized_transactions": dict(categorized_txns), | |
| "total_spending": sum(spending_by_category.values()), | |
| "transaction_count": len(recent_txns) | |
| } | |
| def check_budget_alerts(self, username, spending_analysis): | |
| """Check if any spending categories exceed their budgets""" | |
| alerts = [] | |
| if username not in self.budgets: | |
| return alerts | |
| user_budgets = self.budgets.get(username, {}) | |
| spending = spending_analysis.get('spending_by_category', {}) | |
| for category, budget_info in user_budgets.items(): | |
| if category not in spending: | |
| continue | |
| spent = spending[category] | |
| limit = budget_info.get('limit', 0) | |
| if spent > limit: | |
| percentage = (spent / limit) * 100 | |
| alerts.append({ | |
| "category": category, | |
| "spent": round(spent, 2), | |
| "limit": limit, | |
| "percentage": round(percentage, 1), | |
| "excess": round(spent - limit, 2), | |
| "severity": "high" if percentage > 150 else "medium" if percentage > 100 else "low", | |
| "timestamp": datetime.now().isoformat() | |
| }) | |
| return alerts | |
| def generate_budget_plan(self, username, transactions, monthly_income=50000): | |
| """Generate recommended budget plan based on spending patterns""" | |
| spending_analysis = self.analyze_spending(username, transactions, period_days=90) | |
| spending = spending_analysis.get('spending_by_category', {}) | |
| total_spending = spending_analysis.get('total_spending', 0) | |
| avg_monthly_spending = total_spending / 3 if total_spending > 0 else 0 | |
| # Calculate budget percentages (50/30/20 rule variant) | |
| recommended_budgets = {} | |
| if spending: | |
| for category, amount in spending.items(): | |
| percentage = (amount / total_spending * 100) if total_spending > 0 else 0 | |
| recommended_budget = (percentage / 100) * monthly_income | |
| recommended_budgets[category] = round(recommended_budget, 2) | |
| # Add default categories if not present | |
| default_categories = { | |
| "Food & Dining": monthly_income * 0.08, | |
| "Shopping": monthly_income * 0.10, | |
| "Travel": monthly_income * 0.08, | |
| "Bills & Utilities": monthly_income * 0.15, | |
| "Entertainment": monthly_income * 0.05, | |
| "Savings": monthly_income * 0.20, | |
| } | |
| for category, amount in default_categories.items(): | |
| if category not in recommended_budgets: | |
| recommended_budgets[category] = amount | |
| return { | |
| "monthly_income": monthly_income, | |
| "current_monthly_avg": round(avg_monthly_spending, 2), | |
| "recommended_budgets": recommended_budgets, | |
| "savings_potential": round(monthly_income - avg_monthly_spending, 2), | |
| "budget_breakdown": { | |
| "essentials": round(monthly_income * 0.50, 2), # Bills, groceries, insurance | |
| "lifestyle": round(monthly_income * 0.30, 2), # Entertainment, dining, shopping | |
| "savings": round(monthly_income * 0.20, 2) # Emergency fund, investments | |
| } | |
| } | |
| def predict_monthly_spending(self, username, transactions): | |
| """ | |
| Predict future spending using historical data | |
| Returns: Predicted spending for next month | |
| """ | |
| if not transactions: | |
| return {} | |
| # Analyze last 3 months | |
| predictions = {} | |
| for period in [30, 60, 90]: | |
| analysis = self.analyze_spending(username, transactions, period_days=period) | |
| spending = analysis.get('spending_by_category', {}) | |
| # Calculate trends | |
| for category, amount in spending.items(): | |
| if category not in predictions: | |
| predictions[category] = [] | |
| predictions[category].append(amount) | |
| # Calculate averages and trends | |
| predicted_spending = {} | |
| for category, amounts in predictions.items(): | |
| if amounts: | |
| predicted_spending[category] = { | |
| "predicted": round(np.mean(amounts), 2), | |
| "trend": "increasing" if amounts[-1] > amounts[0] else "decreasing", | |
| "variance": round(np.std(amounts), 2) | |
| } | |
| return predicted_spending | |
| def get_savings_suggestions(self, username, spending_analysis, monthly_income=50000): | |
| """Generate specific savings suggestions""" | |
| suggestions = [] | |
| spending = spending_analysis.get('spending_by_category', {}) | |
| # Check each category and provide suggestions | |
| for category, amount in spending.items(): | |
| percentage = (amount / monthly_income) * 100 if monthly_income > 0 else 0 | |
| if category == "Food & Dining" and percentage > 10: | |
| reduction = amount - (monthly_income * 0.08) | |
| suggestions.append({ | |
| "category": "Food & Dining", | |
| "potential_savings": round(reduction, 2), | |
| "suggestion": f"You can save ₹{round(reduction, 2)} by reducing dining expenses by 10%", | |
| "priority": "high" if reduction > 1000 else "medium" | |
| }) | |
| elif category == "Shopping" and percentage > 12: | |
| reduction = amount - (monthly_income * 0.10) | |
| suggestions.append({ | |
| "category": "Shopping", | |
| "potential_savings": round(reduction, 2), | |
| "suggestion": f"Reduce impulse purchases to save ₹{round(reduction, 2)} monthly", | |
| "priority": "high" if reduction > 1000 else "medium" | |
| }) | |
| elif category == "Entertainment" and percentage > 7: | |
| reduction = amount - (monthly_income * 0.05) | |
| suggestions.append({ | |
| "category": "Entertainment", | |
| "potential_savings": round(reduction, 2), | |
| "suggestion": f"Optimize subscriptions and entertainment to save ₹{round(reduction, 2)}", | |
| "priority": "low" | |
| }) | |
| # Overall savings tip | |
| total_savings = sum(s.get('potential_savings', 0) for s in suggestions) | |
| if total_savings > 0: | |
| suggestions.append({ | |
| "category": "Total Potential Savings", | |
| "potential_savings": round(total_savings, 2), | |
| "suggestion": f"By following these suggestions, you can save ₹{round(total_savings, 2)} per month", | |
| "priority": "high" | |
| }) | |
| return suggestions | |
| def get_budget_insights(username, transactions, users_data): | |
| """Get comprehensive budget insights for a user""" | |
| planner = BudgetPlanner() | |
| user_data = users_data.get(username, {}) | |
| monthly_income = user_data.get('monthly_income', 50000) | |
| spending_analysis = planner.analyze_spending(username, transactions) | |
| budget_alerts = planner.check_budget_alerts(username, spending_analysis) | |
| budget_plan = planner.generate_budget_plan(username, transactions, monthly_income) | |
| savings_suggestions = planner.get_savings_suggestions(username, spending_analysis, monthly_income) | |
| predicted_spending = planner.predict_monthly_spending(username, transactions) | |
| return { | |
| "spending_analysis": spending_analysis, | |
| "budget_alerts": budget_alerts, | |
| "budget_plan": budget_plan, | |
| "savings_suggestions": savings_suggestions, | |
| "predicted_spending": predicted_spending | |
| } | |