144 lines
4.9 KiB
Python
144 lines
4.9 KiB
Python
# gets data/stats
|
|
|
|
import yfinance as yf
|
|
from typing import Annotated, Callable, Any, Optional
|
|
from pandas import DataFrame
|
|
import pandas as pd
|
|
from functools import wraps
|
|
|
|
from .utils import SavePathType, decorate_all_methods
|
|
|
|
|
|
def init_ticker(func: Callable) -> Callable:
|
|
"""Decorator to initialize yf.Ticker and pass it to the function."""
|
|
|
|
@wraps(func)
|
|
def wrapper(
|
|
symbol: Annotated[str, "ticker symbol"], *args, **kwargs
|
|
) -> Any:
|
|
ticker = yf.Ticker(symbol)
|
|
return func(ticker, *args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
|
|
@decorate_all_methods(init_ticker)
|
|
class YFinanceUtils:
|
|
|
|
def get_stock_data(
|
|
symbol: Annotated[str, "ticker symbol"],
|
|
start_date: Annotated[
|
|
str, "start date for retrieving stock price data, YYYY-mm-dd"
|
|
],
|
|
end_date: Annotated[
|
|
str, "end date for retrieving stock price data, YYYY-mm-dd"
|
|
],
|
|
save_path: SavePathType = None,
|
|
) -> DataFrame:
|
|
"""retrieve stock price data for designated ticker symbol"""
|
|
ticker = symbol
|
|
# add one day to the end_date so that the data range is inclusive
|
|
end_date = pd.to_datetime(end_date) + pd.DateOffset(days=1)
|
|
end_date = end_date.strftime("%Y-%m-%d")
|
|
stock_data = ticker.history(start=start_date, end=end_date)
|
|
# save_output(stock_data, f"Stock data for {ticker.ticker}", save_path)
|
|
return stock_data
|
|
|
|
def get_stock_info(
|
|
symbol: Annotated[str, "ticker symbol"],
|
|
) -> dict:
|
|
"""Fetches and returns latest stock information."""
|
|
ticker = symbol
|
|
stock_info = ticker.info
|
|
return stock_info
|
|
|
|
def get_company_info(
|
|
symbol: Annotated[str, "ticker symbol"],
|
|
save_path: Optional[str] = None,
|
|
) -> DataFrame:
|
|
"""Fetches and returns company information as a DataFrame."""
|
|
ticker = symbol
|
|
info = ticker.info
|
|
company_info = {
|
|
"Company Name": info.get("shortName", "N/A"),
|
|
"Industry": info.get("industry", "N/A"),
|
|
"Sector": info.get("sector", "N/A"),
|
|
"Country": info.get("country", "N/A"),
|
|
"Website": info.get("website", "N/A"),
|
|
}
|
|
company_info_df = DataFrame([company_info])
|
|
if save_path:
|
|
company_info_df.to_csv(save_path)
|
|
print(f"Company info for {ticker.ticker} saved to {save_path}")
|
|
return company_info_df
|
|
|
|
def get_stock_dividends(
|
|
symbol: Annotated[str, "ticker symbol"],
|
|
save_path: Optional[str] = None,
|
|
) -> DataFrame:
|
|
"""Fetches and returns the latest dividends data as a DataFrame."""
|
|
ticker = symbol
|
|
dividends = ticker.dividends
|
|
if save_path:
|
|
dividends.to_csv(save_path)
|
|
print(f"Dividends for {ticker.ticker} saved to {save_path}")
|
|
return dividends
|
|
|
|
def get_income_stmt(symbol: Annotated[str, "ticker symbol"]) -> DataFrame:
|
|
"""Fetches and returns the latest income statement of the company."""
|
|
ticker = symbol
|
|
income_stmt = ticker.financials
|
|
return income_stmt
|
|
|
|
def get_balance_sheet(symbol: Annotated[str, "ticker symbol"]) -> DataFrame:
|
|
"""Fetches and returns the latest balance sheet of the company."""
|
|
ticker = symbol
|
|
balance_sheet = ticker.balance_sheet
|
|
return balance_sheet
|
|
|
|
def get_cash_flow(symbol: Annotated[str, "ticker symbol"]) -> DataFrame:
|
|
"""Fetches and returns the latest cash flow statement of the company."""
|
|
ticker = symbol
|
|
cash_flow = ticker.cashflow
|
|
return cash_flow
|
|
|
|
def get_analyst_recommendations(
|
|
symbol: Annotated[str, "ticker symbol"]
|
|
) -> tuple:
|
|
"""Fetches analyst recommendations with statistical validation."""
|
|
ticker = symbol
|
|
try:
|
|
recommendations = ticker.recommendations
|
|
except (AttributeError, ValueError) as e:
|
|
raise ValueError(
|
|
f"Failed to fetch recommendations for {ticker.ticker}: {e}"
|
|
)
|
|
|
|
if recommendations is None or recommendations.empty:
|
|
return None, 0
|
|
|
|
try:
|
|
# Get the most recent recommendations (first row)
|
|
latest_row = recommendations.iloc[0]
|
|
|
|
# Remove non-numeric columns (like 'period')
|
|
numeric_cols = latest_row.select_dtypes(include=['number'])
|
|
if numeric_cols.empty:
|
|
return None, 0
|
|
|
|
# Find maximum with statistical validation
|
|
max_votes = numeric_cols.max()
|
|
if pd.isna(max_votes) or max_votes <= 0:
|
|
return None, 0
|
|
|
|
# Get recommendation with highest count
|
|
max_recommendation = numeric_cols.idxmax()
|
|
|
|
# Convert to int for consistency
|
|
max_votes = int(max_votes)
|
|
|
|
return max_recommendation, max_votes
|
|
|
|
except (IndexError, KeyError, ValueError) as e:
|
|
raise ValueError(f"Error processing recommendations data: {e}")
|