""" Enhanced financial data fetcher using yfinance. Collects company data, sector data, and peer comparison data. """ import yfinance as yf import pandas as pd import numpy as np from typing import Dict, List, Optional, Tuple from datetime import datetime import warnings warnings.filterwarnings('ignore') class FinancialDataFetcher: """Comprehensive data fetcher for fundamental analysis""" def __init__(self, ticker: str): """ Initialize fetcher for a ticker Args: ticker: Stock ticker symbol """ self.ticker = ticker.upper() self.stock = yf.Ticker(self.ticker) self.info = None self.sector = None self.industry = None def fetch_company_info(self) -> Dict: """Fetch basic company information""" try: self.info = self.stock.info self.sector = self.info.get('sector', 'Unknown') self.industry = self.info.get('industry', 'Unknown') return { 'ticker': self.ticker, 'company_name': self.info.get('longName', self.ticker), 'sector': self.sector, 'industry': self.industry, 'market_cap': self.info.get('marketCap', 0), 'country': self.info.get('country', 'Unknown'), 'website': self.info.get('website', ''), 'business_summary': self.info.get('longBusinessSummary', '') } except Exception as e: print(f"Error fetching company info: {e}") return {} def fetch_financial_statements(self) -> Dict[str, pd.DataFrame]: """Fetch all financial statements (annual)""" try: return { 'income_statement': self.stock.income_stmt, 'balance_sheet': self.stock.balance_sheet, 'cash_flow': self.stock.cashflow, 'quarterly_income': self.stock.quarterly_income_stmt, 'quarterly_balance': self.stock.quarterly_balance_sheet, 'quarterly_cashflow': self.stock.quarterly_cashflow } except Exception as e: print(f"Error fetching financial statements: {e}") return {} def fetch_key_metrics(self) -> Dict: """Fetch key financial metrics and ratios""" try: info = self.info if self.info else self.stock.info return { # Price metrics 'current_price': info.get('currentPrice', info.get('regularMarketPrice', 0)), 'previous_close': info.get('previousClose', 0), '52_week_high': info.get('fiftyTwoWeekHigh', 0), '52_week_low': info.get('fiftyTwoWeekLow', 0), # Valuation metrics 'market_cap': info.get('marketCap', 0), 'enterprise_value': info.get('enterpriseValue', 0), 'trailing_pe': info.get('trailingPE'), 'forward_pe': info.get('forwardPE'), 'peg_ratio': info.get('pegRatio'), 'price_to_book': info.get('priceToBook'), 'price_to_sales': info.get('priceToSalesTrailing12Months'), 'ev_to_revenue': info.get('enterpriseToRevenue'), 'ev_to_ebitda': info.get('enterpriseToEbitda'), # Profitability metrics 'profit_margin': info.get('profitMargins'), 'operating_margin': info.get('operatingMargins'), 'gross_margin': info.get('grossMargins'), 'ebitda_margin': self._calculate_ebitda_margin(info), # Returns 'return_on_assets': info.get('returnOnAssets'), 'return_on_equity': info.get('returnOnEquity'), # Growth 'revenue_growth': info.get('revenueGrowth'), 'earnings_growth': info.get('earningsGrowth'), # Financial health 'total_cash': info.get('totalCash', 0), 'total_debt': info.get('totalDebt', 0), 'debt_to_equity': info.get('debtToEquity'), 'current_ratio': info.get('currentRatio'), 'quick_ratio': info.get('quickRatio'), # Cash flow 'operating_cash_flow': info.get('operatingCashflow', 0), 'free_cash_flow': info.get('freeCashflow', 0), # Per share 'book_value_per_share': info.get('bookValue'), 'revenue_per_share': info.get('revenuePerShare'), 'eps_trailing': info.get('trailingEps'), 'eps_forward': info.get('forwardEps'), # Shares 'shares_outstanding': info.get('sharesOutstanding', 0), # Other 'beta': info.get('beta'), 'dividend_yield': info.get('dividendYield'), } except Exception as e: print(f"Error fetching key metrics: {e}") return {} def _calculate_ebitda_margin(self, info: Dict) -> Optional[float]: """Calculate EBITDA margin if available""" try: ebitda = info.get('ebitda') revenue = info.get('totalRevenue') if ebitda and revenue and revenue > 0: return ebitda / revenue except: pass return None def fetch_peer_tickers(self, max_peers: int = 10) -> List[str]: """ Fetch peer company tickers in the same sector/industry Args: max_peers: Maximum number of peers to return Returns: List of peer ticker symbols """ try: if not self.sector or self.sector == 'Unknown': self.fetch_company_info() # Try to get recommendations which sometimes include peers recommendations = self.stock.recommendations peers = set() # Fallback: Use a simple sector-based approach # In production, you'd use a proper database or API for peer identification # For now, we'll return an empty list and let users provide peers manually return [] except Exception as e: print(f"Error fetching peer tickers: {e}") return [] def fetch_peer_data(self, peer_tickers: List[str]) -> Dict[str, Dict]: """ Fetch key metrics for peer companies Args: peer_tickers: List of peer ticker symbols Returns: Dictionary mapping ticker to metrics """ peer_data = {} for ticker in peer_tickers: try: print(f"Fetching data for peer: {ticker}") peer_stock = yf.Ticker(ticker) peer_info = peer_stock.info peer_data[ticker] = { 'company_name': peer_info.get('longName', ticker), 'market_cap': peer_info.get('marketCap', 0), 'trailing_pe': peer_info.get('trailingPE'), 'forward_pe': peer_info.get('forwardPE'), 'peg_ratio': peer_info.get('pegRatio'), 'price_to_book': peer_info.get('priceToBook'), 'profit_margin': peer_info.get('profitMargins'), 'operating_margin': peer_info.get('operatingMargins'), 'gross_margin': peer_info.get('grossMargins'), 'return_on_equity': peer_info.get('returnOnEquity'), 'return_on_assets': peer_info.get('returnOnAssets'), 'revenue_growth': peer_info.get('revenueGrowth'), 'earnings_growth': peer_info.get('earningsGrowth'), 'debt_to_equity': peer_info.get('debtToEquity'), 'current_ratio': peer_info.get('currentRatio'), 'free_cash_flow': peer_info.get('freeCashflow', 0), 'beta': peer_info.get('beta'), } except Exception as e: print(f"Error fetching data for {ticker}: {e}") continue return peer_data def calculate_sector_metrics(self, peer_data: Dict[str, Dict]) -> Dict: """ Calculate sector-wide metrics from peer data Args: peer_data: Dictionary of peer metrics Returns: Dictionary of sector averages/medians """ if not peer_data: return {} # Collect all metrics metrics = { 'trailing_pe': [], 'forward_pe': [], 'peg_ratio': [], 'price_to_book': [], 'profit_margin': [], 'operating_margin': [], 'gross_margin': [], 'return_on_equity': [], 'return_on_assets': [], 'revenue_growth': [], 'earnings_growth': [], 'debt_to_equity': [], 'current_ratio': [], 'beta': [] } # Aggregate peer metrics for ticker, data in peer_data.items(): for key in metrics.keys(): value = data.get(key) if value is not None and not (isinstance(value, float) and np.isnan(value)): metrics[key].append(value) # Calculate sector statistics sector_metrics = {} for key, values in metrics.items(): if values: sector_metrics[f'{key}_median'] = float(np.median(values)) sector_metrics[f'{key}_mean'] = float(np.mean(values)) sector_metrics[f'{key}_min'] = float(np.min(values)) sector_metrics[f'{key}_max'] = float(np.max(values)) sector_metrics[f'{key}_count'] = len(values) return sector_metrics def fetch_all_data(self, peer_tickers: Optional[List[str]] = None) -> Dict: """ Fetch all data for comprehensive analysis Args: peer_tickers: Optional list of peer tickers for comparison Returns: Complete dataset """ print(f"\n{'='*60}") print(f"Fetching data for {self.ticker}...") print(f"{'='*60}\n") # Company data company_info = self.fetch_company_info() print(f"✓ Company: {company_info.get('company_name', self.ticker)}") print(f"✓ Sector: {company_info.get('sector', 'Unknown')}") print(f"✓ Industry: {company_info.get('industry', 'Unknown')}\n") # Financial statements print("Fetching financial statements...") statements = self.fetch_financial_statements() # Key metrics print("Fetching key metrics...") metrics = self.fetch_key_metrics() # Peer data peer_data = {} sector_metrics = {} if peer_tickers: print(f"\nFetching peer data for {len(peer_tickers)} companies...") peer_data = self.fetch_peer_data(peer_tickers) print(f"✓ Successfully fetched data for {len(peer_data)} peers") if peer_data: print("Calculating sector metrics...") sector_metrics = self.calculate_sector_metrics(peer_data) print(f"✓ Sector metrics calculated\n") return { 'ticker': self.ticker, 'fetch_date': datetime.now().isoformat(), 'company_info': company_info, 'financial_statements': statements, 'metrics': metrics, 'peer_data': peer_data, 'sector_metrics': sector_metrics } if __name__ == "__main__": # Example usage ticker = input("Enter ticker symbol: ").upper() fetcher = FinancialDataFetcher(ticker) # Ask for peer tickers peers_input = input("Enter peer tickers (comma-separated, or press Enter to skip): ").strip() peer_tickers = [p.strip().upper() for p in peers_input.split(',')] if peers_input else None data = fetcher.fetch_all_data(peer_tickers) print(f"\n{'='*60}") print("DATA COLLECTION COMPLETE") print(f"{'='*60}") print(f"Company: {data['company_info'].get('company_name')}") print(f"Metrics collected: {len([k for k, v in data['metrics'].items() if v is not None])}") if data['peer_data']: print(f"Peers analyzed: {len(data['peer_data'])}") print(f"Sector metrics: {len(data['sector_metrics'])}")