Spaces:
Running
Running
File size: 12,951 Bytes
3fe0726 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 |
"""
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'])}")
|