Spaces:
Paused
Paused
| import gradio as gr | |
| import pandas as pd | |
| import numpy as np | |
| import matplotlib.pyplot as plt | |
| import seaborn as sns | |
| from datetime import datetime, timedelta | |
| import json | |
| import re | |
| from typing import Dict, List, Tuple | |
| import io | |
| import base64 | |
| # Fixed imports - removed HfApiModel since you don't need it | |
| from smolagents import Tool | |
| class ExpenseCategorizer(Tool): | |
| name = "expense_categorizer" | |
| description = "Categorizes transactions into spending categories based on description and amount" | |
| inputs = { | |
| "description": { | |
| "type": "string", # Fixed from "str" | |
| "description": "Transaction description text" | |
| }, | |
| "amount": { | |
| "type": "number", # Fixed from "float" | |
| "description": "Transaction amount in dollars" | |
| } | |
| } | |
| output_type = "string" # Fixed from "str" | |
| def __init__(self): | |
| super().__init__() | |
| self.categories = { | |
| 'groceries': ['grocery', 'supermarket', 'walmart', 'target', 'costco', 'food', 'market'], | |
| 'utilities': ['electric', 'gas', 'water', 'internet', 'phone', 'cable', 'utility'], | |
| 'transportation': ['gas station', 'uber', 'lyft', 'taxi', 'metro', 'bus', 'parking'], | |
| 'entertainment': ['netflix', 'spotify', 'movie', 'theater', 'game', 'entertainment'], | |
| 'dining': ['restaurant', 'cafe', 'pizza', 'mcdonald', 'starbucks', 'dining'], | |
| 'shopping': ['amazon', 'ebay', 'mall', 'store', 'shop', 'retail'], | |
| 'healthcare': ['pharmacy', 'doctor', 'hospital', 'medical', 'health'], | |
| 'other': [] | |
| } | |
| def forward(self, description: str, amount: float) -> str: | |
| description_lower = description.lower() | |
| for category, keywords in self.categories.items(): | |
| if any(keyword in description_lower for keyword in keywords): | |
| return category | |
| return 'other' | |
| class BudgetAnalyzer(Tool): | |
| name = "budget_analyzer" | |
| description = "Analyzes budget data and provides insights on spending patterns, savings, and recommendations for budget optimization." | |
| inputs = { | |
| "df_json": { | |
| "type": "string", # Fixed from "str" | |
| "description": "JSON representation of transaction data" | |
| }, | |
| "monthly_income": { | |
| "type": "number", # Fixed from "float" | |
| "description": "Monthly income in dollars" | |
| } | |
| } | |
| output_type = "object" # Fixed from "dict" | |
| def forward(self, df_json: str, monthly_income: float) -> Dict: | |
| df = pd.read_json(df_json) | |
| monthly_spending = df.groupby('category')['amount'].sum().to_dict() | |
| total_spending = sum(monthly_spending.values()) | |
| recommended_budget = { | |
| 'needs': monthly_income * 0.5, | |
| 'wants': monthly_income * 0.3, | |
| 'savings': monthly_income * 0.2 | |
| } | |
| need_categories = ['groceries', 'utilities', 'transportation', 'healthcare'] | |
| want_categories = ['entertainment', 'dining', 'shopping'] | |
| current_needs = sum(monthly_spending.get(cat, 0) for cat in need_categories) | |
| current_wants = sum(monthly_spending.get(cat, 0) for cat in want_categories) | |
| current_savings = monthly_income - total_spending | |
| analysis = { | |
| "total_spending": total_spending, | |
| "spending_by_category": monthly_spending, | |
| "current_allocation": { | |
| 'needs': current_needs, | |
| 'wants': current_wants, | |
| 'savings': current_savings | |
| }, | |
| 'recommended_budget': recommended_budget, | |
| 'budget_variance': { | |
| 'needs': current_needs - recommended_budget['needs'], | |
| 'wants': current_wants - recommended_budget['wants'], | |
| 'savings': current_savings - recommended_budget['savings'] | |
| } | |
| } | |
| return analysis | |
| class SpendingAlerts(Tool): | |
| name = "spending_alerts" | |
| description = "Monitors spending patterns and alerts when spending exceeds predefined thresholds." | |
| inputs = { | |
| "analysis": { | |
| "type": "object", # Fixed from "dict" | |
| "description": "Budget analysis results" | |
| }, | |
| "df_json": { | |
| "type": "string", # Fixed from "str" | |
| "description": "JSON representation of transaction data" | |
| } | |
| } | |
| output_type = "array" # Fixed from "list" | |
| def forward(self, analysis: Dict, df_json: str) -> List[str]: | |
| df = pd.read_json(df_json) | |
| alerts = [] | |
| variance = analysis['budget_variance'] | |
| if variance['needs'] > 0: | |
| alerts.append(f"⚠️ You're overspending on needs by ${variance['needs']:.2f}") | |
| if variance['wants'] > 0: | |
| alerts.append(f"⚠️ You're overspending on wants by ${variance['wants']:.2f}") | |
| if variance['savings'] < 0: | |
| alerts.append(f"📉 You're saving ${abs(variance['savings']):.2f} less than recommended") | |
| large_transactions = df[df['amount'] > df['amount'].quantile(0.95)] | |
| if not large_transactions.empty: | |
| alerts.append(f"🚨 Large transactions detected: {len(large_transactions)}") | |
| df['date'] = pd.to_datetime(df['date']) | |
| recent_week = df[df['date'] >= df['date'].max() - pd.Timedelta(days=7)] | |
| if not recent_week.empty: | |
| recent_spending = recent_week['amount'].sum() | |
| avg_weekly = df['amount'].sum() / 4 | |
| if recent_spending > avg_weekly * 1.5: | |
| alerts.append(f"📈 Recent spending is 50% higher than your weekly average") | |
| return alerts if alerts else ["✅ No spending alerts - you're doing great!"] | |
| def process_transactions(file, monthly_income): | |
| try: | |
| if file is None: | |
| return "Please upload a CSV file", None, None | |
| # Read CSV with better error handling and parsing | |
| try: | |
| # Try reading with different separators and quote handling | |
| df = pd.read_csv(file.name) | |
| # If the CSV was read as a single column, try parsing differently | |
| if len(df.columns) == 1: | |
| # Try reading with different delimiter detection | |
| import csv | |
| with open(file.name, 'r') as f: | |
| sample = f.read(1024) | |
| sniffer = csv.Sniffer() | |
| delimiter = sniffer.sniff(sample).delimiter | |
| df = pd.read_csv(file.name, delimiter=delimiter) | |
| except Exception as e: | |
| print(f"Standard CSV reading failed: {e}") | |
| # Fallback method for problematic CSV files | |
| try: | |
| with open(file.name, 'r') as f: | |
| content = f.read() | |
| lines = content.strip().split('\n') | |
| # Parse header | |
| header = lines[0].split(',') | |
| header = [col.strip().strip('"') for col in header] | |
| # Parse data rows | |
| data = [] | |
| for line in lines[1:]: | |
| if line.strip(): | |
| # Split by comma and clean each field | |
| row = [field.strip().strip('"') for field in line.split(',')] | |
| if len(row) >= len(header): | |
| data.append(row[:len(header)]) | |
| df = pd.DataFrame(data, columns=header) | |
| except Exception as e2: | |
| return f"Failed to parse CSV file: {str(e2)}", None, None | |
| print(f"DataFrame shape: {df.shape}") | |
| print(f"DataFrame columns: {df.columns.tolist()}") | |
| print(f"First few rows:\n{df.head()}") | |
| # Clean column names | |
| df.columns = df.columns.str.strip().str.replace('"', '') | |
| required_cols = ['date', 'description', 'amount'] | |
| if not all(col in df.columns for col in required_cols): | |
| available_cols = df.columns.tolist() | |
| return f"CSV file must contain columns: {required_cols}. Found columns: {available_cols}", None, None | |
| # Clean and convert amount column | |
| df['amount'] = df['amount'].astype(str).str.replace('"', '') | |
| df['amount'] = df['amount'].astype(float) | |
| df['amount'] = df['amount'].abs() # Convert to positive values | |
| # Clean description column | |
| df['description'] = df['description'].astype(str).str.replace('"', '').str.strip() | |
| print(f"Cleaned data sample:\n{df[['description', 'amount']].head()}") | |
| # Categorize transactions | |
| categorizer = ExpenseCategorizer() | |
| df['category'] = df.apply(lambda row: categorizer.forward(row['description'], row['amount']), axis=1) | |
| print(f"Categories assigned:\n{df['category'].value_counts()}") | |
| # Analyze budget | |
| analyzer = BudgetAnalyzer() | |
| analysis = analyzer.forward(df.to_json(), float(monthly_income)) | |
| # Get alerts | |
| alerts_tool = SpendingAlerts() | |
| alerts = alerts_tool.forward(analysis, df.to_json()) | |
| # Create charts | |
| print("Creating spending chart...") | |
| fig1 = create_spending_charts(analysis['spending_by_category']) | |
| print("Creating budget comparison chart...") | |
| fig2 = create_budget_comparison(analysis) | |
| results = format_analysis_results(analysis, alerts) | |
| print("Analysis complete!") | |
| return results, fig1, fig2 | |
| except Exception as e: | |
| import traceback | |
| error_details = traceback.format_exc() | |
| return f"Error processing file: {str(e)}\n\nDetails:\n{error_details}", None, None | |
| def create_spending_charts(spending_by_category): | |
| try: | |
| # Clear any existing plots | |
| plt.close('all') | |
| # Filter out zero values | |
| filtered_spending = {k: v for k, v in spending_by_category.items() if v > 0} | |
| print(f"Spending data for chart: {filtered_spending}") | |
| if not filtered_spending: | |
| print("No spending data to plot") | |
| return None | |
| fig, ax = plt.subplots(figsize=(10, 6)) | |
| categories = list(filtered_spending.keys()) | |
| amounts = list(filtered_spending.values()) | |
| # Create pie chart | |
| wedges, texts, autotexts = ax.pie(amounts, labels=categories, autopct='%1.1f%%', startangle=90) | |
| ax.set_title('Spending by Category', fontsize=16, fontweight='bold') | |
| # Improve text readability | |
| for autotext in autotexts: | |
| autotext.set_color('white') | |
| autotext.set_fontweight('bold') | |
| plt.tight_layout() | |
| print("Spending chart created successfully") | |
| return fig | |
| except Exception as e: | |
| print(f"Error creating spending chart: {str(e)}") | |
| return None | |
| def create_budget_comparison(analysis): | |
| try: | |
| # Clear any existing plots | |
| plt.close('all') | |
| fig, ax = plt.subplots(figsize=(12, 8)) | |
| categories = ['Needs', 'Wants', 'Savings'] | |
| current = [ | |
| analysis['current_allocation']['needs'], | |
| analysis['current_allocation']['wants'], | |
| analysis['current_allocation']['savings'] | |
| ] | |
| recommended = [ | |
| analysis['recommended_budget']['needs'], | |
| analysis['recommended_budget']['wants'], | |
| analysis['recommended_budget']['savings'] | |
| ] | |
| print(f"Budget comparison data - Current: {current}, Recommended: {recommended}") | |
| x = np.arange(len(categories)) | |
| width = 0.35 | |
| bars1 = ax.bar(x - width/2, current, width, label='Current', alpha=0.8, color='skyblue') | |
| bars2 = ax.bar(x + width/2, recommended, width, label='Recommended', alpha=0.8, color='lightcoral') | |
| # Add value labels on bars | |
| def add_value_labels(bars): | |
| for bar in bars: | |
| height = bar.get_height() | |
| ax.annotate(f'${height:.0f}', | |
| xy=(bar.get_x() + bar.get_width() / 2, height), | |
| xytext=(0, 3), # 3 points vertical offset | |
| textcoords="offset points", | |
| ha='center', va='bottom', | |
| fontweight='bold') | |
| add_value_labels(bars1) | |
| add_value_labels(bars2) | |
| ax.set_xlabel('Budget Categories', fontsize=12, fontweight='bold') | |
| ax.set_ylabel('Amount ($)', fontsize=12, fontweight='bold') | |
| ax.set_title('Current vs Recommended Budget Allocation', fontsize=16, fontweight='bold') | |
| ax.set_xticks(x) | |
| ax.set_xticklabels(categories) | |
| ax.legend() | |
| ax.grid(True, alpha=0.3) | |
| plt.tight_layout() | |
| print("Budget comparison chart created successfully") | |
| return fig | |
| except Exception as e: | |
| print(f"Error creating budget comparison chart: {str(e)}") | |
| return None | |
| def format_analysis_results(analysis, alerts): | |
| results = f""" | |
| ## 📊 Financial Analysis Results | |
| ### 💰 Total Monthly Spending: ${analysis['total_spending']:.2f} | |
| ### 📈 Spending Breakdown: | |
| """ | |
| for category, amount in analysis['spending_by_category'].items(): | |
| if amount > 0: # Only show categories with spending | |
| percentage = (amount / analysis['total_spending']) * 100 | |
| results += f"- **{category.title()}**: ${amount:.2f} ({percentage:.1f}%)\n" | |
| results += f""" | |
| ### 🎯 Budget Allocation Analysis: | |
| - **Current Needs**: ${analysis['current_allocation']['needs']:.2f} | |
| - **Current Wants**: ${analysis['current_allocation']['wants']:.2f} | |
| - **Current Savings**: ${analysis['current_allocation']['savings']:.2f} | |
| ### 📋 Alerts & Recommendations: | |
| """ | |
| for alert in alerts: | |
| results += f"- {alert}\n" | |
| return results | |
| def chat_with_agent(message, history): | |
| if "save" in message.lower(): | |
| response = "💡 Here are some saving tips:\n- Set up automatic transfers to savings\n- Use the 50/30/20 budgeting rule\n- Track your expenses regularly\n- Cut unnecessary subscriptions" | |
| elif "budget" in message.lower(): | |
| response = "📊 Budget management tips:\n- Track all income and expenses\n- Categorize your spending\n- Set realistic spending limits\n- Review and adjust monthly" | |
| elif "invest" in message.lower(): | |
| response = "📈 Investment basics:\n- Start with emergency fund first\n- Consider low-cost index funds\n- Diversify your portfolio\n- Think long-term" | |
| else: | |
| response = f"I understand you want to discuss: {message}. Here are some general financial tips:\n- Track your spending\n- Create a budget\n- Build an emergency fund\n- Pay off high-interest debt" | |
| # For the new message format, we need to return the updated history | |
| history.append({"role": "user", "content": message}) | |
| history.append({"role": "assistant", "content": response}) | |
| return "", history | |
| def create_interface(): | |
| # Option 1: Using Monochrome dark theme | |
| with gr.Blocks(title="Personal Finance AI Agent", theme=gr.themes.Monochrome(primary_hue="indigo", neutral_hue="slate")) as demo: | |
| # OR Option 2: Using Base dark theme | |
| # with gr.Blocks(title="Personal Finance AI Agent", theme=gr.themes.Base(primary_hue="blue")) as demo: | |
| gr.Markdown("# 🤖 Personal Finance AI Agent") | |
| gr.Markdown("Upload your transaction data and get AI-powered financial insights!") | |
| with gr.Tab("📊 Transaction Analysis"): | |
| with gr.Row(): | |
| with gr.Column(): | |
| file_input = gr.File( | |
| label="Upload CSV File", | |
| file_types=[".csv"] | |
| ) | |
| gr.Markdown("CSV file must contain columns: ['date', 'description', 'amount']") | |
| income_input = gr.Number( | |
| label="Monthly Income ($)", | |
| value=5000, | |
| minimum=0 | |
| ) | |
| analyze_btn = gr.Button("Analyze Transactions", variant="primary") | |
| with gr.Column(): | |
| results_output = gr.Markdown(label="Analysis Results") | |
| with gr.Row(): | |
| spending_chart = gr.Plot(label="Spending by Category") | |
| budget_chart = gr.Plot(label="Budget Comparison") | |
| analyze_btn.click( | |
| process_transactions, | |
| inputs=[file_input, income_input], | |
| outputs=[results_output, spending_chart, budget_chart] | |
| ) | |
| with gr.Tab("💬 Financial Chat"): | |
| chatbot = gr.Chatbot(height=400, type="messages") | |
| msg = gr.Textbox( | |
| label="Ask me about budgeting, saving, or investing", | |
| placeholder="How can I save more money this month?" | |
| ) | |
| msg.submit(chat_with_agent, [msg, chatbot], [msg, chatbot]) | |
| with gr.Tab("📝 Sample Data"): | |
| gr.Markdown(""" | |
| ### Sample CSV Format: | |
| ``` | |
| date,description,amount | |
| 2024-01-01,Grocery Store,-150.00 | |
| 2024-01-02,Netflix Subscription,-15.99 | |
| 2024-01-03,Gas Station,-45.00 | |
| 2024-01-05,Restaurant Dinner,-80.00 | |
| 2024-01-07,Electric Bill,-120.00 | |
| ``` | |
| ### Installation: | |
| ```bash | |
| pip install gradio pandas matplotlib seaborn numpy smolagents | |
| ``` | |
| ### Features: | |
| - 🏷️ Automatic transaction categorization | |
| - 📊 Budget analysis with 50/30/20 rule | |
| - ⚠️ Spending alerts and recommendations | |
| - 📈 Interactive visualizations | |
| - 💬 Financial chat | |
| """) | |
| return demo | |
| if __name__ == "__main__": | |
| demo = create_interface() | |
| demo.launch(debug=True, share=True) |