Spaces:
Sleeping
Sleeping
File size: 10,678 Bytes
d98f7af |
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 |
import yfinance as yf
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
class DataFetcher:
def __init__(self):
# Cache to store fetched data
self.cache = {}
def get_ticker_info(self, ticker):
"""Get basic information about a ticker"""
cache_key = f"{ticker}_info"
if cache_key in self.cache:
return self.cache[cache_key]
try:
stock = yf.Ticker(ticker)
info = stock.info
self.cache[cache_key] = info
return info
except Exception as e:
print(f"Error fetching info for {ticker}: {str(e)}")
return {}
def get_price_history(self, ticker, period="1y"):
"""Get historical price data for a ticker"""
cache_key = f"{ticker}_price_{period}"
if cache_key in self.cache:
return self.cache[cache_key]
try:
stock = yf.Ticker(ticker)
history = stock.history(period=period)
if history.empty:
# Try with a shorter period if 1y fails
history = stock.history(period="6mo")
if history.empty:
# Try with an even shorter period if 6mo fails
history = stock.history(period="3mo")
if not history.empty:
# Use Close prices
price_series = history['Close']
self.cache[cache_key] = price_series
return price_series
else:
print(f"No price history available for {ticker}")
return pd.Series()
except Exception as e:
print(f"Error fetching price history for {ticker}: {str(e)}")
return pd.Series()
def get_key_metrics(self, ticker):
"""Get key financial metrics for a ticker"""
cache_key = f"{ticker}_metrics"
if cache_key in self.cache:
return self.cache[cache_key]
try:
stock = yf.Ticker(ticker)
info = stock.info
# Determine currency symbol based on country or exchange
currency_symbol = '$' # Default to USD
if 'currency' in info:
if info['currency'] == 'INR':
currency_symbol = '₹'
elif info['currency'] == 'EUR':
currency_symbol = '€'
elif info['currency'] == 'GBP':
currency_symbol = '£'
elif info['currency'] == 'JPY':
currency_symbol = '¥'
# Check if it's an Indian stock based on ticker suffix
if ticker.endswith('.NS') or ticker.endswith('.BO'):
currency_symbol = '₹'
# Extract key metrics
metrics = {
'Company Name': info.get('longName', 'N/A'),
'Sector': info.get('sector', 'N/A'),
'Industry': info.get('industry', 'N/A'),
'Country': info.get('country', 'N/A'),
'Currency Symbol': currency_symbol,
'Current Price': info.get('currentPrice', info.get('regularMarketPrice', 'N/A')),
'Market Cap': info.get('marketCap', 'N/A'),
'P/E Ratio': info.get('trailingPE', 'N/A'),
'Forward P/E': info.get('forwardPE', 'N/A'),
'P/B Ratio': info.get('priceToBook', 'N/A'),
'EV/EBITDA': info.get('enterpriseToEbitda', 'N/A'),
'EV/Revenue': info.get('enterpriseToRevenue', 'N/A'),
'PEG Ratio': info.get('pegRatio', 'N/A'),
'Dividend Yield (%)': info.get('dividendYield', 'N/A') * 100 if info.get('dividendYield') is not None else 'N/A',
'EPS': info.get('trailingEps', 'N/A'),
'Profit Margin': info.get('profitMargins', 'N/A') * 100 if info.get('profitMargins') is not None else 'N/A',
'Operating Margin': info.get('operatingMargins', 'N/A') * 100 if info.get('operatingMargins') is not None else 'N/A',
'ROE': info.get('returnOnEquity', 'N/A') * 100 if info.get('returnOnEquity') is not None else 'N/A',
'ROA': info.get('returnOnAssets', 'N/A') * 100 if info.get('returnOnAssets') is not None else 'N/A',
'Revenue Growth': info.get('revenueGrowth', 'N/A') * 100 if info.get('revenueGrowth') is not None else 'N/A',
'Earnings Growth': info.get('earningsGrowth', 'N/A') * 100 if info.get('earningsGrowth') is not None else 'N/A',
'Debt to Equity': info.get('debtToEquity', 'N/A') / 100 if info.get('debtToEquity') is not None else 'N/A',
'Current Ratio': info.get('currentRatio', 'N/A'),
'Quick Ratio': info.get('quickRatio', 'N/A'),
'Beta': info.get('beta', 'N/A'),
'52 Week High': info.get('fiftyTwoWeekHigh', 'N/A'),
'52 Week Low': info.get('fiftyTwoWeekLow', 'N/A'),
'50-Day MA': info.get('fiftyDayAverage', 'N/A'),
'200-Day MA': info.get('twoHundredDayAverage', 'N/A'),
'Shares Outstanding': info.get('sharesOutstanding', 'N/A'),
'Free Cash Flow': info.get('freeCashflow', 'N/A'),
'Operating Cash Flow': info.get('operatingCashflow', 'N/A'),
'Revenue Per Share': info.get('revenuePerShare', 'N/A'),
'Target Mean Price': info.get('targetMeanPrice', 'N/A'),
'Payout Ratio': info.get('payoutRatio', 'N/A') * 100 if info.get('payoutRatio') is not None else 'N/A',
'EBITDA Margins': info.get('ebitdaMargins', 'N/A') * 100 if info.get('ebitdaMargins') is not None else 'N/A',
'Gross Margins': info.get('grossMargins', 'N/A') * 100 if info.get('grossMargins') is not None else 'N/A'
}
self.cache[cache_key] = metrics
return metrics
except Exception as e:
print(f"Error fetching metrics for {ticker}: {str(e)}")
return {'error': str(e), 'Currency Symbol': '$'}
def get_financial_statements(self, ticker):
"""Get financial statements for a ticker"""
cache_key = f"{ticker}_financials"
if cache_key in self.cache:
return self.cache[cache_key]
try:
stock = yf.Ticker(ticker)
# Get annual financial statements
income_stmt = stock.income_stmt
balance_sheet = stock.balance_sheet
cash_flow = stock.cashflow
# Get quarterly financial statements with error handling
try:
quarterly_income_stmt = stock.quarterly_income_stmt
except Exception as e:
print(f"Error fetching quarterly income statement for {ticker}: {str(e)}")
quarterly_income_stmt = pd.DataFrame()
try:
quarterly_balance_sheet = stock.quarterly_balance_sheet
except Exception as e:
print(f"Error fetching quarterly balance sheet for {ticker}: {str(e)}")
quarterly_balance_sheet = pd.DataFrame()
try:
quarterly_cash_flow = stock.quarterly_cashflow
except Exception as e:
print(f"Error fetching quarterly cash flow for {ticker}: {str(e)}")
quarterly_cash_flow = pd.DataFrame()
# Package all statements
financial_statements = {
'income_stmt': income_stmt,
'balance_sheet': balance_sheet,
'cash_flow': cash_flow,
'quarterly_income_stmt': quarterly_income_stmt,
'quarterly_balance_sheet': quarterly_balance_sheet,
'quarterly_cash_flow': quarterly_cash_flow
}
self.cache[cache_key] = financial_statements
return financial_statements
except Exception as e:
print(f"Error fetching financial statements for {ticker}: {str(e)}")
# Return empty DataFrames for all statements
empty_df = pd.DataFrame()
return {
'income_stmt': empty_df,
'balance_sheet': empty_df,
'cash_flow': empty_df,
'quarterly_income_stmt': empty_df,
'quarterly_balance_sheet': empty_df,
'quarterly_cash_flow': empty_df
}
def get_free_cash_flow(self, ticker):
"""Get the most recent free cash flow value"""
try:
# First try to get FCF directly from info
metrics = self.get_key_metrics(ticker)
if metrics.get('Free Cash Flow', 'N/A') != 'N/A':
fcf = metrics.get('Free Cash Flow')
# Handle negative FCF by using 0 as the base
if fcf is not None and fcf < 0:
print(f"Warning: Negative FCF ({fcf}) for {ticker}, using 0 as base for DCF")
return 0
return fcf
# If not available, calculate from cash flow statement
financial_statements = self.get_financial_statements(ticker)
cash_flow = financial_statements['cash_flow']
if cash_flow.empty:
return None
# Check if 'Free Cash Flow' is directly available
if 'Free Cash Flow' in cash_flow.index:
fcf = cash_flow.loc['Free Cash Flow', cash_flow.columns[0]]
# Handle negative FCF by using 0 as the base
if fcf is not None and fcf < 0:
print(f"Warning: Negative FCF ({fcf}) for {ticker}, using 0 as base for DCF")
return 0
return fcf
# If not, try to calculate it from components
if 'Operating Cash Flow' in cash_flow.index and 'Capital Expenditure' in cash_flow.index:
operating_cf = cash_flow.loc['Operating Cash Flow', cash_flow.columns[0]]
capex = cash_flow.loc['Capital Expenditure', cash_flow.columns[0]]
if pd.notnull(operating_cf) and pd.notnull(capex):
fcf = operating_cf + capex # Note: capex is usually negative
# Handle negative FCF by using 0 as the base
if fcf is not None and fcf < 0:
print(f"Warning: Negative FCF ({fcf}) for {ticker}, using 0 as base for DCF")
return 0
return fcf
return None
except Exception as e:
print(f"Error calculating free cash flow for {ticker}: {str(e)}")
return None
|