Deploy CFA AI Agent with Finance-Llama-8B

- Added complete financial analysis agent with LangChain integration
- Implemented comprehensive financial tools (DCF, ratios, risk metrics)
- Added real-time market data fetching capabilities
- Optimized model loading for memory efficiency with 8-bit quantization
- Updated requirements.txt for HF Spaces compatibility
- Enhanced README with detailed feature descriptions

🤖 Generated with Claude Code

Co-Authored-By: Claude <noreply@anthropic.com>

README.md

@@ -1,12 +1,53 @@
 ---
-title: CFA Ai Agent
-emoji: 👁
-colorFrom: pink
-colorTo: yellow
-sdk: gradio
-sdk_version: 5.49.1
+title: CFA AI Agent
+emoji: 📊
+colorFrom: blue
+colorTo: green
+sdk: chainlit
+sdk_version: 1.0.0
 app_file: app.py
 pinned: false
+python_version: 3.11
+models:
+  - tarun7r/Finance-Llama-8B
+license: mit
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# 📊 CFA AI Agent
+
+A specialized financial analysis AI agent powered by Finance-Llama-8B and equipped with comprehensive financial tools.
+
+## 🚀 Features
+
+- **Advanced Financial Analysis**: DCF valuation, ratio analysis, risk assessment
+- **Real-time Market Data**: Stock prices, historical data, financial statements
+- **Portfolio Management**: Beta calculation, WACC, Sharpe ratio analysis
+- **Interactive Chat Interface**: Powered by Chainlit for seamless user experience
+
+## 💼 Capabilities
+
+- Discounted Cash Flow (DCF) valuation
+- Financial ratio analysis and comparison
+- Risk metrics calculation (Beta, Sharpe ratio)
+- Weighted Average Cost of Capital (WACC)
+- Stock price analysis and comparison
+- Market data fetching and analysis
+
+## 🔧 Built With
+
+- **Model**: Finance-Llama-8B (specialized for financial analysis)
+- **Framework**: LangChain for agent orchestration
+- **Interface**: Chainlit for web interface
+- **Data**: Yahoo Finance for real-time market data
+
+## 📈 Use Cases
+
+Perfect for financial professionals, students, and anyone needing:
+- Investment analysis and valuation
+- Portfolio risk assessment
+- Financial statement analysis
+- Market research and comparison
+
+---
+
+*This is a CFA-level financial analysis tool designed to assist with professional financial analysis tasks.*

agent.py

@@ -0,0 +1,329 @@
+"""
+CFA AI Agent - LangChain Agent Setup
+This module sets up the LangChain agent with Finance-Llama-8B model and financial tools.
+"""
+
+import os
+import torch
+from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
+from langchain_community.llms import HuggingFacePipeline
+from langchain.agents import initialize_agent, AgentType
+from langchain.memory import ConversationBufferMemory
+from langchain.schema import SystemMessage
+from langchain.prompts import MessagesPlaceholder
+from typing import List, Any, Optional
+
+# Import our custom tools
+from tools.finance_tools import (
+    calculate_dcf,
+    calculate_sharpe_ratio,
+    compare_pe_ratios,
+    calculate_beta,
+    calculate_wacc,
+    financial_ratios_analysis
+)
+from tools.data_fetcher import (
+    get_stock_price,
+    get_historical_data,
+    get_company_info,
+    get_financial_statements,
+    get_market_indices,
+    compare_stocks
+)
+
+
+class CFAAgent:
+    """
+    CFA AI Agent that combines Finance-Llama-8B model with financial analysis tools.
+    """
+
+    def __init__(self, model_name: str = "tarun7r/Finance-Llama-8B"):
+        """
+        Initialize the CFA Agent with model and tools.
+
+        Args:
+            model_name: Hugging Face model name for financial analysis
+        """
+        self.model_name = model_name
+        self.tokenizer = None
+        self.model = None
+        self.llm = None
+        self.agent = None
+        self.memory = None
+        self._setup_model()
+        self._setup_tools()
+        self._setup_agent()
+
+    def _setup_model(self):
+        """Load and setup the Finance-Llama-8B model."""
+        try:
+            print(f"Loading model: {self.model_name}")
+
+            # Check if CUDA is available
+            device = "cuda" if torch.cuda.is_available() else "cpu"
+            print(f"Using device: {device}")
+
+            # Load tokenizer
+            self.tokenizer = AutoTokenizer.from_pretrained(
+                self.model_name,
+                trust_remote_code=True
+            )
+
+            # Add pad token if not present
+            if self.tokenizer.pad_token is None:
+                self.tokenizer.pad_token = self.tokenizer.eos_token
+
+            # Load model with appropriate settings and memory optimization
+            if device == "cuda":
+                self.model = AutoModelForCausalLM.from_pretrained(
+                    self.model_name,
+                    torch_dtype=torch.float16,
+                    device_map="auto",
+                    trust_remote_code=True,
+                    low_cpu_mem_usage=True,
+                    load_in_8bit=True,  # Enable 8-bit quantization for memory efficiency
+                    max_memory={0: "6GB"}  # Limit GPU memory usage
+                )
+            else:
+                # For CPU, use aggressive memory optimization
+                self.model = AutoModelForCausalLM.from_pretrained(
+                    self.model_name,
+                    trust_remote_code=True,
+                    low_cpu_mem_usage=True,
+                    torch_dtype=torch.float32,
+                    device_map="cpu",
+                    max_memory={"cpu": "8GB"}  # Limit CPU memory usage
+                )
+
+            # Create pipeline
+            pipe = pipeline(
+                "text-generation",
+                model=self.model,
+                tokenizer=self.tokenizer,
+                max_new_tokens=512,
+                temperature=0.1,
+                do_sample=True,
+                pad_token_id=self.tokenizer.eos_token_id,
+                repetition_penalty=1.1
+            )
+
+            # Wrap in LangChain
+            self.llm = HuggingFacePipeline(pipeline=pipe)
+            print("✅ Model loaded successfully")
+
+        except Exception as e:
+            print(f"❌ Error loading model: {str(e)}")
+            # Fallback to a smaller model or OpenAI if Finance-Llama-8B fails
+            self._setup_fallback_model()
+
+    def _setup_fallback_model(self):
+        """Setup a fallback model if Finance-Llama-8B fails to load."""
+        try:
+            print("Setting up fallback model...")
+            from langchain_community.llms import OpenAI
+
+            # Check for OpenAI API key
+            if os.getenv("OPENAI_API_KEY"):
+                self.llm = OpenAI(
+                    temperature=0.1,
+                    model_name="gpt-3.5-turbo-instruct",
+                    max_tokens=512
+                )
+                print("✅ Using OpenAI GPT-3.5 as fallback")
+            else:
+                raise ValueError("No OpenAI API key found")
+
+        except Exception as e:
+            print(f"❌ Fallback model failed: {str(e)}")
+            # Last resort: use a very small local model
+            try:
+                pipe = pipeline(
+                    "text-generation",
+                    model="distilgpt2",
+                    max_new_tokens=256,
+                    temperature=0.7
+                )
+                self.llm = HuggingFacePipeline(pipeline=pipe)
+                print("✅ Using DistilGPT2 as emergency fallback")
+            except Exception as final_e:
+                raise RuntimeError(f"All model loading attempts failed: {final_e}")
+
+    def _setup_tools(self):
+        """Setup all available financial analysis tools."""
+        self.tools = [
+            # Finance calculation tools
+            calculate_dcf,
+            calculate_sharpe_ratio,
+            compare_pe_ratios,
+            calculate_beta,
+            calculate_wacc,
+            financial_ratios_analysis,
+
+            # Data fetching tools
+            get_stock_price,
+            get_historical_data,
+            get_company_info,
+            get_financial_statements,
+            get_market_indices,
+            compare_stocks
+        ]
+        print(f"✅ Loaded {len(self.tools)} financial analysis tools")
+
+    def _setup_agent(self):
+        """Setup the LangChain agent with memory and tools."""
+        try:
+            # Setup conversation memory
+            self.memory = ConversationBufferMemory(
+                memory_key="chat_history",
+                return_messages=True,
+                output_key="output"
+            )
+
+            # Initialize agent
+            self.agent = initialize_agent(
+                tools=self.tools,
+                llm=self.llm,
+                agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
+                memory=self.memory,
+                verbose=True,
+                handle_parsing_errors=True,
+                max_iterations=3,
+                early_stopping_method="generate"
+            )
+
+            # Add system message for financial context
+            system_message = """You are a CFA (Chartered Financial Analyst) AI assistant specialized in financial analysis, investment valuation, and portfolio management.
+
+Your expertise includes:
+- Financial statement analysis and ratio calculations
+- Valuation models (DCF, comparable company analysis, etc.)
+- Risk assessment and portfolio theory
+- Market analysis and economic indicators
+- Investment recommendations based on fundamental analysis
+
+When answering questions:
+1. Use the available financial tools to fetch real data when needed
+2. Provide clear, professional explanations suitable for CFA-level analysis
+3. Show your calculations and reasoning
+4. Consider both quantitative and qualitative factors
+5. Acknowledge limitations and assumptions in your analysis
+
+You have access to real-time financial data and calculation tools. Use them to provide accurate, data-driven insights."""
+
+            # Store system message for context
+            self.system_message = system_message
+            print("✅ Agent initialized successfully")
+
+        except Exception as e:
+            print(f"❌ Error setting up agent: {str(e)}")
+            raise
+
+    def query(self, question: str) -> str:
+        """
+        Process a financial query using the CFA agent.
+
+        Args:
+            question: User's financial question or request
+
+        Returns:
+            Agent's response with analysis and recommendations
+        """
+        try:
+            # Enhance the question with context
+            enhanced_question = f"""As a CFA analyst, please help with the following:
+
+{question}
+
+Please provide a thorough analysis using available data and financial tools. Show your work and explain your reasoning."""
+
+            # Get response from agent
+            response = self.agent.run(enhanced_question)
+            return response
+
+        except Exception as e:
+            error_msg = f"Error processing query: {str(e)}"
+            print(error_msg)
+            return error_msg
+
+    def get_conversation_history(self) -> List[Any]:
+        """Get the current conversation history."""
+        if self.memory:
+            return self.memory.chat_memory.messages
+        return []
+
+    def clear_memory(self):
+        """Clear the conversation memory."""
+        if self.memory:
+            self.memory.clear()
+            print("✅ Conversation memory cleared")
+
+    def get_available_tools(self) -> List[str]:
+        """Get list of available tool names."""
+        return [tool.name for tool in self.tools]
+
+    def health_check(self) -> dict:
+        """Perform a health check of the agent components."""
+        status = {
+            "model_loaded": self.model is not None,
+            "llm_ready": self.llm is not None,
+            "agent_ready": self.agent is not None,
+            "memory_ready": self.memory is not None,
+            "tools_count": len(self.tools),
+            "device": "cuda" if torch.cuda.is_available() else "cpu"
+        }
+        return status
+
+
+def create_cfa_agent(model_name: str = "tarun7r/Finance-Llama-8B") -> CFAAgent:
+    """
+    Factory function to create and return a CFA Agent instance.
+
+    Args:
+        model_name: Hugging Face model name for financial analysis
+
+    Returns:
+        Initialized CFAAgent instance
+    """
+    try:
+        agent = CFAAgent(model_name=model_name)
+        print("🎯 CFA Agent created successfully")
+        return agent
+    except Exception as e:
+        print(f"❌ Failed to create CFA Agent: {str(e)}")
+        raise
+
+
+# Example usage and testing
+if __name__ == "__main__":
+    print("🚀 Initializing CFA AI Agent...")
+
+    try:
+        # Create agent
+        cfa_agent = create_cfa_agent()
+
+        # Health check
+        health = cfa_agent.health_check()
+        print("📊 Health Check Results:")
+        for key, value in health.items():
+            status = "✅" if value else "❌"
+            print(f"  {status} {key}: {value}")
+
+        # Test queries
+        test_queries = [
+            "What is the current stock price of Apple (AAPL)?",
+            "Calculate the PE ratio comparison between Apple and Microsoft",
+            "Explain the CAPM model in simple terms"
+        ]
+
+        print("\n🧪 Running test queries...")
+        for i, query in enumerate(test_queries, 1):
+            print(f"\n--- Test Query {i} ---")
+            print(f"Q: {query}")
+            try:
+                response = cfa_agent.query(query)
+                print(f"A: {response}")
+            except Exception as e:
+                print(f"❌ Query failed: {str(e)}")
+
+    except Exception as e:
+        print(f"❌ CFA Agent initialization failed: {str(e)}")

app.py

@@ -0,0 +1,166 @@
+"""
+CFA AI Agent - Chainlit Chat Interface
+Main application file for the CFA AI Agent web interface.
+"""
+
+import chainlit as cl
+import pandas as pd
+import numpy as np
+import plotly.graph_objects as go
+import plotly.express as px
+from datetime import datetime, timedelta
+import sys
+import os
+import asyncio
+
+# Add the current directory to Python path for imports
+sys.path.append(os.path.dirname(os.path.abspath(__file__)))
+
+try:
+    from agent import create_cfa_agent, CFAAgent
+    from tools.data_fetcher import get_stock_price, get_market_indices
+except ImportError as e:
+    print(f"Import error: {e}")
+    raise
+
+
+# Global agent instance
+agent = None
+
+
+@cl.on_chat_start
+async def start():
+    """Initialize the CFA agent when chat starts."""
+    global agent
+
+    await cl.Message(
+        content="🤖 **Welcome to CFA AI Agent!**\n\nI'm your professional financial analysis assistant powered by Finance-Llama-8B and real-time market data.\n\n**What I can help you with:**\n\n📊 **Market Data & Analysis:**\n• Real-time stock prices and company information\n• Historical data analysis and trends\n• Market indices and sector performance\n• Stock comparisons and valuations\n\n🧮 **Financial Calculations:**\n• DCF (Discounted Cash Flow) valuations\n• Risk metrics (Sharpe ratio, Beta, volatility)\n• Financial ratios analysis\n• WACC and cost of capital calculations\n\n📚 **CFA Knowledge:**\n• Investment concepts and theories\n• Portfolio management principles\n• Financial statement analysis\n• Derivatives and fixed income\n\n**Example queries:**\n• \"What is the current stock price of Apple (AAPL)?\"\n• \"Perform a DCF valuation for Tesla with 10% growth rate\"\n• \"Explain the CAPM model with examples\"\n• \"Compare the PE ratios of Apple and Microsoft\"\n\nInitializing the agent... This may take a moment on first run."
+    ).send()
+
+    try:
+        # Show loading message
+        loading_msg = cl.Message(content="🔄 Loading CFA AI Agent...")
+        await loading_msg.send()
+
+        # Initialize the agent
+        agent = create_cfa_agent()
+
+        # Update loading message
+        loading_msg.content = "✅ **CFA AI Agent is ready!**\n\nYou can now ask me anything about finance, stocks, valuations, or CFA topics. Try asking about a specific stock or financial concept!"
+        await loading_msg.update()
+
+        # Add market overview as an action
+        await display_market_overview()
+
+        # Add action buttons
+        actions = [
+            cl.Action(name="market_summary", value="market_summary", label="📈 Market Summary"),
+            cl.Action(name="stock_analysis", value="stock_analysis", label="🔍 Stock Analysis"),
+            cl.Action(name="cfa_concept", value="cfa_concept", label="📚 CFA Concept"),
+            cl.Action(name="clear_memory", value="clear_memory", label="🗑️ Clear History")
+        ]
+        await cl.Message(content="**Quick Actions:**", actions=actions).send()
+
+    except Exception as e:
+        error_msg = f"❌ **Failed to initialize CFA Agent:** {str(e)}\n\nPlease check your setup and try again."
+        await cl.Message(content=error_msg).send()
+
+
+async def display_market_overview():
+    """Display market overview as an action."""
+    try:
+        indices_data = get_market_indices()
+
+        if "error" not in indices_data:
+            market_content = "📊 **Market Overview:**\n\n"
+            for name, data in indices_data.items():
+                if "error" not in data:
+                    change_emoji = "🟢" if data["change"] >= 0 else "🔴"
+                    market_content += f"{change_emoji} **{name}:** {data['current_value']:,.2f} "
+                    market_content += f"({data['change']:+.2f}, {data['change_percent']:+.2f}%)\n"
+
+            await cl.Message(content=market_content).send()
+    except Exception as e:
+        await cl.Message(content=f"⚠️ Unable to load market data: {str(e)}").send()
+
+
+@cl.on_message
+async def main(message: cl.Message):
+    """Handle incoming messages."""
+    global agent
+
+    if not agent:
+        await cl.Message(content="❌ Agent not initialized. Please restart the chat.").send()
+        return
+
+    # Show thinking message
+    thinking_msg = cl.Message(content="🤔 Analyzing your query...")
+    await thinking_msg.send()
+
+    try:
+        # Get agent response
+        response = agent.query(message.content)
+
+        # Format the response
+        formatted_response = format_agent_response(response)
+
+        # Update the thinking message with the response
+        thinking_msg.content = formatted_response
+        await thinking_msg.update()
+
+    except Exception as e:
+        error_response = f"❌ **Error processing your query:** {str(e)}\n\nPlease try rephrasing your question or check if the requested data is available."
+        thinking_msg.content = error_response
+        await thinking_msg.update()
+
+
+def format_agent_response(response):
+    """Format the agent response for better display."""
+    if isinstance(response, dict) and "error" in response:
+        return f"❌ **Error:** {response['error']}"
+
+    # Convert to string if needed
+    response_str = str(response)
+
+    # Add some basic formatting for common patterns
+    response_str = response_str.replace("Thought:", "\n**🤔 Thought:**")
+    response_str = response_str.replace("Action:", "\n**⚡ Action:**")
+    response_str = response_str.replace("Action Input:", "\n**📝 Action Input:**")
+    response_str = response_str.replace("Observation:", "\n**👀 Observation:**")
+    response_str = response_str.replace("Final Answer:", "\n**✅ Final Answer:**")
+
+    return response_str
+
+
+@cl.action_callback("market_summary")
+async def market_summary():
+    """Handle market summary action."""
+    await main(cl.Message(content="Give me a summary of today's market performance"))
+
+
+@cl.action_callback("stock_analysis")
+async def stock_analysis():
+    """Handle stock analysis action."""
+    await cl.AskUserMessage(content="Please enter a stock ticker symbol for analysis:", timeout=30).send()
+
+
+@cl.action_callback("cfa_concept")
+async def cfa_concept():
+    """Handle CFA concept explanation action."""
+    await main(cl.Message(content="Explain a key CFA concept with examples"))
+
+
+@cl.action_callback("clear_memory")
+async def clear_memory():
+    """Handle clear memory action."""
+    global agent
+    if agent:
+        agent.clear_memory()
+        await cl.Message(content="🗑️ **Chat history cleared!** You can start a fresh conversation.").send()
+
+
+
+
+if __name__ == "__main__":
+    # Chainlit apps are run with `chainlit run app.py`
+    pass

requirements.txt

@@ -0,0 +1,28 @@
+# Core ML and LLM dependencies - optimized for HF Spaces
+torch>=2.0.0,<2.5.0
+transformers>=4.35.0,<5.0.0
+accelerate>=0.24.0
+tokenizers>=0.15.0
+bitsandbytes
+
+# LangChain ecosystem
+langchain>=0.1.0,<0.3.0
+langchain-community>=0.0.10
+langchain-core>=0.1.0
+
+# Financial data and analysis
+yfinance>=0.2.18
+pandas>=1.5.0,<3.0.0
+numpy>=1.24.0,<2.0.0
+
+# Web interface for HF Spaces
+chainlit>=1.0.0
+plotly>=5.15.0
+gradio>=4.0.0
+
+# Additional utilities
+requests>=2.31.0
+python-dotenv>=1.0.0
+
+# HF Spaces optimization
+huggingface_hub>=0.20.0

tools/data_fetcher.py

@@ -0,0 +1,339 @@
+"""
+CFA AI Agent - Real-time Financial Data Fetcher
+This module handles fetching real-time financial data using yfinance.
+"""
+
+import yfinance as yf
+import pandas as pd
+import numpy as np
+from typing import Dict, List, Optional, Union
+from datetime import datetime, timedelta
+from langchain.tools import tool
+
+
+@tool
+def get_stock_price(ticker: str) -> Dict[str, Union[float, str]]:
+    """
+    Get current stock price and basic information.
+
+    Args:
+        ticker: Stock ticker symbol
+
+    Returns:
+        Dictionary with current price and market data
+    """
+    try:
+        stock = yf.Ticker(ticker.upper())
+        info = stock.info
+
+        # Get latest price data
+        hist = stock.history(period="1d")
+        if hist.empty:
+            raise ValueError(f"No data available for ticker {ticker}")
+
+        current_price = hist['Close'].iloc[-1]
+        previous_close = info.get('previousClose', current_price)
+        change = current_price - previous_close
+        change_percent = (change / previous_close) * 100 if previous_close != 0 else 0
+
+        return {
+            "ticker": ticker.upper(),
+            "company_name": info.get('longName', 'Unknown'),
+            "current_price": round(current_price, 2),
+            "previous_close": round(previous_close, 2),
+            "change": round(change, 2),
+            "change_percent": round(change_percent, 2),
+            "volume": hist['Volume'].iloc[-1],
+            "market_cap": info.get('marketCap'),
+            "currency": info.get('currency', 'USD'),
+            "last_updated": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        }
+    except Exception as e:
+        return {"error": f"Failed to fetch stock price for {ticker}: {str(e)}"}
+
+
+@tool
+def get_historical_data(
+    ticker: str,
+    period: str = "1y",
+    interval: str = "1d"
+) -> Dict[str, Union[List, str]]:
+    """
+    Get historical stock data for analysis.
+
+    Args:
+        ticker: Stock ticker symbol
+        period: Time period (1d, 5d, 1mo, 3mo, 6mo, 1y, 2y, 5y, 10y, ytd, max)
+        interval: Data interval (1m, 2m, 5m, 15m, 30m, 60m, 90m, 1h, 1d, 5d, 1wk, 1mo, 3mo)
+
+    Returns:
+        Dictionary with historical price data and statistics
+    """
+    try:
+        stock = yf.Ticker(ticker.upper())
+        hist = stock.history(period=period, interval=interval)
+
+        if hist.empty:
+            raise ValueError(f"No historical data available for {ticker}")
+
+        # Calculate basic statistics
+        returns = hist['Close'].pct_change().dropna()
+
+        stats = {
+            "ticker": ticker.upper(),
+            "period": period,
+            "interval": interval,
+            "data_points": len(hist),
+            "start_date": hist.index[0].strftime("%Y-%m-%d"),
+            "end_date": hist.index[-1].strftime("%Y-%m-%d"),
+
+            # Price statistics
+            "highest_price": round(hist['High'].max(), 2),
+            "lowest_price": round(hist['Low'].min(), 2),
+            "avg_price": round(hist['Close'].mean(), 2),
+            "current_price": round(hist['Close'].iloc[-1], 2),
+
+            # Return statistics
+            "total_return": round(((hist['Close'].iloc[-1] / hist['Close'].iloc[0]) - 1) * 100, 2),
+            "volatility": round(returns.std() * np.sqrt(252) * 100, 2),  # Annualized volatility
+            "avg_daily_return": round(returns.mean() * 100, 4),
+            "max_daily_gain": round(returns.max() * 100, 2),
+            "max_daily_loss": round(returns.min() * 100, 2),
+
+            # Volume statistics
+            "avg_volume": int(hist['Volume'].mean()),
+            "max_volume": int(hist['Volume'].max()),
+            "min_volume": int(hist['Volume'].min()),
+
+            # Recent data (last 5 days)
+            "recent_prices": hist['Close'].tail(5).round(2).tolist(),
+            "recent_dates": [date.strftime("%Y-%m-%d") for date in hist.index[-5:]],
+            "recent_volumes": hist['Volume'].tail(5).tolist()
+        }
+
+        return stats
+    except Exception as e:
+        return {"error": f"Failed to fetch historical data for {ticker}: {str(e)}"}
+
+
+@tool
+def get_company_info(ticker: str) -> Dict[str, Union[str, float, int]]:
+    """
+    Get comprehensive company information and fundamentals.
+
+    Args:
+        ticker: Stock ticker symbol
+
+    Returns:
+        Dictionary with company information and key metrics
+    """
+    try:
+        stock = yf.Ticker(ticker.upper())
+        info = stock.info
+
+        company_data = {
+            "ticker": ticker.upper(),
+            "company_name": info.get('longName', 'Unknown'),
+            "sector": info.get('sector', 'Unknown'),
+            "industry": info.get('industry', 'Unknown'),
+            "country": info.get('country', 'Unknown'),
+            "website": info.get('website', 'N/A'),
+            "business_summary": info.get('longBusinessSummary', 'N/A'),
+
+            # Key executives
+            "ceo": info.get('companyOfficers', [{}])[0].get('name', 'N/A') if info.get('companyOfficers') else 'N/A',
+
+            # Financial metrics
+            "market_cap": info.get('marketCap'),
+            "enterprise_value": info.get('enterpriseValue'),
+            "shares_outstanding": info.get('sharesOutstanding'),
+            "float_shares": info.get('floatShares'),
+
+            # Employee info
+            "full_time_employees": info.get('fullTimeEmployees'),
+
+            # Exchange info
+            "exchange": info.get('exchange', 'Unknown'),
+            "quote_type": info.get('quoteType', 'Unknown'),
+            "currency": info.get('currency', 'USD'),
+
+            # ESG scores (if available)
+            "esg_scores": info.get('esgScores'),
+            "sustainability_score": info.get('sustainabilityScore'),
+
+            # Analyst recommendations
+            "recommendation": info.get('recommendationKey', 'N/A'),
+            "target_high_price": info.get('targetHighPrice'),
+            "target_low_price": info.get('targetLowPrice'),
+            "target_mean_price": info.get('targetMeanPrice'),
+            "number_of_analyst_opinions": info.get('numberOfAnalystOpinions'),
+
+            # Risk metrics
+            "audit_risk": info.get('auditRisk'),
+            "board_risk": info.get('boardRisk'),
+            "compensation_risk": info.get('compensationRisk'),
+            "shareholder_rights_risk": info.get('shareHolderRightsRisk'),
+            "overall_risk": info.get('overallRisk')
+        }
+
+        # Remove None values
+        company_data = {k: v for k, v in company_data.items() if v is not None}
+
+        return company_data
+    except Exception as e:
+        return {"error": f"Failed to fetch company info for {ticker}: {str(e)}"}
+
+
+@tool
+def get_financial_statements(ticker: str) -> Dict[str, Union[pd.DataFrame, str]]:
+    """
+    Get financial statements (income statement, balance sheet, cash flow).
+
+    Args:
+        ticker: Stock ticker symbol
+
+    Returns:
+        Dictionary with financial statement data
+    """
+    try:
+        stock = yf.Ticker(ticker.upper())
+
+        # Fetch financial statements
+        income_stmt = stock.financials
+        balance_sheet = stock.balance_sheet
+        cash_flow = stock.cashflow
+
+        result = {
+            "ticker": ticker.upper(),
+            "has_income_statement": not income_stmt.empty,
+            "has_balance_sheet": not balance_sheet.empty,
+            "has_cash_flow": not cash_flow.empty,
+        }
+
+        # Convert to dictionaries for easier handling
+        if not income_stmt.empty:
+            result["income_statement_years"] = [str(col.year) for col in income_stmt.columns]
+            result["total_revenue"] = income_stmt.loc['Total Revenue'].to_dict() if 'Total Revenue' in income_stmt.index else {}
+            result["net_income"] = income_stmt.loc['Net Income'].to_dict() if 'Net Income' in income_stmt.index else {}
+
+        if not balance_sheet.empty:
+            result["balance_sheet_years"] = [str(col.year) for col in balance_sheet.columns]
+            result["total_assets"] = balance_sheet.loc['Total Assets'].to_dict() if 'Total Assets' in balance_sheet.index else {}
+            result["total_debt"] = balance_sheet.loc['Total Debt'].to_dict() if 'Total Debt' in balance_sheet.index else {}
+
+        if not cash_flow.empty:
+            result["cash_flow_years"] = [str(col.year) for col in cash_flow.columns]
+            result["operating_cash_flow"] = cash_flow.loc['Operating Cash Flow'].to_dict() if 'Operating Cash Flow' in cash_flow.index else {}
+            result["free_cash_flow"] = cash_flow.loc['Free Cash Flow'].to_dict() if 'Free Cash Flow' in cash_flow.index else {}
+
+        return result
+    except Exception as e:
+        return {"error": f"Failed to fetch financial statements for {ticker}: {str(e)}"}
+
+
+@tool
+def get_market_indices() -> Dict[str, Dict[str, Union[float, str]]]:
+    """
+    Get current prices and performance of major market indices.
+
+    Returns:
+        Dictionary with major market index data
+    """
+    try:
+        indices = {
+            "S&P 500": "^GSPC",
+            "Dow Jones": "^DJI",
+            "NASDAQ": "^IXIC",
+            "Russell 2000": "^RUT",
+            "VIX": "^VIX",
+            "10-Year Treasury": "^TNX"
+        }
+
+        results = {}
+        for name, ticker in indices.items():
+            try:
+                index = yf.Ticker(ticker)
+                hist = index.history(period="2d")
+                if not hist.empty:
+                    current_price = hist['Close'].iloc[-1]
+                    previous_close = hist['Close'].iloc[-2] if len(hist) > 1 else current_price
+                    change = current_price - previous_close
+                    change_percent = (change / previous_close) * 100 if previous_close != 0 else 0
+
+                    results[name] = {
+                        "ticker": ticker,
+                        "current_value": round(current_price, 2),
+                        "change": round(change, 2),
+                        "change_percent": round(change_percent, 2),
+                        "last_updated": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+                    }
+            except Exception:
+                results[name] = {"error": f"Failed to fetch data for {name}"}
+
+        return results
+    except Exception as e:
+        return {"error": f"Failed to fetch market indices: {str(e)}"}
+
+
+@tool
+def compare_stocks(tickers: List[str], metric: str = "performance") -> Dict[str, Union[List, str]]:
+    """
+    Compare multiple stocks on various metrics.
+
+    Args:
+        tickers: List of stock ticker symbols
+        metric: Comparison metric ('performance', 'valuation', 'volatility')
+
+    Returns:
+        Dictionary with comparison results
+    """
+    try:
+        if len(tickers) < 2:
+            raise ValueError("Need at least 2 tickers for comparison")
+
+        results = {
+            "tickers": [t.upper() for t in tickers],
+            "metric": metric,
+            "comparison_data": {}
+        }
+
+        for ticker in tickers:
+            try:
+                stock = yf.Ticker(ticker.upper())
+                info = stock.info
+                hist = stock.history(period="1y")
+
+                if metric == "performance":
+                    if not hist.empty:
+                        ytd_return = ((hist['Close'].iloc[-1] / hist['Close'].iloc[0]) - 1) * 100
+                        results["comparison_data"][ticker.upper()] = {
+                            "ytd_return": round(ytd_return, 2),
+                            "current_price": round(hist['Close'].iloc[-1], 2),
+                            "52_week_high": info.get('fiftyTwoWeekHigh'),
+                            "52_week_low": info.get('fiftyTwoWeekLow')
+                        }
+
+                elif metric == "valuation":
+                    results["comparison_data"][ticker.upper()] = {
+                        "pe_ratio": info.get('trailingPE'),
+                        "price_to_book": info.get('priceToBook'),
+                        "price_to_sales": info.get('priceToSalesTrailing12Months'),
+                        "market_cap": info.get('marketCap')
+                    }
+
+                elif metric == "volatility":
+                    if not hist.empty:
+                        returns = hist['Close'].pct_change().dropna()
+                        volatility = returns.std() * np.sqrt(252) * 100
+                        results["comparison_data"][ticker.upper()] = {
+                            "volatility": round(volatility, 2),
+                            "beta": info.get('beta'),
+                            "max_drawdown": round((hist['Close'].min() / hist['Close'].max() - 1) * 100, 2)
+                        }
+
+            except Exception as e:
+                results["comparison_data"][ticker.upper()] = {"error": str(e)}
+
+        return results
+    except Exception as e:
+        return {"error": f"Stock comparison failed: {str(e)}"}

tools/finance_tools.py

@@ -0,0 +1,323 @@
+"""
+CFA AI Agent - Finance Calculation Tools
+This module contains various financial calculation functions for CFA analysis.
+"""
+
+import numpy as np
+import pandas as pd
+from typing import List, Dict, Union, Optional
+import yfinance as yf
+from langchain.tools import tool
+
+
+@tool
+def calculate_dcf(
+    cash_flows: List[float],
+    terminal_value: float,
+    discount_rate: float
+) -> Dict[str, float]:
+    """
+    Calculate Discounted Cash Flow (DCF) valuation.
+
+    Args:
+        cash_flows: List of projected free cash flows
+        terminal_value: Terminal value at end of projection period
+        discount_rate: Weighted average cost of capital (WACC) as decimal
+
+    Returns:
+        Dictionary with NPV, terminal value present value, and total enterprise value
+    """
+    try:
+        if not cash_flows:
+            raise ValueError("Cash flows list cannot be empty")
+        if discount_rate <= 0:
+            raise ValueError("Discount rate must be positive")
+
+        # Calculate present value of cash flows
+        pv_cash_flows = []
+        for i, cf in enumerate(cash_flows, 1):
+            pv = cf / ((1 + discount_rate) ** i)
+            pv_cash_flows.append(pv)
+
+        # Calculate present value of terminal value
+        years = len(cash_flows)
+        pv_terminal = terminal_value / ((1 + discount_rate) ** years)
+
+        # Total enterprise value
+        enterprise_value = sum(pv_cash_flows) + pv_terminal
+
+        return {
+            "pv_cash_flows": sum(pv_cash_flows),
+            "pv_terminal_value": pv_terminal,
+            "enterprise_value": enterprise_value,
+            "cash_flow_details": pv_cash_flows
+        }
+    except Exception as e:
+        return {"error": f"DCF calculation failed: {str(e)}"}
+
+
+@tool
+def calculate_sharpe_ratio(
+    returns: List[float],
+    risk_free_rate: float
+) -> Dict[str, float]:
+    """
+    Calculate Sharpe Ratio for risk-adjusted returns.
+
+    Args:
+        returns: List of periodic returns (as decimals)
+        risk_free_rate: Risk-free rate (as decimal)
+
+    Returns:
+        Dictionary with Sharpe ratio, average return, and standard deviation
+    """
+    try:
+        if not returns:
+            raise ValueError("Returns list cannot be empty")
+        if len(returns) < 2:
+            raise ValueError("Need at least 2 return observations")
+
+        returns_array = np.array(returns)
+
+        # Calculate metrics
+        avg_return = np.mean(returns_array)
+        std_dev = np.std(returns_array, ddof=1)  # Sample standard deviation
+        excess_return = avg_return - risk_free_rate
+
+        if std_dev == 0:
+            raise ValueError("Standard deviation cannot be zero")
+
+        sharpe_ratio = excess_return / std_dev
+
+        return {
+            "sharpe_ratio": sharpe_ratio,
+            "average_return": avg_return,
+            "standard_deviation": std_dev,
+            "excess_return": excess_return,
+            "risk_free_rate": risk_free_rate
+        }
+    except Exception as e:
+        return {"error": f"Sharpe ratio calculation failed: {str(e)}"}
+
+
+@tool
+def compare_pe_ratios(ticker1: str, ticker2: str) -> Dict[str, Union[float, str]]:
+    """
+    Compare P/E ratios of two stocks using real-time data.
+
+    Args:
+        ticker1: First stock ticker symbol
+        ticker2: Second stock ticker symbol
+
+    Returns:
+        Dictionary with P/E ratios and comparison analysis
+    """
+    try:
+        # Fetch stock data
+        stock1 = yf.Ticker(ticker1.upper())
+        stock2 = yf.Ticker(ticker2.upper())
+
+        # Get info
+        info1 = stock1.info
+        info2 = stock2.info
+
+        # Extract P/E ratios
+        pe1 = info1.get('trailingPE') or info1.get('forwardPE')
+        pe2 = info2.get('trailingPE') or info2.get('forwardPE')
+
+        if pe1 is None or pe2 is None:
+            return {"error": f"Could not retrieve P/E ratios for {ticker1} or {ticker2}"}
+
+        # Calculate comparison metrics
+        pe_difference = pe1 - pe2
+        pe_ratio = pe1 / pe2 if pe2 != 0 else None
+
+        # Determine which is more expensive
+        comparison = "equal"
+        if pe1 > pe2:
+            comparison = f"{ticker1} is more expensive"
+        elif pe1 < pe2:
+            comparison = f"{ticker2} is more expensive"
+
+        return {
+            f"{ticker1}_pe": pe1,
+            f"{ticker2}_pe": pe2,
+            "pe_difference": pe_difference,
+            "pe_ratio": pe_ratio,
+            "comparison": comparison,
+            f"{ticker1}_name": info1.get('longName', ticker1),
+            f"{ticker2}_name": info2.get('longName', ticker2)
+        }
+    except Exception as e:
+        return {"error": f"P/E comparison failed: {str(e)}"}
+
+
+@tool
+def calculate_beta(ticker: str, market_ticker: str = "^GSPC", period: str = "2y") -> Dict[str, float]:
+    """
+    Calculate beta coefficient for a stock relative to market.
+
+    Args:
+        ticker: Stock ticker symbol
+        market_ticker: Market index ticker (default S&P 500)
+        period: Time period for calculation
+
+    Returns:
+        Dictionary with beta, correlation, and other metrics
+    """
+    try:
+        # Fetch data
+        stock = yf.Ticker(ticker.upper())
+        market = yf.Ticker(market_ticker)
+
+        # Get historical data
+        stock_data = stock.history(period=period)
+        market_data = market.history(period=period)
+
+        if stock_data.empty or market_data.empty:
+            raise ValueError("Could not fetch historical data")
+
+        # Calculate returns
+        stock_returns = stock_data['Close'].pct_change().dropna()
+        market_returns = market_data['Close'].pct_change().dropna()
+
+        # Align data
+        aligned_data = pd.concat([stock_returns, market_returns], axis=1, join='inner')
+        aligned_data.columns = ['stock', 'market']
+        aligned_data = aligned_data.dropna()
+
+        if len(aligned_data) < 20:
+            raise ValueError("Insufficient data points for beta calculation")
+
+        # Calculate beta
+        covariance = np.cov(aligned_data['stock'], aligned_data['market'])[0][1]
+        market_variance = np.var(aligned_data['market'], ddof=1)
+        beta = covariance / market_variance
+
+        # Calculate correlation
+        correlation = np.corrcoef(aligned_data['stock'], aligned_data['market'])[0][1]
+
+        return {
+            "beta": beta,
+            "correlation": correlation,
+            "stock_volatility": np.std(aligned_data['stock'], ddof=1),
+            "market_volatility": np.std(aligned_data['market'], ddof=1),
+            "data_points": len(aligned_data),
+            "period": period
+        }
+    except Exception as e:
+        return {"error": f"Beta calculation failed: {str(e)}"}
+
+
+@tool
+def calculate_wacc(
+    cost_of_equity: float,
+    cost_of_debt: float,
+    tax_rate: float,
+    market_value_equity: float,
+    market_value_debt: float
+) -> Dict[str, float]:
+    """
+    Calculate Weighted Average Cost of Capital (WACC).
+
+    Args:
+        cost_of_equity: Cost of equity as decimal
+        cost_of_debt: Cost of debt as decimal
+        tax_rate: Corporate tax rate as decimal
+        market_value_equity: Market value of equity
+        market_value_debt: Market value of debt
+
+    Returns:
+        Dictionary with WACC and component calculations
+    """
+    try:
+        total_value = market_value_equity + market_value_debt
+
+        if total_value <= 0:
+            raise ValueError("Total market value must be positive")
+
+        # Calculate weights
+        weight_equity = market_value_equity / total_value
+        weight_debt = market_value_debt / total_value
+
+        # Calculate after-tax cost of debt
+        after_tax_cost_debt = cost_of_debt * (1 - tax_rate)
+
+        # Calculate WACC
+        wacc = (weight_equity * cost_of_equity) + (weight_debt * after_tax_cost_debt)
+
+        return {
+            "wacc": wacc,
+            "weight_equity": weight_equity,
+            "weight_debt": weight_debt,
+            "after_tax_cost_debt": after_tax_cost_debt,
+            "cost_of_equity": cost_of_equity,
+            "cost_of_debt": cost_of_debt,
+            "tax_rate": tax_rate
+        }
+    except Exception as e:
+        return {"error": f"WACC calculation failed: {str(e)}"}
+
+
+@tool
+def financial_ratios_analysis(ticker: str) -> Dict[str, Union[float, str]]:
+    """
+    Perform comprehensive financial ratios analysis for a stock.
+
+    Args:
+        ticker: Stock ticker symbol
+
+    Returns:
+        Dictionary with various financial ratios and metrics
+    """
+    try:
+        stock = yf.Ticker(ticker.upper())
+        info = stock.info
+
+        # Extract key metrics
+        ratios = {
+            "ticker": ticker.upper(),
+            "company_name": info.get('longName', 'N/A'),
+
+            # Valuation ratios
+            "pe_ratio": info.get('trailingPE'),
+            "forward_pe": info.get('forwardPE'),
+            "price_to_book": info.get('priceToBook'),
+            "price_to_sales": info.get('priceToSalesTrailing12Months'),
+            "peg_ratio": info.get('pegRatio'),
+
+            # Profitability ratios
+            "profit_margin": info.get('profitMargins'),
+            "operating_margin": info.get('operatingMargins'),
+            "roe": info.get('returnOnEquity'),
+            "roa": info.get('returnOnAssets'),
+
+            # Financial health
+            "current_ratio": info.get('currentRatio'),
+            "quick_ratio": info.get('quickRatio'),
+            "debt_to_equity": info.get('debtToEquity'),
+            "total_debt": info.get('totalDebt'),
+            "total_cash": info.get('totalCash'),
+
+            # Market data
+            "market_cap": info.get('marketCap'),
+            "enterprise_value": info.get('enterpriseValue'),
+            "beta": info.get('beta'),
+            "52_week_high": info.get('fiftyTwoWeekHigh'),
+            "52_week_low": info.get('fiftyTwoWeekLow'),
+
+            # Dividend info
+            "dividend_yield": info.get('dividendYield'),
+            "payout_ratio": info.get('payoutRatio'),
+
+            # Growth metrics
+            "earnings_growth": info.get('earningsGrowth'),
+            "revenue_growth": info.get('revenueGrowth')
+        }
+
+        # Remove None values
+        ratios = {k: v for k, v in ratios.items() if v is not None}
+
+        return ratios
+    except Exception as e:
+        return {"error": f"Financial ratios analysis failed: {str(e)}"}