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_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