import gradio as gr import requests import json from datetime import datetime import pytz def validate_date_format(date_str): """Validate if the date is in dd-mm-yyyy format""" try: datetime.strptime(date_str, "%d-%m-%Y") return True except ValueError: return False def convert_to_unix_millis(date_str, time_str): """Convert dd-mm-yyyy date and time to Unix timestamp in milliseconds (IST)""" ist = pytz.timezone('Asia/Kolkata') datetime_str = f"{date_str} {time_str}" dt = datetime.strptime(datetime_str, "%d-%m-%Y %H:%M") dt_ist = ist.localize(dt) return int(dt_ist.timestamp() * 1000) def epoch_to_ist_time(epoch_seconds): """Convert epoch timestamp (seconds) to IST time string (HH:MM:SS)""" ist = pytz.timezone('Asia/Kolkata') dt = datetime.fromtimestamp(epoch_seconds, tz=pytz.UTC) dt_ist = dt.astimezone(ist) return dt_ist.strftime("%H:%M:%S") def fetch_stock_data(ticker, date): """Fetch stock data from Groww API and return simplified JSON""" debug_log = [] # Validate date format debug_log.append(f"[DEBUG] Input - Ticker: {ticker}, Date: {date}") if not validate_date_format(date): return "Error: Please enter date in dd-mm-yyyy format" # Validate ticker if not ticker or ticker.strip() == "": return "Error: Please enter a valid ticker symbol" ticker = ticker.strip().upper() debug_log.append(f"[DEBUG] Cleaned Ticker: {ticker}") try: # Calculate start and end timestamps start_millis = convert_to_unix_millis(date, "00:05") end_millis = convert_to_unix_millis(date, "23:55") debug_log.append(f"[DEBUG] Start Time (IST 00:05): {start_millis} ms") debug_log.append(f"[DEBUG] End Time (IST 23:55): {end_millis} ms") # Construct API URL url = f"https://groww.in/v1/api/charting_service/v4/chart/exchange/NSE/segment/CASH/{ticker}" params = { "endTimeInMillis": end_millis, "intervalInMinutes": 1, "startTimeInMillis": start_millis } # Full URL with parameters full_url = f"{url}?endTimeInMillis={end_millis}&intervalInMinutes=1&startTimeInMillis={start_millis}" debug_log.append(f"[DEBUG] Full API URL: {full_url}") # Make API request debug_log.append("[DEBUG] Making API request...") response = requests.get(url, params=params, timeout=10) debug_log.append(f"[DEBUG] Response Status Code: {response.status_code}") debug_log.append(f"[DEBUG] Response Headers: {dict(response.headers)}") # Check if request was successful if response.status_code != 200: debug_log.append(f"[DEBUG] Response Body: {response.text[:500]}") return "\n".join(debug_log) + "\n\n[ERROR] Invalid ticker or no data available for the given date." debug_log.append(f"[DEBUG] Response Content Length: {len(response.text)} bytes") data = response.json() debug_log.append(f"[DEBUG] Response JSON Keys: {list(data.keys())}") # Check if candles data exists if "candles" not in data or not data["candles"]: debug_log.append("[DEBUG] No 'candles' key found or candles array is empty") debug_log.append(f"[DEBUG] Full Response: {json.dumps(data, indent=2)[:1000]}") return "\n".join(debug_log) + "\n\n[ERROR] Invalid ticker or no data available for the given date." debug_log.append(f"[DEBUG] Number of candles received: {len(data['candles'])}") debug_log.append(f"[DEBUG] First candle sample: {data['candles'][0]}") debug_log.append(f"[DEBUG] Last candle sample: {data['candles'][-1]}") # Process candles data simplified_data = [] for candle in data["candles"]: epoch_seconds = candle[0] opening_price = candle[1] time_ist = epoch_to_ist_time(epoch_seconds) simplified_data.append({ "time": time_ist, "open": opening_price }) debug_log.append(f"[DEBUG] Successfully processed {len(simplified_data)} candles") debug_log.append("\n" + "="*50 + "\n[SUCCESS] DATA OUTPUT:\n" + "="*50 + "\n") # Return debug log + JSON string return "\n".join(debug_log) + "\n" + json.dumps(simplified_data, indent=2) except requests.exceptions.Timeout: debug_log.append("[DEBUG] Request timed out") return "\n".join(debug_log) + "\n\n[ERROR] Request timed out. Please try again." except requests.exceptions.RequestException as e: debug_log.append(f"[DEBUG] Request Exception: {type(e).__name__} - {str(e)}") return "\n".join(debug_log) + "\n\n[ERROR] Invalid ticker or no data available for the given date." except Exception as e: debug_log.append(f"[DEBUG] Unexpected Exception: {type(e).__name__} - {str(e)}") import traceback debug_log.append(f"[DEBUG] Traceback:\n{traceback.format_exc()}") return "\n".join(debug_log) + "\n\n[ERROR] An unexpected error occurred" # Create Gradio interface with gr.Blocks(title="Groww Stock Data Fetcher") as app: gr.Markdown("# Groww Stock Data Fetcher") gr.Markdown("Fetch intraday 1-minute candle data for NSE stocks") with gr.Row(): with gr.Column(): ticker_input = gr.Textbox( label="Ticker / Symbol", placeholder="e.g., RELIANCE, TCS, HDFCBANK", lines=1 ) date_input = gr.Textbox( label="Date (dd-mm-yyyy)", placeholder="e.g., 05-11-2025", lines=1 ) submit_btn = gr.Button("Submit", variant="primary") with gr.Row(): output = gr.Textbox( label="Output", lines=20, max_lines=30, show_copy_button=True ) submit_btn.click( fn=fetch_stock_data, inputs=[ticker_input, date_input], outputs=output ) gr.Markdown("### Instructions") gr.Markdown(""" 1. Enter a valid NSE stock ticker (e.g., RELIANCE, TCS, HDFCBANK) 2. Enter the date in **dd-mm-yyyy** format (e.g., 05-11-2025) 3. Click Submit to fetch the data 4. The output will show timestamp (IST) and opening price for each 1-minute candle """) # Launch the app if __name__ == "__main__": app.launch()