Update app.py

app.py

@@ -1,313 +1,138 @@
 import gradio as gr
-import re
 import asyncio
-import aiohttp
-from datetime import datetime, time, date
-from typing import Dict, Optional, List
-import pytz
-import tzlocal
-import logging
 import json
-import ast
-
-# --- Basic Setup ---
-
-# 1. Logger Configuration
-logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
-logger = logging.getLogger(__name__)
-
-# 2. Custom Exception
-class DateTimeValidationError(ValueError):
-    """Custom exception for datetime validation errors."""
-    pass
-
-# 3. API URL
-hist_url = "https://groww.in/v1/api/charting_service/v2/chart/exchange/NSE/segment/CASH"
-
-
-# --- Date/Time Functions ---
-
-def validate_datetime_format(dt_str: str) -> datetime:
-    """Validate datetime string in strict 'YYYY-MM-DD' format."""
-    if not isinstance(dt_str, str):
-        raise TypeError(f"Input must be a string, got {type(dt_str).__name__}")
-    date_pattern = re.compile(r'^\d{4}-\d{2}-\d{2}$')
-    if not date_pattern.match(dt_str):
-        raise DateTimeValidationError(f"Invalid date format: '{dt_str}'. Expected 'YYYY-MM-DD'")
-    try:
-        return datetime.strptime(dt_str, '%Y-%m-%d')
-    except ValueError as e:
-        raise DateTimeValidationError(f"Invalid date value: '{dt_str}'.") from e
-
-def _resolve_timezone(timezone: Optional[str]) -> pytz.BaseTzInfo:
-    """Resolve timezone string to pytz timezone object."""
-    if timezone:
-        try:
-            return pytz.timezone(timezone)
-        except pytz.UnknownTimeZoneError as e:
-            raise ValueError(f"Unknown timezone: '{timezone}'") from e
-    return tzlocal.get_localzone()
-
-def convert_to_unixtimestamp(date_time_str: str, timezone: Optional[str] = None) -> int:
-    """Convert 'YYYY-MM-DD HH:MM' string to Unix timestamp in milliseconds."""
-    if not isinstance(date_time_str, str):
-        raise TypeError(f"DateTime input must be a string, got {type(date_time_str).__name__}")
-    try:
-        dt = datetime.strptime(date_time_str, '%Y-%m-%d %H:%M')
-    except ValueError as e:
-        raise DateTimeValidationError(f"Invalid format: '{date_time_str}'. Expected 'YYYY-MM-DD HH:MM'") from e
-    
-    target_tz = _resolve_timezone(timezone)
-    localized_dt = target_tz.localize(dt) if dt.tzinfo is None else dt.astimezone(target_tz)
-    return int(localized_dt.timestamp() * 1000)
-
-def get_time_range_in_unix_ms(start_date_str: str, end_date_str: str, timezone: str = 'Asia/Kolkata') -> Dict[str, int]:
-    """Calculates the start (00:00) and end (23:59) Unix timestamps for a date range."""
-    start_date = validate_datetime_format(start_date_str)
-    end_date = validate_datetime_format(end_date_str)
-    start_datetime = datetime.combine(start_date, time.min)
-    end_datetime = datetime.combine(end_date, time(23, 59))
-    start_timestamp = convert_to_unixtimestamp(start_datetime.strftime('%Y-%m-%d %H:%M'), timezone)
-    end_timestamp = convert_to_unixtimestamp(end_datetime.strftime('%Y-%m-%d %H:%M'), timezone)
-    return {"start_timestamp_ms": start_timestamp, "end_timestamp_ms": end_timestamp}
-
+from typing import List, Dict
 
-# --- OPTIMIZED API Functions - Interval-Based Batching ---
+import asyncio
+import aiohttp
+from typing import Dict, List
+
+# Assume 'hist_url' is globally defined
+# For example:
+hist_url="https://groww.in/v1/api/charting_service/v2/chart/exchange/NSE/segment/CASH"
+
+
+async def call_price_api_async(
+    session: aiohttp.ClientSession,
+    ticker: str,
+    start: int,
+    end: int,
+    interval: int,
+    debug: bool = False
+) -> Dict:
+    """
+    Asynchronously calls the Groww candle API and returns the raw JSON response.
+
+    Parameters:
+    - session (aiohttp.ClientSession): An aiohttp client session.
+    - ticker (str): NSE stock symbol or Groww-compliant identifier.
+    - start (int): Start time in epoch milliseconds (UTC).
+    - end (int): End time in epoch milliseconds (UTC).
+    - interval (int): Candle interval in minutes.
+    - debug (bool): If True, includes detailed debug logs.
+
+    Returns:
+    - dict: A dictionary with "data", "debug_info", and "error" keys.
+    """
+    debug_logs = []
+    if debug:
+        debug_logs.append(f"[DEBUG] Ticker: {ticker} - Starting API call.")
 
-async def fetch_single_ticker_interval(session: aiohttp.ClientSession, ticker: str, start: int, end: int, interval: int, max_retries: int = 2) -> Dict:
-    """Fetch data for a single ticker-interval combination with retry logic."""
     url = f"{hist_url}/{ticker}"
-    params = {"startTimeInMillis": start, "endTimeInMillis": end, "intervalInMinutes": interval}
-    
-    for attempt in range(max_retries + 1):
-        try:
-            async with session.get(url, params=params) as response:
-                response.raise_for_status()
-                json_data = await response.json()
-                return {"ticker": ticker, "interval": interval, "data": json_data, "error": None}
-        except asyncio.TimeoutError:
-            if attempt < max_retries:
-                logger.warning(f"Timeout for {ticker} ({interval}m) - Retry {attempt + 1}/{max_retries}")
-                await asyncio.sleep(0.5 * (attempt + 1))  # Exponential backoff
-                continue
-            return {"ticker": ticker, "interval": interval, "data": None, "error": f"Timeout after {max_retries + 1} attempts"}
-        except Exception as e:
-            if attempt < max_retries and "timeout" in str(e).lower():
-                logger.warning(f"Error for {ticker} ({interval}m) - Retry {attempt + 1}/{max_retries}: {e}")
-                await asyncio.sleep(0.5 * (attempt + 1))
-                continue
-            return {"ticker": ticker, "interval": interval, "data": None, "error": str(e)}
-    
-    return {"ticker": ticker, "interval": interval, "data": None, "error": "Max retries exceeded"}
+    params = {
+        "startTimeInMillis": start,
+        "endTimeInMillis": end,
+        "intervalInMinutes": interval,
+    }
 
-async def fetch_interval_batch(session: aiohttp.ClientSession, tickers: List[str], start_time: int, end_time: int, interval: int, batch_size: int = 30) -> List[Dict]:
-    """Fetch data for all tickers for a specific interval in batches."""
-    results = []
-    
-    for i in range(0, len(tickers), batch_size):
-        batch_tickers = tickers[i:i+batch_size]
-        logger.info(f"Processing {interval}m batch {i//batch_size + 1}: {len(batch_tickers)} tickers")
-        
-        batch_start = datetime.now()
-        
-        # Create tasks for this batch with retry logic
+    try:
+        async with session.get(url, params=params) as response:
+            if debug:
+                debug_logs.append(f"[DEBUG] Ticker: {ticker} - Status Code: {response.status}")
+
+            response.raise_for_status()
+            json_data = await response.json()
+
+            if debug:
+                debug_logs.append(f"[DEBUG] Ticker: {ticker} - API call successful.")
+
+            return {
+                "ticker": ticker,
+                "data": json_data,
+                "debug_info": debug_logs if debug else None,
+                "error": None,
+            }
+    except aiohttp.ClientError as e:
+        if debug:
+            debug_logs.append(f"[DEBUG] Ticker: {ticker} - Request failed: {e}")
+
+        return {
+            "ticker": ticker,
+            "data": None,
+            "debug_info": debug_logs if debug else None,
+            "error": [str(e)],
+        }
+
+
+async def main(tickers: List[str], start_time: int, end_time: int, interval: int) -> List[Dict]:
+    """
+    Main function to run asynchronous API calls for multiple tickers.
+    """
+    async with aiohttp.ClientSession() as session:
         tasks = [
-            fetch_single_ticker_interval(session, ticker, start_time, end_time, interval, max_retries=2)
-            for ticker in batch_tickers
+            call_price_api_async(session, ticker, start_time, end_time, interval, debug=True)
+            for ticker in tickers
         ]
-        
-        # Execute batch concurrently
-        batch_results = await asyncio.gather(*tasks, return_exceptions=True)
-        
-        # Handle any remaining exceptions
-        for result in batch_results:
-            if isinstance(result, Exception):
-                logger.error(f"Unhandled exception in batch: {result}")
-            else:
-                results.append(result)
-        
-        batch_duration = (datetime.now() - batch_start).total_seconds()
-        logger.info(f"Batch {i//batch_size + 1} completed in {batch_duration:.2f}s")
-        
-        # Longer delay between batches to reduce API load
-        if i + batch_size < len(tickers):
-            await asyncio.sleep(0.2)
-    
-    return results
+        results = await asyncio.gather(*tasks)
+        return results
 
-async def fetch_all_data_by_intervals(tickers: List[str], start_time: int, end_time: int, intervals: List[int]) -> List[Dict]:
-    """Fetch data by processing each interval separately in batches of 30."""
-    all_results = []
-    
-    # INCREASED timeout limits and better connection management
-    connector = aiohttp.TCPConnector(
-        limit=40,           # Reduced concurrent connections
-        limit_per_host=25,  # Reduced per-host connections
-        keepalive_timeout=60,
-        enable_cleanup_closed=True
-    )
-    
-    # MUCH LONGER timeouts
-    timeout = aiohttp.ClientTimeout(
-        total=60,           # Increased from 20 to 60 seconds
-        connect=15,         # Increased from 5 to 15 seconds
-        sock_read=45        # Added socket read timeout
-    )
-    
-    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
-        
-        # Process each interval separately
-        for interval in intervals:
-            logger.info(f"Starting {interval}m interval requests for {len(tickers)} tickers...")
-            interval_start = datetime.now()
-            
-            interval_results = await fetch_interval_batch(session, tickers, start_time, end_time, interval, batch_size=25)  # Reduced batch size
-            
-            interval_duration = (datetime.now() - interval_start).total_seconds()
-            successful = len([r for r in interval_results if r.get("error") is None])
-            logger.info(f"Completed {interval}m interval in {interval_duration:.2f}s - {successful}/{len(interval_results)} successful")
-            
-            all_results.extend(interval_results)
-            
-            # Longer pause between intervals to let API recover
-            if interval != intervals[-1]:  # Not the last interval
-                logger.info("Pausing 1 second between intervals...")
-                await asyncio.sleep(1.0)
-    
-    return all_results
 
-def merge_data_fast(results: List[Dict]) -> str:
-    """Fast data merging without unnecessary overhead."""
-    merged_data = {}
-    
-    for result in results:
-        ticker = result["ticker"]
-        interval = result["interval"]
-        data = result["data"]
-        error = result["error"]
-        
-        # Initialize ticker entry if not exists
-        if ticker not in merged_data:
-            merged_data[ticker] = {"symbol": ticker}
-        
-        # Handle errors
-        if error:
-            merged_data[ticker][f"error_{interval}m"] = error
-            continue
-        
-        # Process successful data
-        if data and data.get("candles") and len(data["candles"]) > 0:
-            candle = data["candles"][0]
-            if len(candle) > 4:
-                prefix = "day" if interval == 1440 else "start"
-                merged_data[ticker].update({
-                    f"{prefix} open": candle[1],
-                    f"{prefix} high": candle[2],
-                    f"{prefix} low": candle[3],
-                    f"{prefix} close": candle[4]
-                })
-        else:
-            merged_data[ticker][f"error_{interval}m"] = "No candles data"
-    
-    return json.dumps(list(merged_data.values()), indent=2)
 
+def run_main(ticker_input: str, start: int, end: int, interval: int) -> str:
+    """
+    Wrapper to run the async main() function in sync for Gradio.
+    """
+    try:
+        # Convert input string like ["TCS","HDFCBANK"] into list
+        tickers: List[str] = json.loads(ticker_input)
 
-# --- MAIN Gradio Backend Function ---
+        # Run the async loop
+        results: List[Dict] = asyncio.run(main(tickers, start, end, interval))
 
-async def run_backend_processing(tickers_text: str, progress=gr.Progress(track_tqdm=True)):
-    """Main processing function - optimized for speed."""
-    
-    # Input validation
-    if not tickers_text.strip():
-        yield "❌ Error: Input is empty", "{}"
-        return
+        # Pretty-print JSON
+        return json.dumps(results, indent=2)
 
-    try:
-        parsed_input = ast.literal_eval(tickers_text)
-        if not isinstance(parsed_input, list):
-            raise TypeError("Must be a list")
-        tickers = [str(item).strip().upper() for item in parsed_input if str(item).strip()]
-        if not tickers:
-            yield "❌ Error: No valid tickers found", "{}"
-            return
-    except:
-        yield '❌ Invalid format. Use: ["TICKER1", "TICKER2"]', "{}"
-        return
-
-    # Setup
-    today_str = date.today().strftime("%Y-%m-%d")
-    time_range = get_time_range_in_unix_ms(today_str, today_str)
-    start_time, end_time = time_range["start_timestamp_ms"], time_range["end_timestamp_ms"]
-    intervals = [15, 1440]  # Process 15m first, then 1440m
-    
-    total_requests = len(tickers) * len(intervals)
-    
-    try:
-        overall_start = datetime.now()
-        
-        # Update progress and show timing
-        yield f"🚀 Starting: {len(tickers)} tickers × {len(intervals)} intervals = {total_requests} requests", "{}"
-        
-        # Fetch all data using interval-based batching with improved timeouts
-        results = await fetch_all_data_by_intervals(tickers, start_time, end_time, intervals)
-        
-        # Quick merge
-        final_json = merge_data_fast(results)
-        
-        total_time = (datetime.now() - overall_start).total_seconds()
-        successful = len([r for r in results if r.get("error") is None])
-        failed = total_requests - successful
-        
-        status = "✅ Complete!" if failed == 0 else f"⚠️ Complete with {failed} timeouts"
-        yield f"{status} {successful}/{total_requests} successful in {total_time:.2f}s", final_json
-        
     except Exception as e:
-        logger.error(f"Processing error: {e}")
-        yield f"❌ Error: {str(e)}", "{}"
+        return f"Error: {str(e)}"
 
 
-# --- Gradio UI ---
+with gr.Blocks() as demo:
+    gr.Markdown("## 📈 Groww Async Candle API Fetcher")
 
-with gr.Blocks(theme=gr.themes.Soft(), title="Stock Data Processor") as demo:
-    gr.Markdown("# ⚡ Stock Data Processor - Interval-Optimized")
-    gr.Markdown("**Strategy**: Process 15m requests in batches of 30, then 1440m requests in batches of 30, then merge results.")
-    
     with gr.Row():
-        with gr.Column(scale=1):
-            tickers_input = gr.Textbox(
-                lines=4, 
-                label='📈 Stock Tickers (Python List Format)', 
-                value='["RELIANCE", "INFY", "TCS"]',
-                placeholder='["TICKER1", "TICKER2", "TICKER3", ...]'
-            )
-            start_button = gr.Button("🚀 Process Data", variant="primary", size="lg")
-            
-            with gr.Accordion("ℹ️ Performance Info", open=False):
-                gr.Markdown("""
-                **Optimization Strategy:**
-                1. Process all 15m interval requests first (batches of 30)
-                2. Process all 1440m interval requests next (batches of 30) 
-                3. Merge all results into final JSON
-                
-                **Expected Speed:** ~1-3 seconds for small ticker lists
-                """)
-        
-        with gr.Column(scale=2):
-            logs_output = gr.Textbox(label="📊 Processing Logs", lines=8, interactive=False)
-            json_output = gr.JSON(label="📋 Final Results", show_label=True)
+        ticker_box = gr.Textbox(
+            label="Tickers (JSON list)",
+            placeholder='["TCS","HDFCBANK"]'
+        )
+    with gr.Row():
+        start_box = gr.Number(label="Start (epoch millis)", value=1660707000000)
+        end_box = gr.Number(label="End (epoch millis)", value=1660793400000)
+        interval_box = gr.Number(label="Interval (minutes)", value=15)
 
-    # Event handlers
-    start_button.click(
-        fn=run_backend_processing,
-        inputs=[tickers_input],
-        outputs=[logs_output, json_output]
+    run_button = gr.Button("Fetch Data")
+
+    output_box = gr.Textbox(
+        label="Raw API Output",
+        lines=20,
+        placeholder="Results will appear here..."
     )
-    
-    tickers_input.submit(
-        fn=run_backend_processing,
-        inputs=[tickers_input],
-        outputs=[logs_output, json_output]
+
+    run_button.click(
+        fn=run_main,
+        inputs=[ticker_box, start_box, end_box, interval_box],
+        outputs=output_box
     )
 
+
 if __name__ == "__main__":
-    demo.launch(share=False, server_name="0.0.0.0")
+    demo.launch()