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)