| import requests | |
| import pandas as pd | |
| import datetime | |
| import pytz | |
| import numpy as np | |
| import math | |
| import ta | |
| class StockDataFetcher: | |
| def __init__(self): | |
| self.base_url = "https://groww.in/v1/api/charting_service/v3/chart/exchange/NSE/segment/CASH/" | |
| self.base_fno_url = "https://groww.in/v1/api/stocks_fo_data/v3/charting_service/chart/exchange/NSE/segment/FNO/" | |
| self.latest_stock_price = "https://groww.in/v1/api/stocks_data/v1/tr_live_prices/exchange/NSE/segment/CASH/" | |
| self.latest_option_price = "https://groww.in/v1/api/stocks_fo_data/v1/tr_live_prices/exchange/NSE/segment/FNO/" | |
| self.option_chain = "https://groww.in/v1/api/option_chain_service/v1/option_chain/derivatives/" | |
| self.search_url = "https://groww.in/v1/api/search/v1/entity" | |
| self.news_url = "https://groww.in/v1/api/stocks_company_master/v1/company_news/groww_contract_id/" | |
| self.all_stocks_url = "https://groww.in/v1/api/stocks_data/v1/all_stocks" | |
| self.indian_timezone = pytz.timezone('Asia/Kolkata') | |
| self.utc_timezone = pytz.timezone('UTC') | |
| self.headers = { | |
| 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36 Edg/119.0.0.0' | |
| } | |
| def _get_time_range(self, days=7): | |
| current_time = datetime.datetime.now(self.indian_timezone) | |
| start_time = current_time - datetime.timedelta(days=days) | |
| start_time_utc = start_time.astimezone(pytz.utc) | |
| current_time_utc = current_time.astimezone(pytz.utc) | |
| start_time_millis = int(start_time_utc.timestamp() * 1000) | |
| end_time_millis = int(current_time_utc.timestamp() * 1000) | |
| return start_time_millis, end_time_millis | |
| def fetch_stock_data(self, symbol, interval=15, days=7): | |
| start_time, end_time = self._get_time_range(days) | |
| params = { | |
| 'endTimeInMillis': end_time, | |
| 'intervalInMinutes': interval, | |
| 'startTimeInMillis': start_time, | |
| } | |
| try: | |
| print("Downloading data of", symbol.upper()) | |
| if symbol[-2:].upper() == "PE" or symbol[-2:].upper() == "CE" or symbol[-3:].upper() == "FUT": | |
| response = requests.get(self.base_fno_url + symbol.upper(), params=params, headers=self.headers) | |
| else: | |
| response = requests.get(self.base_url + symbol.upper(), params=params, headers=self.headers) | |
| response.raise_for_status() | |
| data = response.json() | |
| columns = ['Date', 'Open', 'High', 'Low', 'Close', 'Volume'] | |
| for row in data['candles']: | |
| row[0] = datetime.datetime.utcfromtimestamp(row[0]) | |
| df = pd.DataFrame(data['candles'], columns=columns) | |
| df['Date'] = pd.to_datetime(df['Date']) | |
| df['Date'] = df['Date'].dt.tz_localize(self.utc_timezone).dt.tz_convert(self.indian_timezone) | |
| return df | |
| except requests.exceptions.RequestException as e: | |
| print(f"Error during API request: {e}") | |
| return None | |
| def fetch_latest_price(self, symbol): | |
| try: | |
| if symbol[-2:].upper() == "PE" or symbol[-2:].upper() == "CE" or symbol[-3:].upper() == "FUT": | |
| response = requests.get(self.latest_option_price + symbol.upper() + "/latest", headers=self.headers) | |
| else: | |
| response = requests.get(self.latest_stock_price + symbol.upper() + "/latest", headers=self.headers) | |
| if response.status_code == 200: | |
| data = response.json() | |
| latest_price = data.get('ltp') | |
| print(symbol, 'Price: ', latest_price) | |
| return latest_price | |
| else: | |
| print(f"Failed to fetch data. Status code: {response.status_code}") | |
| return None | |
| except Exception as e: | |
| print(f"An error occurred: {e}") | |
| return None | |
| def fetch_option_chain(self, symbol): | |
| response = requests.get(self.option_chain + symbol, headers=self.headers) | |
| data = response.json()['optionChain']['optionChains'] | |
| ltp = response.json()['livePrice']['value'] | |
| chain = [] | |
| for i in range(len(data)): | |
| chain.append({"Symbol_CE": data[i]["callOption"]['growwContractId'], "OI_CALL": data[i]["callOption"]['openInterest'] , "CALL": data[i]["callOption"]['ltp'], "strikePrice": data[i]['strikePrice']/100, "PUT": data[i]["putOption"]['ltp'], "OI_PUT": data[i]["putOption"]['openInterest'], "Symbol_PE": data[i]["putOption"]['growwContractId']} | |
| ) | |
| chain = pd.DataFrame(chain) | |
| index = chain[(chain['strikePrice'] >= ltp)].head(1).index[0] | |
| print(response.json()['livePrice']) | |
| chain = chain[index-6:index+7].reset_index(drop=True) | |
| optin_exp = chain['Symbol_CE'][0][:-7] | |
| return chain, optin_exp | |
| def search_entity(self, symbol, entity=None, page=0, size=1, app=False): | |
| params = { | |
| 'app': app, | |
| 'entity_type': entity, | |
| 'page': page, | |
| 'q': f"{symbol}", | |
| 'size': size | |
| } | |
| try: | |
| response = requests.get(self.search_url, params=params, headers=self.headers) | |
| response.raise_for_status() | |
| data = response.json() | |
| entity = data['content'][0] | |
| return {"ID": entity['id'], "title": entity['title'], "NSE_Symbol": entity['nse_scrip_code'], "contract_id" : entity["groww_contract_id"]} | |
| except requests.exceptions.RequestException as e: | |
| print(f"Error during API request: {e}") | |
| return None | |
| def fetch_stock_news(self, symbol, page=1, size=1): | |
| params = { | |
| "page" : page, | |
| "size" : size | |
| } | |
| try: | |
| symbol_id = self.search_entity(symbol.upper())['contract_id'] | |
| response = requests.get(self.news_url + symbol_id, headers=self.headers, params=params).json()['results'] | |
| print(response) | |
| news = [] | |
| for i in range(len(response)): | |
| Title = response[i]['title'] | |
| Summary = response[i]['summary'] | |
| Url = response[i]['url'] | |
| Date = response[i]['pubDate'] | |
| Source = response[i]['source'] | |
| CompanyName = response[i]['companies'][0]['companyName'] | |
| ScripCode = response[i]['companies'][0]['nseScripCode'] | |
| BlogUrl = response[i]['companies'][0]['blogUrl'] | |
| Topics = response[i]['topics'][0] | |
| news.append({ | |
| 'title': Title, | |
| 'summary': Summary, | |
| 'url': Url, | |
| 'pubDate': Date, | |
| 'source': Source, | |
| 'companyName': CompanyName, | |
| 'symbol': ScripCode, | |
| 'blogUrl': BlogUrl, | |
| 'topics': Topics | |
| }) | |
| news_table = pd.DataFrame(news) | |
| return news_table | |
| except: | |
| print("Something went wrong") | |
| return None | |
| def fetch_all_stock(self): | |
| try: | |
| params = { | |
| 'listFilters': {'INDUSTRY': [], 'INDEX': []}, | |
| 'INDEX': ["BSE 100", "Nifty 100", "Nifty Bank", "Nifty Next 50", "Nifty Midcap 100", "SENSEX", "Nifty 50"], | |
| 'INDUSTRY': [], | |
| 'objFilters': {'CLOSE_PRICE': {'max': 100000, 'min': 0}, 'MARKET_CAP': {'min': 0, 'max': 2000000000000000}}, | |
| 'CLOSE_PRICE': {'max': 100000, 'min': 0}, | |
| 'MARKET_CAP': {'min': 0, 'max': 2000000000000000}, | |
| 'size': "1000", | |
| 'sortBy': "NA", | |
| 'sortType': "ASC" | |
| } | |
| all_data = [] | |
| page = 0 | |
| while True: | |
| params['page'] = str(page) | |
| response = requests.post(self.all_stocks_url, headers=self.headers, json=params) | |
| data = response.json() | |
| records = data.get('records', []) | |
| if not records: | |
| break | |
| all_data.extend(records) | |
| page += 1 | |
| df = pd.DataFrame(all_data) | |
| live_price_df = pd.json_normalize(df['livePriceDto']) | |
| df = pd.concat([df, live_price_df], axis=1) | |
| df = df.drop(columns=['livePriceDto']) | |
| return df | |
| except: | |
| return None | |
| def realtime_signal(self, symbol, intervals=15, days=10): | |
| rounding_value=None | |
| if symbol.upper() == "NIFTY": | |
| index_symbol = "NIFTY" | |
| rounding_value = 50 | |
| elif symbol.upper() == "NIFTY-BANK": | |
| index_symbol = "BANKNIFTY" | |
| rounding_value = 100 | |
| else: | |
| pass | |
| stock_data = self.fetch_stock_data(index_symbol, intervals, days) | |
| chain, exp = self.fetch_option_chain(symbol.upper()) | |
| stock_data['RSI'] = ta.momentum.rsi(stock_data['Close'], window=14) | |
| stock_data = stock_data.drop(columns=['Volume']) | |
| stock_data['Prev_RSI'] = stock_data['RSI'].shift(1) | |
| stock_data['Signal'] = 0 | |
| call_condition = (stock_data['RSI'] > 60) & (stock_data['Prev_RSI'] < 60) | |
| put_condition = (stock_data['RSI'] < 40) & (stock_data['Prev_RSI'] > 40) | |
| stock_data.loc[call_condition, 'Signal'] = 1 | |
| stock_data.loc[put_condition, 'Signal'] = 2 | |
| stock_data = stock_data.dropna().reset_index(drop=True) | |
| def floor_to_nearest(value, nearest): | |
| return math.ceil(value / nearest) * nearest | |
| stock_data['Option'] = stock_data['Close'].apply(lambda x: floor_to_nearest(x, rounding_value)) | |
| stock_data['direction'] = np.where(stock_data['Signal'] == 2, "PE", np.where(stock_data['Signal'] == 1, "CE", "")) | |
| stock_data['symbol'] = exp + stock_data['Option'].astype(str) + stock_data['direction'] | |
| return stock_data |