Spaces:
Running
Running
| """ | |
| 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'])}") | |