|
|
import yfinance as yf |
|
|
from typing import Annotated, Callable, Any, Optional |
|
|
from pandas import DataFrame |
|
|
from functools import wraps |
|
|
|
|
|
from ..utils import save_output, 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: |
|
|
ticker = symbol |
|
|
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 stock information.""" |
|
|
ticker = symbol |
|
|
stock_info = ticker.info |
|
|
return stock_info |
|
|
|
|
|
def get_company_info( |
|
|
self, |
|
|
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( |
|
|
self, |
|
|
symbol: Annotated[str, "ticker symbol"], |
|
|
save_path: Optional[str] = None, |
|
|
) -> DataFrame: |
|
|
"""Fetches and returns the 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 income statement of the company as a DataFrame.""" |
|
|
ticker = symbol |
|
|
income_stmt = ticker.financials |
|
|
return income_stmt |
|
|
|
|
|
def get_balance_sheet(symbol: Annotated[str, "ticker symbol"]) -> DataFrame: |
|
|
"""Fetches and returns the balance sheet of the company as a DataFrame.""" |
|
|
ticker = symbol |
|
|
balance_sheet = ticker.balance_sheet |
|
|
return balance_sheet |
|
|
|
|
|
def get_cash_flow(symbol: Annotated[str, "ticker symbol"]) -> DataFrame: |
|
|
"""Fetches and returns the cash flow statement of the company as a DataFrame.""" |
|
|
ticker = symbol |
|
|
cash_flow = ticker.cashflow |
|
|
return cash_flow |
|
|
|
|
|
def get_analyst_recommendations(symbol: Annotated[str, "ticker symbol"]) -> tuple: |
|
|
"""Fetches the latest analyst recommendations and returns the most common recommendation and its count.""" |
|
|
ticker = symbol |
|
|
recommendations = ticker.recommendations |
|
|
if recommendations.empty: |
|
|
return None, 0 |
|
|
|
|
|
|
|
|
row_0 = recommendations.iloc[0, 1:] |
|
|
|
|
|
|
|
|
max_votes = row_0.max() |
|
|
majority_voting_result = row_0[row_0 == max_votes].index.tolist() |
|
|
|
|
|
return majority_voting_result[0], max_votes |
|
|
|
|
|
|
|
|
if __name__ == "__main__": |
|
|
print(YFinanceUtils.get_stock_data("AAPL", "2021-01-01", "2021-12-31")) |
|
|
|
|
|
|
