Implement Analysis

This commit is contained in:
Filip Krása
2024-08-11 10:53:09 +02:00
parent 2594eef3a3
commit 408d0659e5
3 changed files with 315 additions and 65 deletions

View File

@@ -240,40 +240,67 @@ class TickerBase:
return data.to_dict()
return data
def get_analyst_price_target(self, proxy=None, as_dict=False):
def get_analyst_price_targets(self, proxy=None) -> dict:
"""
Keys: current low high mean median
"""
self._analysis.proxy = proxy or self.proxy
data = self._analysis.analyst_price_target
if as_dict:
return data.to_dict()
data = self._analysis.analyst_price_targets
return data
def get_rev_forecast(self, proxy=None, as_dict=False):
def get_earnings_estimate(self, proxy=None, as_dict=False):
"""
Index: 0q +1q 0y +1y
Columns: numberOfAnalysts avg low high yearAgoEps growth
"""
self._analysis.proxy = proxy or self.proxy
data = self._analysis.rev_est
if as_dict:
return data.to_dict()
return data
data = self._analysis.earnings_estimate
return data.to_dict() if as_dict else data
def get_earnings_forecast(self, proxy=None, as_dict=False):
def get_revenue_estimate(self, proxy=None, as_dict=False):
"""
Index: 0q +1q 0y +1y
Columns: numberOfAnalysts avg low high yearAgoRevenue growth
"""
self._analysis.proxy = proxy or self.proxy
data = self._analysis.eps_est
if as_dict:
return data.to_dict()
return data
data = self._analysis.revenue_estimate
return data.to_dict() if as_dict else data
def get_trend_details(self, proxy=None, as_dict=False):
def get_earnings_history(self, proxy=None, as_dict=False):
"""
Index: pd.DatetimeIndex
Columns: epsEstimate epsActual epsDifference surprisePercent
"""
self._analysis.proxy = proxy or self.proxy
data = self._analysis.analyst_trend_details
if as_dict:
return data.to_dict()
return data
data = self._analysis.earnings_history
return data.to_dict() if as_dict else data
def get_earnings_trend(self, proxy=None, as_dict=False):
def get_eps_trend(self, proxy=None, as_dict=False):
"""
Index: 0q +1q 0y +1y
Columns: current 7daysAgo 30daysAgo 60daysAgo 90daysAgo
"""
self._analysis.proxy = proxy or self.proxy
data = self._analysis.earnings_trend
if as_dict:
return data.to_dict()
return data
data = self._analysis.eps_trend
return data.to_dict() if as_dict else data
def get_eps_revisions(self, proxy=None, as_dict=False):
"""
Index: 0q +1q 0y +1y
Columns: upLast7days upLast30days downLast7days downLast30days
"""
self._analysis.proxy = proxy or self.proxy
data = self._analysis.eps_revisions
return data.to_dict() if as_dict else data
def get_growth_estimates(self, proxy=None, as_dict=False):
"""
Index: 0q +1q 0y +1y +5y -5y
Columns: stock industry sector index
"""
self._analysis.proxy = proxy or self.proxy
data = self._analysis.growth_estimates
return data.to_dict() if as_dict else data
def get_earnings(self, proxy=None, as_dict=False, freq="yearly"):
"""

View File

@@ -1,7 +1,11 @@
import pandas as pd
import requests
from yfinance import utils
from yfinance.data import YfData
from yfinance.exceptions import YFNotImplementedError
from yfinance.const import quote_summary_valid_modules
from yfinance.scrapers.quote import _QUOTE_SUMMARY_URL_
from yfinance.exceptions import YFException
class Analysis:
@@ -11,39 +15,250 @@ class Analysis:
self._symbol = symbol
self.proxy = proxy
# In quoteSummary the 'earningsTrend' module contains most of the data below.
# The format of data is not optimal so each function will process it's part of the data.
# This variable works as a cache.
self._earnings_trend = None
self._analyst_trend_details = None
self._analyst_price_target = None
self._rev_est = None
self._eps_est = None
self._already_scraped = False
self._analyst_price_targets = None
self._earnings_estimate = None
self._revenue_estimate = None
self._earnings_history = None
self._eps_trend = None
self._eps_revisions = None
self._growth_estimates = None
@property
def earnings_trend(self) -> pd.DataFrame:
def analyst_price_targets(self) -> dict:
if self._analyst_price_targets is not None:
return self._analyst_price_targets
try:
data = self._fetch(['financialData'])
data = data['quoteSummary']['result'][0]['financialData']
except:
self._analyst_price_targets = {}
return self._analyst_price_targets
keys = [
('currentPrice', 'current'),
('targetLowPrice', 'low'),
('targetHighPrice', 'high'),
('targetMeanPrice', 'mean'),
('targetMedianPrice', 'median'),
]
self._analyst_price_targets = {newKey: data.get(oldKey, None) for oldKey, newKey in keys}
return self._analyst_price_targets
@property
def earnings_estimate(self) -> pd.DataFrame:
if self._earnings_estimate is not None:
return self._earnings_estimate
if self._earnings_trend is None:
raise YFNotImplementedError('earnings_trend')
return self._earnings_trend
self._fetch_earnings_trend()
data_dict = {
'numberOfAnalysts': [],
'avg': [],
'low': [],
'high': [],
'yearAgoEps': [],
'growth': []
}
periods = []
for item in self._earnings_trend[:4]:
periods.append(item['period'])
earnings_estimate = item.get('earningsEstimate', {})
for key in data_dict.keys():
data_dict[key].append(earnings_estimate.get(key, {}).get('raw', None))
self._earnings_estimate = pd.DataFrame(data_dict, index=periods)
return self._earnings_estimate
@property
def analyst_trend_details(self) -> pd.DataFrame:
if self._analyst_trend_details is None:
raise YFNotImplementedError('analyst_trend_details')
return self._analyst_trend_details
def revenue_estimate(self) -> pd.DataFrame:
if self._revenue_estimate is not None:
return self._revenue_estimate
if self._earnings_trend is None:
self._fetch_earnings_trend()
data_dict = {
'numberOfAnalysts': [],
'avg': [],
'low': [],
'high': [],
'yearAgoRevenue': [],
'growth': []
}
periods = []
for item in self._earnings_trend[:4]:
periods.append(item['period'])
revenue_estimate = item.get('revenueEstimate', {})
for key in data_dict.keys():
data_dict[key].append(revenue_estimate.get(key, {}).get('raw', None))
self._revenue_estimate = pd.DataFrame(data_dict, index=periods)
return self._revenue_estimate
@property
def analyst_price_target(self) -> pd.DataFrame:
if self._analyst_price_target is None:
raise YFNotImplementedError('analyst_price_target')
return self._analyst_price_target
def earnings_history(self) -> pd.DataFrame:
if self._earnings_history is not None:
return self._earnings_history
try:
data = self._fetch(['earningsHistory'])
data = data['quoteSummary']['result'][0]['earningsHistory']['history']
except:
self._earnings_history = pd.DataFrame()
return self._earnings_history
data_dict = {
'epsEstimate': [],
'epsActual': [],
'epsDifference': [],
'surprisePercent': []
}
quarters = []
for item in data:
quarters.append(item.get('quarter', {}).get('fmt', None))
for key in data_dict.keys():
data_dict[key].append(item.get(key, {}).get('raw', None))
datetime_index = pd.to_datetime(quarters, format='%Y-%m-%d')
self._earnings_history = pd.DataFrame(data_dict, index=datetime_index)
return self._earnings_history
@property
def rev_est(self) -> pd.DataFrame:
if self._rev_est is None:
raise YFNotImplementedError('rev_est')
return self._rev_est
def eps_trend(self) -> pd.DataFrame:
if self._eps_trend is not None:
return self._eps_trend
if self._earnings_trend is None:
self._fetch_earnings_trend()
data_dict = {
'current': [],
'7daysAgo': [],
'30daysAgo': [],
'60daysAgo': [],
'90daysAgo': []
}
periods = []
for item in self._earnings_trend[:4]:
periods.append(item['period'])
eps_trend = item.get('epsTrend', {})
for key in data_dict.keys():
data_dict[key].append(eps_trend.get(key, {}).get('raw', None))
self._eps_trend = pd.DataFrame(data_dict, index=periods)
return self._eps_trend
@property
def eps_est(self) -> pd.DataFrame:
if self._eps_est is None:
raise YFNotImplementedError('eps_est')
return self._eps_est
def eps_revisions(self) -> pd.DataFrame:
if self._eps_revisions is not None:
return self._eps_revisions
if self._earnings_trend is None:
self._fetch_earnings_trend()
data_dict = {
'upLast7days': [],
'upLast30days': [],
'downLast7days': [],
'downLast30days': []
}
periods = []
for item in self._earnings_trend[:4]:
periods.append(item['period'])
eps_revisions = item.get('epsRevisions', {})
for key in data_dict.keys():
data_dict[key].append(eps_revisions.get(key, {}).get('raw', None))
self._eps_revisions = pd.DataFrame(data_dict, index=periods)
return self._eps_revisions
@property
def growth_estimates(self) -> pd.DataFrame:
if self._growth_estimates is not None:
return self._growth_estimates
if self._earnings_trend is None:
self._fetch_earnings_trend()
try:
trends = self._fetch(['industryTrend', 'sectorTrend', 'indexTrend'])
trends = trends['quoteSummary']['result'][0]
except:
self._growth_estimates = pd.DataFrame()
return self._growth_estimates
data_dict = {
'0q': [],
'+1q': [],
'0y': [],
'+1y': [],
'+5y': [],
'-5y': []
}
# make sure no column is empty
dummy_trend = [{'period': key, 'growth': None} for key in data_dict.keys()]
industry_trend = trends['industryTrend']['estimates'] or dummy_trend
sector_trend = trends['sectorTrend']['estimates'] or dummy_trend
index_trend = trends['indexTrend']['estimates'] or dummy_trend
for item in self._earnings_trend:
period = item['period']
data_dict[period].append(item.get('growth', {}).get('raw', None))
for item in industry_trend:
period = item['period']
data_dict[period].append(item.get('growth', None))
for item in sector_trend:
period = item['period']
data_dict[period].append(item.get('growth', None))
for item in index_trend:
period = item['period']
data_dict[period].append(item.get('growth', None))
cols = ['stock', 'industry', 'sector', 'index']
self._growth_estimates = pd.DataFrame(data_dict, index=cols).T
return self._growth_estimates
# modified version from quote.py
def _fetch(self, modules: list):
if not isinstance(modules, list):
raise YFException("Should provide a list of modules, see available modules using `valid_modules`")
modules = ','.join([m for m in modules if m in quote_summary_valid_modules])
if len(modules) == 0:
raise YFException("No valid modules provided, see available modules using `valid_modules`")
params_dict = {"modules": modules, "corsDomain": "finance.yahoo.com", "formatted": "false", "symbol": self._symbol}
try:
result = self._data.get_raw_json(_QUOTE_SUMMARY_URL_ + f"/{self._symbol}", user_agent_headers=self._data.user_agent_headers, params=params_dict, proxy=self.proxy)
except requests.exceptions.HTTPError as e:
utils.get_yf_logger().error(str(e))
return None
return result
def _fetch_earnings_trend(self) -> None:
try:
data = self._fetch(['earningsTrend'])
self._earnings_trend = data['quoteSummary']['result'][0]['earningsTrend']['trend']
except:
self._earnings_trend = []

View File

@@ -240,12 +240,32 @@ class Ticker(TickerBase):
return self.quarterly_cash_flow
@property
def analyst_price_target(self) -> _pd.DataFrame:
return self.get_analyst_price_target()
def analyst_price_targets(self) -> dict:
return self.get_analyst_price_targets()
@property
def revenue_forecasts(self) -> _pd.DataFrame:
return self.get_rev_forecast()
def earnings_estimate(self) -> _pd.DataFrame:
return self.get_earnings_estimate()
@property
def revenue_estimate(self) -> _pd.DataFrame:
return self.get_revenue_estimate()
@property
def earnings_history(self) -> _pd.DataFrame:
return self.get_earnings_history()
@property
def eps_trend(self) -> _pd.DataFrame:
return self.get_eps_trend()
@property
def eps_revisions(self) -> _pd.DataFrame:
return self.get_eps_revisions()
@property
def growth_estimates(self) -> _pd.DataFrame:
return self.get_growth_estimates()
@property
def sustainability(self) -> _pd.DataFrame:
@@ -261,22 +281,10 @@ class Ticker(TickerBase):
def news(self) -> list:
return self.get_news()
@property
def trend_details(self) -> _pd.DataFrame:
return self.get_trend_details()
@property
def earnings_trend(self) -> _pd.DataFrame:
return self.get_earnings_trend()
@property
def earnings_dates(self) -> _pd.DataFrame:
return self.get_earnings_dates()
@property
def earnings_forecasts(self) -> _pd.DataFrame:
return self.get_earnings_forecast()
@property
def history_metadata(self) -> dict:
return self.get_history_metadata()