Spaces:
Running
Running
| 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.""" | |
| def wrapper(symbol: Annotated[str, "ticker symbol"], *args, **kwargs) -> Any: | |
| ticker = yf.Ticker(symbol) | |
| return func(ticker, *args, **kwargs) | |
| return wrapper | |
| 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 # No recommendations available | |
| # Assuming 'period' column exists and needs to be excluded | |
| row_0 = recommendations.iloc[0, 1:] # Exclude 'period' column if necessary | |
| # Find the maximum voting result | |
| 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")) | |
| # print(YFinanceUtils.get_stock_data()) | |