Build a Smart AI Financial Advisor Using Google ADK

# 1. Create project directory
mkdir ai_financial_advisor
cd ai_financial_advisor

# 2. Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# 3. Install dependencies
pip install google-adk numpy-financial google-cloud-aiplatform

import logging
from typing import List, Dict

logger = logging.getLogger(__name__)

# --- 2025 US Federal Tax Brackets (Single Filer) ---
# Source: IRS Inflation Adjustments for Tax Year 2025
TAX_BRACKETS = [
    {"limit": 11925, "rate": 0.10},
    {"limit": 48475, "rate": 0.12},
    {"limit": 103350, "rate": 0.22},
    {"limit": 197300, "rate": 0.24},
    {"limit": 250525, "rate": 0.32},
    {"limit": 626350, "rate": 0.35},
    {"limit": float('inf'), "rate": 0.37},
]

# 2025 Standard Deduction for Single Filers
STANDARD_DEDUCTION = 15750

def _calculate_federal_tax(gross_income):
    """Calculates US federal tax using progressive brackets."""
    # Apply Standard Deduction
    taxable_income = max(0, gross_income - STANDARD_DEDUCTION)
    
    tax_owing = 0
    previous_limit = 0
    
    for bracket in TAX_BRACKETS:
        limit = bracket["limit"]
        rate = bracket["rate"]
        
        # Determine income falling into this bracket
        if taxable_income > previous_limit:
            income_in_bracket = min(taxable_income, limit) - previous_limit
            tax_owing += income_in_bracket * rate
        
        previous_limit = limit
        
        if taxable_income <= limit:
            break
            
    return tax_owing

def calculate_tax_scenario(
    income: float,
    contribution_401k: float = 0.0
) -> dict:
    """
    Calculates estimated federal taxes before and after a Traditional 401(k) contribution.
    """
    try:
        if income < 0 or contribution_401k < 0:
             return {"status": "error", "error_message": "Income and contribution cannot be negative."}

        # Traditional 401(k) reduces current year taxable gross income
        income_after_contrib = max(0, income - contribution_401k)

        tax_before = _calculate_federal_tax(income)
        tax_after = _calculate_federal_tax(income_after_contrib)

        tax_savings = max(0, tax_before - tax_after)

        logger.info(f"Tax Scenario: Income=${income:,.2f}, 401k=${contribution_401k:,.2f}, Savings=${tax_savings:,.2f}")

        return {
            "status": "success",
            "gross_income": income,
            "contribution_401k": contribution_401k,
            "standard_deduction": STANDARD_DEDUCTION,
            "estimated_tax_before_401k": round(tax_before, 2),
            "estimated_tax_after_401k": round(tax_after, 2),
            "estimated_tax_savings": round(tax_savings, 2)
        }
    except Exception as e:
        logger.error(f"Error in calculate_tax_scenario: {e}", exc_info=True)
        return {"status": "error", "error_message": str(e)}

def recommend_contribution_strategy(
    income: float,
    debts: List[dict], 
    available_funds: float = 0.0,
    age: int = 30
) -> dict:
    """
    Recommends a 401(k)/IRA strategy based on debt and 2025 contribution limits.
    """
    try:
        recommendations = []
        remaining_funds = available_funds

        # 2025 Contribution Limits
        LIMIT_401K = 23500
        LIMIT_IRA = 7000
        
        # Catch-up contributions for age 50+
        if age >= 50:
            LIMIT_401K += 7500
            LIMIT_IRA += 1000

        # 1. High-Interest Debt Check (Avalanche Method)
        high_interest_threshold = 7.0 
        high_interest_debts = [d for d in debts if d.get('interest_rate', 0) > high_interest_threshold]
        
        if high_interest_debts:
            high_interest_debts.sort(key=lambda x: x.get('interest_rate', 0), reverse=True)
            top_debt = high_interest_debts[0]
            recommendations.append(
                f"CRITICAL: You have high-interest debt ({top_debt['type']} at {top_debt['interest_rate']}%). "
                f"Prioritize paying this off before maximizing retirement contributions beyond an employer match."
            )
            # Allocate funds to debt first
            debt_payment = min(remaining_funds, top_debt.get('balance', 0))
            remaining_funds -= debt_payment

        # 2. Retirement Contributions
        if remaining_funds > 0:
            rec_401k = min(remaining_funds, LIMIT_401K)
            if rec_401k > 0:
                recommendations.append(
                    f"Contribute ${rec_401k:,.2f} to your 401(k). This lowers your taxable income immediately. "
                    f"2025 Limit is ${LIMIT_401K:,.0f}."
                )
                remaining_funds -= rec_401k
            
            if remaining_funds > 0:
                rec_ira = min(remaining_funds, LIMIT_IRA)
                recommendations.append(
                    f"With remaining funds, consider funding an IRA (Traditional or Roth) up to ${rec_ira:,.2f}. "
                    f"2025 Limit is ${LIMIT_IRA:,.0f}."
                )

        return {
            "status": "success",
            "recommendations": recommendations,
            "limits_used": {"401k_limit": LIMIT_401K, "ira_limit": LIMIT_IRA}
        }
    except Exception as e:
        logger.error(f"Error in strategy: {e}")
        return {"status": "error", "error_message": str(e)}

import logging
import numpy_financial as npf

logger = logging.getLogger(__name__)

def _calculate_monthly_payment(principal, annual_interest_rate, years):
    """Calculates the fixed monthly mortgage payment (P&I)."""
    if principal <= 0 or annual_interest_rate < 0 or years <= 0:
        return 0
    monthly_rate = (annual_interest_rate / 100) / 12
    num_payments = years * 12
    # Returns negative value, so we negate it
    return -npf.pmt(monthly_rate, num_payments, principal)

def analyze_mortgage_profile(
    principal: float,
    annual_interest_rate: float,
    remaining_years: float
) -> dict:
    """
    Analyzes current mortgage status.
    """
    try:
        monthly_payment = _calculate_monthly_payment(principal, annual_interest_rate, remaining_years)
        total_paid = monthly_payment * remaining_years * 12
        total_interest = total_paid - principal

        return {
            "status": "success",
            "current_principal": principal,
            "interest_rate": annual_interest_rate,
            "monthly_payment": round(monthly_payment, 2),
            "total_interest_remaining": round(total_interest, 2)
        }
    except Exception as e:
        return {"status": "error", "error_message": str(e)}

def generate_acceleration_scenario(
    principal: float,
    annual_interest_rate: float,
    remaining_years: float,
    lump_sum_payment: float = 0.0
) -> dict:
    """
    Calculates impact of applying a lump sum (e.g., tax refund) to principal.
    """
    try:
        # Baseline
        orig_payment = _calculate_monthly_payment(principal, annual_interest_rate, remaining_years)
        orig_interest = (orig_payment * remaining_years * 12) - principal

        # New Scenario: Reduce principal immediately
        new_principal = principal - lump_sum_payment
        if new_principal <= 0:
             return {"status": "success", "message": "Mortgage fully paid off!"}

        monthly_rate = (annual_interest_rate / 100) / 12
        
        # Calculate new number of periods (nper) keeping payment same
        new_num_payments = npf.nper(monthly_rate, -orig_payment, new_principal)
        new_years = new_num_payments / 12
        
        new_total_cost = (orig_payment * new_num_payments)
        new_interest = new_total_cost - new_principal

        interest_saved = orig_interest - new_interest

        return {
            "status": "success",
            "lump_sum_applied": lump_sum_payment,
            "original_payoff_years": remaining_years,
            "new_payoff_years": round(new_years, 2),
            "years_saved": round(remaining_years - new_years, 2),
            "interest_saved": round(interest_saved, 2)
        }
    except Exception as e:
        return {"status": "error", "error_message": str(e)}

from google.adk.agents import Agent
from tools.tax_tools import calculate_tax_scenario, recommend_contribution_strategy
from tools.mortgage_tools import analyze_mortgage_profile, generate_acceleration_scenario

# Define the tools list
financial_tools = [
    calculate_tax_scenario,
    recommend_contribution_strategy,
    analyze_mortgage_profile,
    generate_acceleration_scenario
]

# Configure the Agent
advisor_agent = Agent(
    name="us_financial_advisor",
    model="gemini-2.5-flash",  # Fast, efficient model
    description="A US-focused financial advisor for tax and debt planning.",
    instruction=(
        "You are an expert US Financial Advisor. Your goal is to optimize the user's financial health "
        "by minimizing federal tax liability and efficiently managing debt.\n\n"
        
        "RULES & BEHAVIORS:\n"
        "1. ALWAYS prioritize high-interest debt (Credit Cards, Personal Loans > 7%) over investing.\n"
        "2. Use 'calculate_tax_scenario' to show the specific dollar benefit of 401(k) contributions.\n"
        "3. If the user gets a tax refund (savings), suggest applying it to their mortgage using 'generate_acceleration_scenario' IF they plan to stay in the home long-term.\n"
        "4. All tax references must use 2025 IRS rules. The standard deduction is $15,750 for singles.\n"
        "5. Explain concepts simply. Do not just output numbers; explain WHY a strategy works."
    ),
    tools=financial_tools
)

ai_financial_advisor/
├── venv/
├── backend/
│   ├── __init__.py
│   ├── agents/
│   │   ├── __init__.py
│   │   └── financial_advisor_agent.py
│   └── tools/
│       ├── __init__.py
│       ├── tax_tools.py
│       └── mortgage_tools.py
└── main.py  # Entry point script