utils.py

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_index_price = "https://groww.in/v1/api/stocks_data/v1/tr_live_indices/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.nearest_expiries = "https://groww.in/v1/api/stocks_fo_data/v1/nearest_expiries?instrumentType=INDEX"

        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)
        elif symbol.upper() == "NIFTY" or symbol.upper() == "BANKNIFTY":
          response = requests.get(self.latest_index_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()
            if 'ltp' in data:
              latest_price = data['ltp']
            else:
              latest_price = data['value']
            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_nearest_expiries(self):
      try:
          response = requests.get(self.nearest_expiries, headers=self.headers).json()
          return response
      except Exception as e:
          print(f"An error occurred: {e}")
          return None

    def fetch_option_chain(self, symbol):

      if symbol.upper() == "BANKNIFTY":
        symbol = "nifty-bank"

      response = requests.get(self.option_chain + symbol, headers=self.headers)
      data = response.json()['optionChain']['optionChains']
      ltp = response.json()['livePrice']['value']

      if symbol == "nifty-bank":
        index_ltp = requests.get(self.latest_index_price + "BANKNIFTY" + "/latest", headers=self.headers).json()['value']

      if symbol.upper() == "NIFTY":
        index_ltp = requests.get(self.latest_index_price + "NIFTY" + "/latest", headers=self.headers).json()['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]
      chain = chain[index-6:index+7].reset_index(drop=True)
      optin_exp = chain['Symbol_CE'][0][:-7]
      return chain, optin_exp, index_ltp

    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:
          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']
        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:
        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
      signal_status = None

      if symbol.upper() == "NIFTY":
        index_symbol = "NIFTY"
        rounding_value = 50

      elif symbol.upper() == "BANKNIFTY":
        index_symbol = "BANKNIFTY"
        rounding_value = 100

      else:
        pass

      stock_data = self.fetch_stock_data(index_symbol, intervals, days)
      chain, exp, index_ltp = 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']
      stock_data = stock_data.tail(1).to_dict(orient="records")[0]
      if stock_data['direction'] == "PE" or stock_data['direction'] == "CE":
        signal_status = True
        signal_data = self.fetch_stock_data(stock_data['symbol'], intervals, 1).tail(1).to_dict(orient="records")[0]
        print(signal_data)
        entry_price = signal_data['High']
        stoploss = signal_data['Low']
        response = {"signal_status" : signal_status, "strike" : stock_data['Option'], "option_type": stock_data['direction'], "entry_price" : entry_price, "stoploss" : stoploss,  "index_candle": stock_data, "symbol_candle": signal_data}
      else:
        signal_status = False
        response = {"signal_status" : "No signals"}
      return response