import yfinance as yf import pandas as pd import plotly.graph_objects as go import plotly.io as pio from datetime import datetime, timedelta import json import os import modal import logging import requests from typing import Dict, Any # DEBUG: Print modal module path and version print(f"DEBUG: modal module loaded from: {modal.__file__}") print(f"DEBUG: modal version: {modal.__version__}") from .data_collector import StockDataCollector, YFinanceRateLimitError from .moat_analyzer import MoatAnalyzer from .config import AppConfig class StockAnalyzer: """Handles stock analysis and visualization.""" def __init__(self): self.config = AppConfig() self.data_collector = StockDataCollector() self.moat_analyzer = MoatAnalyzer() self._setup_logging() # Helper function to format numbers or return 'N/A' def _format_num(self, value): """Helper to format numbers, handling N/A gracefully.""" if isinstance(value, (int, float)): return f"{value:,.2f}" return "N/A" def _setup_logging(self): """Setup logging configuration.""" logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) self.logger = logging.getLogger(__name__) def analyze(self, ticker): """Analyze a stock and return the analysis results.""" try: stock = yf.Ticker(ticker) info = stock.info if not info or not info.get('regularMarketPrice'): # Check for essential info return {"analysis": f"Could not retrieve comprehensive stock information for {ticker}. Please check the ticker symbol and try again.", "revenue_chart": None, "fcf_chart": None, "shares_chart": None} # Initialize variables to None or default empty values latest_revenue = 'N/A' latest_fcf = 'N/A' latest_shares = 'N/A' latest_bs_data = {} latest_is_data = {} moat_summary_text = 'N/A' moat_rating = 'N/A' llm_insights = "AI insights temporarily unavailable." serper_results_str = "" revenue_chart_json = None fcf_chart_json = None shares_chart_json = None # Collect data for charts and LLM revenue_data = self.data_collector.get_revenue_history(ticker) fcf_data = self.data_collector.get_fcf_history(ticker) shares_data = self.data_collector.get_shares_history(ticker) # Generate charts revenue_chart = self._create_revenue_chart(revenue_data) fcf_chart = self._create_fcf_chart(fcf_data) shares_chart = self._create_shares_chart(shares_data) # Convert Plotly figures to JSON strings revenue_chart_json = pio.to_json(revenue_chart) if revenue_chart else None fcf_chart_json = pio.to_json(fcf_chart) if fcf_chart else None shares_chart_json = pio.to_json(shares_chart) if shares_chart else None # Safely get latest values with proper error handling if revenue_data is not None and not revenue_data.empty: latest_revenue = revenue_data.iloc[-1]['Revenue'] if fcf_data is not None and not fcf_data.empty: latest_fcf = fcf_data.iloc[-1]['FCF'] if shares_data is not None and not shares_data.empty: latest_shares = shares_data.iloc[-1]['Shares'] # Attempt to get full balance sheet and income statement data balance_sheet = stock.balance_sheet income_statement = stock.financials # Extract relevant recent quarterly/annual data (handle potential empty dataframes) if balance_sheet is not None and not balance_sheet.empty: latest_bs_data = balance_sheet.iloc[:, 0].to_dict() if income_statement is not None and not income_statement.empty: latest_is_data = income_statement.iloc[:, 0].to_dict() # Get moat analysis moat_analyzer = MoatAnalyzer() moat_analysis_results = moat_analyzer.analyze_moat(ticker) moat_summary_text = moat_analysis_results["summary"] moat_rating = moat_analysis_results["rating"] try: # Look up the deployed Modal LLM app and get the remote function here generate_llm_response_modal = modal.Function.from_name("buffetbot-llm", "generate_llm_response") # Perform Serper search for moat articles data_collector = StockDataCollector() moat_search_results = data_collector._perform_serper_search(f"{ticker} economic moat analysis articles") # Format Serper search results serper_results_str = "" if moat_search_results: serper_results_str = "\n\n#### External Moat Analysis (Web Search)\n" + "\n".join([ f"- **{r['title']}** (Source: {r['link']})\n Snippet: {r['snippet']}" for r in moat_search_results ]) # Get additional comprehensive data for LLM prompt revenue_data = data_collector.get_revenue_history(ticker) fcf_data = data_collector.get_fcf_history(ticker) shares_data = data_collector.get_shares_history(ticker) # Generate charts revenue_chart = self._create_revenue_chart(revenue_data) fcf_chart = self._create_fcf_chart(fcf_data) shares_chart = self._create_shares_chart(shares_data) # Convert Plotly figures to JSON strings revenue_chart_json = pio.to_json(revenue_chart) if revenue_chart else None fcf_chart_json = pio.to_json(fcf_chart) if fcf_chart else None shares_chart_json = pio.to_json(shares_chart) if shares_chart else None # Safely get latest values with proper error handling if revenue_data is not None and not revenue_data.empty: latest_revenue = revenue_data.iloc[-1]['Revenue'] if fcf_data is not None and not fcf_data.empty: latest_fcf = fcf_data.iloc[-1]['FCF'] if shares_data is not None and not shares_data.empty: latest_shares = shares_data.iloc[-1]['Shares'] # Attempt to get full balance sheet and income statement data balance_sheet = stock.balance_sheet income_statement = stock.financials # Extract relevant recent quarterly/annual data (handle potential empty dataframes) if balance_sheet is not None and not balance_sheet.empty: latest_bs_data = balance_sheet.iloc[:, 0].to_dict() if income_statement is not None and not income_statement.empty: latest_is_data = income_statement.iloc[:, 0].to_dict() # Format detailed financial data for the LLM prompt financial_data_str = f""" #### Recent Financials (Latest Available) - Latest Revenue: {self._format_num(latest_revenue)} - Latest Free Cash Flow: {self._format_num(latest_fcf)} - Latest Shares Outstanding: {self._format_num(latest_shares)} #### Key Balance Sheet Items - Total Assets: {self._format_num(latest_bs_data.get('Total Assets', 'N/A'))} - Total Liabilities: {self._format_num(latest_bs_data.get('Total Liabilities', 'N/A'))} - Total Debt: {self._format_num(info.get('totalDebt', 'N/A'))} - Total Cash: {self._format_num(info.get('totalCash', 'N/A'))} - Total Equity: {self._format_num(latest_bs_data.get('Total Equity', 'N/A'))} - Debt to Equity: {self._format_num(info.get('debtToEquity', 'N/A'))} #### Key Income Statement Items - Gross Profits: {self._format_num(latest_is_data.get('Gross Profit', 'N/A'))} - Operating Income: {self._format_num(latest_is_data.get('Operating Income', 'N/A'))} - Net Income: {self._format_num(latest_is_data.get('Net Income', 'N/A'))} - Earnings Growth (YoY): {self._format_num(info.get('earningsQuarterlyGrowth', 'N/A'))}% """ # Ensure the LLM knows to use BuffetBot's persona and clearly mark AI insights system_message = """ As an AI-powered stock analyst named BuffetBot, your goal is to provide concise, clear, and actionable investment insights. When giving recommendations, state "BuffetBot recommends" instead of "I recommend". Clearly mark your analysis as "AI Insights:" at the beginning of your comprehensive insight. Focus on explaining key numbers, financial health, growth metrics, economic moat significance, and overall investment potential. Do not include "What to look for" sections for charts or specific metrics, as these will be provided separately. Avoid garbled text or incomplete sentences. """ llm_prompt = self._generate_llm_prompt(ticker, { 'info': info, 'moat_summary': moat_summary_text, 'serper_results': serper_results_str, 'financial_data_str': financial_data_str }) llm_insights = generate_llm_response_modal.remote( system_message=system_message, prompt=llm_prompt ) except Exception as llm_e: self.logger.warning(f"Could not get LLM insights for {ticker}: {llm_e}") llm_insights = "AI insights temporarily unavailable. Please ensure Modal LLM service is deployed and accessible and Modal API keys are set as Hugging Face Space secrets." # Generate analysis text analysis = self._generate_analysis(ticker, info, moat_summary_text, moat_rating, llm_insights, latest_bs_data, latest_is_data, latest_revenue, latest_fcf, latest_shares) return { "analysis": analysis, "revenue_chart": revenue_chart_json, "fcf_chart": fcf_chart_json, "shares_chart": shares_chart_json } except YFinanceRateLimitError as e: self.logger.error(f"Yfinance rate limit hit for {ticker}: {e}") return {"analysis": f"Error analyzing stock: {e}. Charts may be unavailable.", "revenue_chart": None, "fcf_chart": None, "shares_chart": None} except requests.exceptions.HTTPError as e: if e.response.status_code == 429: # This case should now be handled by YFinanceRateLimitError self.logger.error(f"Unexpected 429 HTTP Error for {ticker}: {e}") return {"analysis": f"Error analyzing stock: You have been rate-limited by Yahoo Finance. Please wait a few minutes and try again. Charts may be unavailable.", "revenue_chart": None, "fcf_chart": None, "shares_chart": None} else: self.logger.error(f"HTTP error fetching data for {ticker}: {e}") return {"analysis": f"Error analyzing stock: An HTTP error occurred: {e}. Charts may be unavailable.", "revenue_chart": None, "fcf_chart": None, "shares_chart": None} except Exception as e: self.logger.error(f"Error analyzing stock {ticker}: {e}", exc_info=True) # Log full traceback return {"analysis": f"An unexpected error occurred while analyzing {ticker}: {str(e)}. Charts may be unavailable.", "revenue_chart": None, "fcf_chart": None, "shares_chart": None} def _create_revenue_chart(self, revenue_data): """Create revenue growth chart.""" if revenue_data is None or revenue_data.empty: return None fig = go.Figure() fig.add_trace(go.Scatter( x=revenue_data.index, y=revenue_data['Revenue'], mode='lines+markers', name='Revenue' )) fig.update_layout( title='Revenue Growth', xaxis_title='Year', yaxis_title='Revenue (USD)', template='plotly_white' ) return fig def _create_fcf_chart(self, fcf_data): """Create free cash flow chart.""" if fcf_data is None or fcf_data.empty: return None fig = go.Figure() fig.add_trace(go.Scatter( x=fcf_data.index, y=fcf_data['FCF'], mode='lines+markers', name='Free Cash Flow' )) fig.update_layout( title='Free Cash Flow', xaxis_title='Year', yaxis_title='Free Cash Flow (USD)', template='plotly_white' ) return fig def _create_shares_chart(self, shares_data): """Create shares outstanding chart.""" if shares_data is None or shares_data.empty: return None fig = go.Figure() fig.add_trace(go.Scatter( x=shares_data.index, y=shares_data['Shares'], mode='lines+markers', name='Shares Outstanding' )) fig.update_layout( title='Shares Outstanding', xaxis_title='Year', yaxis_title='Shares (millions)', template='plotly_white' ) return fig def _generate_analysis(self, ticker, info, moat_summary, moat_rating, llm_insights, latest_bs_data, latest_is_data, latest_revenue, latest_fcf, latest_shares): """Generate analysis text.""" data_as_of_timestamp = info.get('regularMarketTime') data_as_of_date = datetime.fromtimestamp(data_as_of_timestamp).strftime('%Y-%m-%d %H:%M:%S') if data_as_of_timestamp else 'N/A' # Initialize serper_results_str serper_results_str = "" analysis = f""" # {info.get('longName', ticker)} ({ticker}) Analysis *Data as of: {data_as_of_date}* ## Company Overview - Sector: {info.get('sector', 'N/A')} - Industry: {info.get('industry', 'N/A')} - Full-time Employees: {self._format_num(info.get('fullTimeEmployees', 'N/A'))} ## Current Status - Current Price: {self._format_num(info.get('currentPrice', 'N/A'))} - 52 Week High: {self._format_num(info.get('fiftyTwoWeekHigh', 'N/A'))} - 52 Week Low: {self._format_num(info.get('fiftyTwoWeekLow', 'N/A'))} - Market Cap: {self._format_num(info.get('marketCap', 'N/A'))} ## Key Valuation Metrics - P/E Ratio: {self._format_num(info.get('trailingPE', 'N/A'))} - Forward P/E: {self._format_num(info.get('forwardPE', 'N/A'))} - PEG Ratio: {self._format_num(info.get('pegRatio', 'N/A'))} - Dividend Yield: {self._format_num(info.get('dividendYield', 'N/A'))}% ## Growth and Profitability - Revenue Growth (YoY): {self._format_num(info.get('revenueGrowth', 'N/A'))}% - Profit Margins: {self._format_num(info.get('profitMargins', 'N/A'))}% - Operating Margins: {self._format_num(info.get('operatingMargins', 'N/A'))}% ## Financial Health Indicators - Current Ratio: {self._format_num(info.get('currentRatio', 'N/A'))} - Debt to Equity: {self._format_num(info.get('debtToEquity', 'N/A'))} - Return on Equity: {self._format_num(info.get('returnOnEquity', 'N/A'))}% - Debt to Asset Ratio: {self._format_num(info.get('totalDebt', 0) / info.get('totalAssets', 1) if info.get('totalAssets') else 'N/A')} #### Recent Financials (Latest Available) - Latest Revenue: {self._format_num(latest_revenue)} - Latest Free Cash Flow: {self._format_num(latest_fcf)} - Latest Shares Outstanding: {self._format_num(latest_shares)} #### Key Balance Sheet Items - Total Assets: {self._format_num(latest_bs_data.get('Total Assets', 'N/A'))} - Total Liabilities: {self._format_num(latest_bs_data.get('Total Liabilities', 'N/A'))} - Total Debt: {self._format_num(info.get('totalDebt', 'N/A'))} - Total Cash: {self._format_num(info.get('totalCash', 'N/A'))} - Total Equity: {self._format_num(latest_bs_data.get('Total Equity', 'N/A'))} - Debt to Equity: {self._format_num(info.get('debtToEquity', 'N/A'))} #### Key Income Statement Items - Gross Profits: {self._format_num(latest_is_data.get('Gross Profit', 'N/A'))} - Operating Income: {self._format_num(latest_is_data.get('Operating Income', 'N/A'))} - Net Income: {self._format_num(latest_is_data.get('Net Income', 'N/A'))} - Earnings Growth (YoY): {self._format_num(info.get('earningsQuarterlyGrowth', 'N/A'))}% ## Economic Moat Analysis {moat_summary} {serper_results_str} ## AI Insights {llm_insights} """ return analysis def _generate_llm_prompt(self, ticker: str, data: Dict[str, Any]) -> str: """Generate a focused prompt for the LLM.""" info = data.get('info', {}) moat_summary_text = data.get('moat_summary', 'N/A') serper_results_str = data.get('serper_results', 'N/A') financial_data_str = data.get('financial_data_str', 'N/A') llm_prompt = f""" Analyze {ticker} ({info.get('longName', ticker)}) as an investment opportunity. Key Metrics: - Price: ${self._format_num(info.get('currentPrice', 'N/A'))} - P/E: {self._format_num(info.get('trailingPE', 'N/A'))} - Revenue Growth: {self._format_num(info.get('revenueGrowth', 'N/A'))}% - Profit Margin: {self._format_num(info.get('profitMargins', 'N/A'))}% - Debt/Equity: {self._format_num(info.get('debtToEquity', 'N/A'))} - ROE: {self._format_num(info.get('returnOnEquity', 'N/A'))}% Moat Analysis: {moat_summary_text} {serper_results_str} Provide a concise analysis focusing on: 1. Key strengths and risks 2. Economic moat significance 3. Clear investment recommendation Keep your response under 200 words. """ return llm_prompt