📈 Stock Analysis Agent

import google.generativeai as genai import yfinance as yf from tavily import TavilyClient from typing import Dict, List, Optional import json class StockAnalysisPipeline: """ Multi-agent orchestration system for comprehensive stock analysis. Coordinates 5 specialized AI agents for screening, fundamental analysis, business moat research, dividend evaluation, and report generation. """ def __init__(self, gemini_api_key: str, tavily_api_key: str): genai.configure(api_key=gemini_api_key) self.model = genai.GenerativeModel('gemini-2.0-flash-exp') self.tavily = TavilyClient(api_key=tavily_api_key) # Investment yardstick scoring criteria (7 categories, 100 points total) self.scoring_categories = { 'fundamentals_profitability': 20, # ROE, EPS, margins 'financial_health': 15, # Debt, current ratio 'valuation': 20, # PE, P/B, P/S ratios 'business_moat': 15, # Competitive advantages 'dividends_cashflow': 15, # Dividend sustainability 'management_outlook': 10, # Track record, governance 'liquidity': 5 # Trading volume } async run_stock_analysis_pipeline(self, ticker: str) -> Dict: """ Execute complete 5-agent analysis pipeline for a stock. Args: ticker: Stock symbol (e.g., '1818.KL' for Bursa Malaysia) Returns: Complete investment analysis with scoring and recommendation """ print(f"🔍 Starting analysis for {ticker}...") # Step 1: Fetch real-time financial data stock_data = self._fetch_stock_data(ticker) if not stock_data['valid']: return {'error': 'Failed to fetch stock data'} # Step 2: Agent 1 - Stock Screener (validate investment criteria) screening_result = await self._agent_stock_screener(stock_data) if not screening_result['passed']: return { 'ticker': ticker, 'screening': screening_result, 'recommendation': 'FAIL - Does not meet investment criteria' } # Step 3: Agent 2 - Fundamental Analyst (score 55 points) fundamental_analysis = await self._agent_fundamental_analyst(stock_data) # Step 4: Agent 3 - Business Moat Analyst (score 15 points via web research) moat_analysis = await self._agent_business_moat(ticker, stock_data) # Step 5: Agent 4 - Dividend & Cash Flow Analyst (score 15 points) dividend_analysis = await self._agent_dividend_cashflow(stock_data) # Step 6: Calculate total yardstick score (0-100) total_score = self._calculate_total_score( fundamental_analysis, moat_analysis, dividend_analysis ) # Step 7: Agent 5 - Investment Report Generator report = await self._agent_report_generator( ticker, stock_data, screening_result, fundamental_analysis, moat_analysis, dividend_analysis, total_score ) return { 'ticker': ticker, 'total_score': total_score, 'recommendation': self._get_recommendation(total_score), 'screening': screening_result, 'fundamental_analysis': fundamental_analysis, 'moat_analysis': moat_analysis, 'dividend_analysis': dividend_analysis, 'report_path': report['markdown_path'], 'chart_path': report['chart_path'] } def _fetch_stock_data(self, ticker: str) -> Dict: """ Fetch comprehensive stock data from Yahoo Finance. Includes price, fundamentals, dividends, and financial statements. """ try: stock = yf.Ticker(ticker) info = stock.info return { 'valid': True, 'ticker': ticker, 'name': info.get('longName', ticker), 'current_price': info.get('currentPrice'), 'market_cap': info.get('marketCap'), 'pe_ratio': info.get('trailingPE'), 'forward_pe': info.get('forwardPE'), 'roe': info.get('returnOnEquity', 0) * 100, 'eps': info.get('trailingEps'), 'dividend_yield': info.get('dividendYield', 0) * 100, 'payout_ratio': info.get('payoutRatio', 0) * 100, 'debt_to_equity': info.get('debtToEquity'), 'current_ratio': info.get('currentRatio'), 'price_to_book': info.get('priceToBook'), 'volume': info.get('volume'), 'avg_volume': info.get('averageVolume') } except Exception as e: print(f"Error fetching data: {e}") return {'valid': False}

async _agent_stock_screener(self, stock_data: Dict) -> Dict: """ Agent 1: Validate stock against investment screening criteria. Criteria: - Price: Within budget (e.g., ≤ RM 1.00) - PE Ratio: 4-15 (fair valuation, avoid loss-makers) - ROE: ≥ 5% (profitability threshold) - EPS: ≥ 0.01 (actual profit generation) - Dividend Yield: ≥ 1% (passive income potential) - Market Cap: ≥ RM 50M (stability filter) - Volume: ≥ 50,000 shares/day (liquidity) """ criteria = { 'max_price': 1.00, 'min_pe': 4, 'max_pe': 15, 'min_roe': 5, 'min_eps': 0.01, 'min_dividend_yield': 1, 'min_market_cap': 50_000_000, 'min_volume': 50_000 } # Build screening prompt for AI validation screening_prompt = f""" You are a stock screening agent. Validate this stock against investment criteria. Stock Data: - Ticker: {stock_data['ticker']} - Price: RM {stock_data['current_price']} - PE Ratio: {stock_data['pe_ratio']} - ROE: {stock_data['roe']}% - EPS: {stock_data['eps']} - Dividend Yield: {stock_data['dividend_yield']}% - Market Cap: RM {stock_data['market_cap']:,} - Avg Volume: {stock_data['avg_volume']:,} Criteria: {json.dumps(criteria, indent=2)} Return JSON: {{ "passed": true/false, "criteria_results": {{ "price": {{"passed": bool, "actual": value, "limit": value}}, "pe_ratio": {{"passed": bool, "actual": value, "range": [min, max]}}, ... }}, "summary": "Brief explanation of pass/fail" }} """ response = self.model.generate_content(screening_prompt) screening_result = self._parse_json_response(response.text) return screening_result

async _agent_fundamental_analyst(self, stock_data: Dict) -> Dict: """ Agent 2: Score fundamentals, financial health, and valuation. Scoring breakdown: 1. Fundamentals & Profitability (20 points): - ROE quality (0-10 points) - EPS growth (0-5 points) - Profit margins (0-5 points) 2. Financial Health & Solvency (15 points): - Debt-to-equity ratio (0-7 points) - Current ratio (0-5 points) - Book value strength (0-3 points) 3. Valuation (20 points): - PE ratio fairness (0-10 points) - Price-to-book (0-5 points) - Price-to-sales (0-5 points) """ fundamental_prompt = f""" You are a fundamental analysis expert. Score this stock across 3 categories. Stock: {stock_data['ticker']} - {stock_data['name']} Financial Metrics: - ROE: {stock_data['roe']:.2f}% - EPS: {stock_data['eps']} - PE Ratio: {stock_data['pe_ratio']} - Price-to-Book: {stock_data['price_to_book']} - Debt-to-Equity: {stock_data['debt_to_equity']} - Current Ratio: {stock_data['current_ratio']} Score each category (0-max points) and provide detailed reasoning: 1. Fundamentals & Profitability (0-20 points) - Evaluate ROE quality, EPS strength, margins - Higher ROE (>15%) = excellent, 5-15% = good, <5% = weak 2. Financial Health & Solvency (0-15 points) - Assess debt levels, liquidity, balance sheet strength - Low debt (<0.5 D/E) = strong, 0.5-1.5 = moderate, >1.5 = risky 3. Valuation (0-20 points) - Compare PE to industry average and historical range - PE 4-10 = undervalued, 10-15 = fair, >15 = expensive Return JSON with scores and detailed analysis for each category. """ response = self.model.generate_content(fundamental_prompt) analysis = self._parse_json_response(response.text) return { 'category_scores': { 'fundamentals_profitability': analysis['fundamentals_score'], 'financial_health': analysis['health_score'], 'valuation': analysis['valuation_score'] }, 'total_fundamental_score': sum(analysis['category_scores'].values()), 'detailed_analysis': analysis['reasoning'] }

async _agent_business_moat(self, ticker: str, stock_data: Dict) -> Dict: """ Agent 3: Research business moat via web search and score competitive advantages. Evaluates: - Brand strength and recognition - Network effects and switching costs - Cost advantages and economies of scale - Regulatory barriers and patents - Competitive positioning """ # Search for company competitive advantages search_query = f"{stock_data['name']} competitive advantages business moat Malaysia" search_results = self.tavily.search( query=search_query, search_depth="advanced", max_results=5 ) # Extract relevant information from search results research_content = "\n\n".join([ f"Source: {r['url']}\n{r['content']}" for r in search_results['results'] ]) moat_prompt = f""" You are a business moat analyst. Evaluate the competitive advantages of this company. Company: {stock_data['name']} ({ticker}) Industry Research: {research_content} Analyze and score (0-15 points) based on: 1. Brand Strength (0-5 points): - Market recognition, customer loyalty, brand premium 2. Competitive Barriers (0-5 points): - Switching costs, network effects, regulatory moats 3. Cost/Scale Advantages (0-5 points): - Economies of scale, proprietary technology, unique assets Rate each dimension and provide total moat score (0-15). Explain strengths and weaknesses clearly. Return JSON with scores and detailed moat analysis. """ response = self.model.generate_content(moat_prompt) moat_analysis = self._parse_json_response(response.text) return { 'moat_score': moat_analysis['total_moat_score'], 'moat_strength': self._classify_moat(moat_analysis['total_moat_score']), 'key_advantages': moat_analysis['strengths'], 'vulnerabilities': moat_analysis['weaknesses'], 'research_sources': [r['url'] for r in search_results['results']] } def _classify_moat(self, score: float) -> str: """Classify moat strength based on score.""" if score >= 12: return "Wide Moat" elif score >= 8: return "Narrow Moat" else: return "No Moat"