Spaces:
Running
Running
| import yfinance as yf | |
| import pandas as pd | |
| from typing import Dict, Optional, Any, List | |
| from config import NIFTY50_COMPANIES, MARKET_INDICES, SECTOR_ETFS | |
| from models.market_data import StockDataResponse, IndexData, SectorPerformance | |
| class IndianMarketDataFetcher: | |
| """Fetch real-time Indian market data using yfinance""" | |
| def get_nifty50_companies(self) -> Dict[str, str]: | |
| """Get Nifty 50 company list""" | |
| return NIFTY50_COMPANIES | |
| def get_market_indices(self) -> Dict[str, IndexData]: | |
| """Fetch Indian market indices data""" | |
| indices_data = {} | |
| for symbol, name in MARKET_INDICES.items(): | |
| try: | |
| ticker = yf.Ticker(symbol) | |
| hist = ticker.history(period="5d") | |
| if not hist.empty: | |
| current_price = hist['Close'].iloc[-1] | |
| previous_close = hist['Close'].iloc[-2] if len(hist) > 1 else current_price | |
| change = current_price - previous_close | |
| change_pct = (change / previous_close) * 100 | |
| indices_data[name] = IndexData( | |
| current_price=current_price, | |
| change=change, | |
| change_pct=change_pct, | |
| volume=hist['Volume'].iloc[-1] if 'Volume' in hist else 0 | |
| ) | |
| except Exception as e: | |
| print(f"Could not fetch data for {name}: {e}") | |
| return indices_data | |
| def _process_financial_df(self, df: pd.DataFrame, date_col_name: str = 'Date') -> List[Dict[str, Any]]: | |
| """ | |
| Helper to process yfinance financial DataFrames (financials, balance_sheet, cashflow). | |
| It transposes the DataFrame, resets the index, and converts the date column | |
| to string format, then returns a list of dictionaries. | |
| """ | |
| if df.empty: | |
| return [] | |
| # Transpose the DataFrame so dates become the index, and metrics become columns | |
| df_transposed = df.T | |
| # Reset the index to turn the date index into a regular column, | |
| # then convert that column to string format. | |
| df_processed = df_transposed.reset_index() | |
| df_processed.rename(columns={'index': date_col_name}, inplace=True) | |
| # Ensure the date column is explicitly converted to string type (YYYY-MM-DD) | |
| if pd.api.types.is_datetime64_any_dtype(df_processed[date_col_name]): | |
| df_processed[date_col_name] = df_processed[date_col_name].dt.strftime('%Y-%m-%d') | |
| # Fill any remaining NaN values with 0 to ensure Pydantic doesn't complain about None | |
| df_processed.fillna(0, inplace=True) | |
| return df_processed.to_dict('records') | |
| def get_stock_data(self, symbol: str="RELIANCE", period: str = "1y") -> Optional[StockDataResponse]: | |
| """Fetch stock data for analysis""" | |
| if not symbol.endswith(".NS"): | |
| symbol = f"{symbol}.NS" | |
| try: | |
| ticker = yf.Ticker(symbol) | |
| # Get historical data | |
| hist = ticker.history(period=period) | |
| if not hist.empty: | |
| # Reset index to make 'Date' a column, then convert to string | |
| hist_processed = hist.reset_index() | |
| hist_processed['Date'] = hist_processed['Date'].dt.strftime('%Y-%m-%d') | |
| hist_dict = hist_processed.to_dict(orient='list') | |
| else: | |
| hist_dict = {} | |
| # Get company info | |
| info = ticker.info | |
| # Process financials, balance sheet, and cash flow using the helper function | |
| financials_list = self._process_financial_df(ticker.financials) | |
| balance_sheet_list = self._process_financial_df(ticker.balance_sheet) | |
| cash_flow_list = self._process_financial_df(ticker.cashflow) | |
| return StockDataResponse( | |
| history=hist_dict, | |
| info=info, | |
| financials=financials_list, | |
| balance_sheet=balance_sheet_list, | |
| cash_flow=cash_flow_list | |
| ) | |
| except Exception as e: | |
| print(f"Error fetching data for {symbol}: {e}") | |
| return None | |
| def get_sector_performance(self) -> Dict[str, SectorPerformance]: | |
| """Get sector-wise performance""" | |
| sector_data = {} | |
| for etf, sector in SECTOR_ETFS.items(): | |
| try: | |
| ticker = yf.Ticker(etf) | |
| hist = ticker.history(period="5d") | |
| if not hist.empty: | |
| current = hist['Close'].iloc[-1] | |
| prev = hist['Close'].iloc[-2] if len(hist) > 1 else current | |
| change_pct = ((current - prev) / prev) * 100 | |
| sector_data[sector] = SectorPerformance(sector=sector, change_pct=change_pct) | |
| except Exception as e: | |
| print(f"Could not fetch sector data for {etf} ({sector}): {e}") | |
| return sector_data | |
| def get_company_name(self, symbol: str) -> Optional[str]: | |
| """Get company name from symbol""" | |
| return NIFTY50_COMPANIES.get(symbol) | |