Update app.py

app.py

@@ -8,7 +8,8 @@ import pytz
 import tzlocal
 import logging
 import json
-import ast # Import the ast module for safe string evaluation
+import ast
+from concurrent.futures import ThreadPoolExecutor
 
 # --- Basic Setup ---
 
@@ -25,7 +26,7 @@ class DateTimeValidationError(ValueError):
 hist_url = "https://groww.in/v1/api/charting_service/v2/chart/exchange/NSE/segment/CASH"
 
 
-# --- Date/Time Functions (No changes here) ---
+# --- Date/Time Functions ---
 
 def validate_datetime_format(dt_str: str) -> datetime:
     """Validate datetime string in strict 'YYYY-MM-DD' format."""
@@ -72,10 +73,10 @@ def get_time_range_in_unix_ms(start_date_str: str, end_date_str: str, timezone:
     return {"start_timestamp_ms": start_timestamp, "end_timestamp_ms": end_timestamp}
 
 
-# --- FIXED Asynchronous API & Processing Functions ---
+# --- OPTIMIZED Async Functions ---
 
-async def call_price_api_async(session: aiohttp.ClientSession, ticker: str, start: int, end: int, interval: int) -> Dict:
-    """Asynchronously calls the candle API."""
+async def fetch_single_request(session: aiohttp.ClientSession, ticker: str, start: int, end: int, interval: int) -> Dict:
+    """Fetch data for a single ticker-interval combination."""
     url = f"{hist_url}/{ticker}"
     params = {"startTimeInMillis": start, "endTimeInMillis": end, "intervalInMinutes": interval}
     try:
@@ -83,67 +84,69 @@ async def call_price_api_async(session: aiohttp.ClientSession, ticker: str, star
             response.raise_for_status()
             json_data = await response.json()
             return {"ticker": ticker, "interval": interval, "data": json_data, "error": None}
-    except aiohttp.ClientError as e:
+    except Exception as e:
+        logger.error(f"API call failed for {ticker} ({interval}m): {e}")
         return {"ticker": ticker, "interval": interval, "data": None, "error": str(e)}
 
-async def main_task(tickers: List[str], start_time: int, end_time: int, intervals: List[int], progress: gr.Progress):
-    """OPTIMIZED: Calculate all requests first, then process in batches of 30."""
+async def fetch_all_data_optimized(tickers: List[str], start_time: int, end_time: int, intervals: List[int]) -> List[Dict]:
+    """Fetch all data with maximum concurrency - NO artificial batching."""
     
-    # Step 1: Calculate all requests upfront
-    all_requests = []
+    # Create ALL tasks at once
+    all_tasks = []
     for ticker in tickers:
         for interval in intervals:
-            all_requests.append((ticker, start_time, end_time, interval))
+            all_tasks.append((ticker, start_time, end_time, interval))
     
-    total_requests = len(all_requests)
-    logger.info(f"Total requests to make: {total_requests}")
+    logger.info(f"Creating {len(all_tasks)} concurrent requests...")
     
-    # Step 2: Process requests in batches of 30
-    results = []
-    batch_size = 30
+    # Configure session for high concurrency
+    connector = aiohttp.TCPConnector(
+        limit=100,  # Max concurrent connections
+        limit_per_host=50,  # Max per host
+        keepalive_timeout=30
+    )
     
-    async with aiohttp.ClientSession(
-        timeout=aiohttp.ClientTimeout(total=30),  # Add timeout for better reliability
-        connector=aiohttp.TCPConnector(limit=50)  # Limit concurrent connections
-    ) as session:
-        
-        for i in progress.tqdm(range(0, total_requests, batch_size), desc="Processing Request Batches"):
-            batch_requests = all_requests[i:i+batch_size]
-            
-            # Create tasks for this batch
-            tasks = [
-                call_price_api_async(session, ticker, start, end, interval)
-                for ticker, start, end, interval in batch_requests
-            ]
-            
-            # Execute batch concurrently
-            batch_results = await asyncio.gather(*tasks, return_exceptions=True)
-            
-            # Handle any exceptions in results
-            for result in batch_results:
-                if isinstance(result, Exception):
-                    logger.error(f"Request failed with exception: {result}")
-                    # You can add error handling here if needed
-                else:
-                    results.append(result)
-            
-            # Optional: Add small delay between batches to be respectful to the API
-            if i + batch_size < total_requests:
-                await asyncio.sleep(0.1)
+    timeout = aiohttp.ClientTimeout(total=30, connect=10)
     
-    logger.info(f"Completed {len(results)} requests out of {total_requests}")
-    return results
+    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
+        # Create all coroutines
+        tasks = [
+            fetch_single_request(session, ticker, start, end, interval)
+            for ticker, start, end, interval in all_tasks
+        ]
+        
+        # Execute ALL requests concurrently
+        logger.info("Executing all requests concurrently...")
+        start_time_fetch = datetime.now()
+        
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+        
+        fetch_duration = (datetime.now() - start_time_fetch).total_seconds()
+        logger.info(f"All API calls completed in {fetch_duration:.2f} seconds")
+        
+        # Filter out exceptions
+        valid_results = [r for r in results if not isinstance(r, Exception)]
+        logger.info(f"Got {len(valid_results)} valid responses out of {len(all_tasks)} requests")
+        
+        return valid_results
 
-def process_and_merge_data(results: list) -> str:
-    """Processes and merges the raw API results for different intervals."""
+def process_data_sync(results: List[Dict]) -> str:
+    """Process data synchronously - this is fast."""
     merged_data = {}
+    
     for result in results:
-        ticker, interval, data, error = result.get("ticker"), result.get("interval"), result.get("data"), result.get("error")
+        ticker = result.get("ticker")
+        interval = result.get("interval")
+        data = result.get("data")
+        error = result.get("error")
+        
         if ticker not in merged_data:
             merged_data[ticker] = {"symbol": ticker}
+            
         if error:
             merged_data[ticker][f"error_{interval}m"] = error
             continue
+            
         if data and data.get("candles") and data["candles"]:
             first_candle = data["candles"][0]
             if len(first_candle) > 4:
@@ -154,86 +157,85 @@ def process_and_merge_data(results: list) -> str:
                 merged_data[ticker][f"{prefix} close"] = first_candle[4]
         else:
             merged_data[ticker][f"error_{interval}m"] = "No data or candles found"
-            
+    
     return json.dumps(list(merged_data.values()), indent=4)
 
 
-# --- Gradio Backend Function ---
+# --- STREAMLINED Gradio Backend Function ---
 
 async def run_backend_processing(tickers_text: str, progress=gr.Progress(track_tqdm=True)):
-    """The main function to be called by the Gradio interface."""
+    """Streamlined backend processing with minimal yields."""
     if not tickers_text.strip():
         yield "Error: Input is empty. Please provide a list of tickers.", "{}"
         return
 
     try:
-        # Safely parse the string input into a Python list
         parsed_input = ast.literal_eval(tickers_text)
         if not isinstance(parsed_input, list):
             raise TypeError("Input must be a list.")
-        # Clean and validate tickers
         tickers = [str(item).strip().upper() for item in parsed_input]
         if not tickers:
             yield "Error: The provided list is empty.", "{}"
             return
-
-    except (ValueError, SyntaxError, TypeError) as e:
-        error_message = 'Invalid input format. Please provide a list of strings, e.g., ["RELIANCE", "INFY"]'
-        yield error_message, "{}"
+    except (ValueError, SyntaxError, TypeError):
+        yield 'Invalid input format. Please provide a list of strings, e.g., ["RELIANCE", "INFY"]', "{}"
         return
 
+    # Get timestamps
     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 = [1440, 15]
-    total_requests = len(tickers) * len(intervals)
     
-    yield f"Starting processing for {len(tickers)} tickers: {tickers}\nTotal API requests: {total_requests}", "{}"
+    total_requests = len(tickers) * len(intervals)
+    yield f"Fetching data for {len(tickers)} tickers ({total_requests} total requests)...", "{}"
     
     try:
-        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"]
-
-        yield f"Fetching data from API in batches of 30 (Total: {total_requests} requests)...", "{}"
-        start_fetch_time = datetime.now()
+        # Time the entire operation
+        overall_start = datetime.now()
         
-        results = await main_task(tickers, start_time, end_time, intervals, progress)
+        # Fetch all data with maximum concurrency
+        results = await fetch_all_data_optimized(tickers, start_time, end_time, intervals)
         
-        fetch_duration = (datetime.now() - start_fetch_time).total_seconds()
-        yield f"Data fetching completed in {fetch_duration:.2f} seconds. Processing and merging data...", "{}"
+        # Process data (this is fast)
+        processed_json = process_data_sync(results)
         
-        processed_json = process_and_merge_data(results)
+        total_time = (datetime.now() - overall_start).total_seconds()
         
-        total_duration = (datetime.now() - start_fetch_time).total_seconds()
-        yield f"Processing complete in {total_duration:.2f} seconds. Successfully processed {len(results)} API responses.", processed_json
+        yield f"✅ Complete! Processed {len(results)} responses in {total_time:.2f} seconds", processed_json
         
     except Exception as e:
-        logger.error(f"An unhandled error occurred: {e}")
+        logger.error(f"Error: {e}")
         yield f"An error occurred: {e}", "{}"
 
 
 # --- Gradio UI ---
 
 with gr.Blocks(theme=gr.themes.Soft()) as demo:
-    gr.Markdown("## Stock Data Backend Processor (Optimized)")
-    gr.Markdown("Enter a list of stock tickers in Python list format. The app will make concurrent API requests in batches of 30.")
+    gr.Markdown("## ⚡ High-Performance Stock Data Processor")
+    gr.Markdown("Optimized for maximum concurrency - fetches all data simultaneously!")
     
     with gr.Row():
         with gr.Column(scale=1):
             tickers_input = gr.Textbox(
-                lines=2, 
-                label='Enter Tickers as a List', 
-                value='["RELIANCE", "INFY", "TCS"]',
+                lines=3, 
+                label='Enter Tickers as List', 
+                value='["RELIANCE", "INFY", "TCS", "HDFCBANK", "ICICIBANK"]',
                 placeholder='["TICKER1", "TICKER2", "TICKER3"]'
             )
-            start_button = gr.Button("Start Processing", variant="primary")
+            start_button = gr.Button("🚀 Start Processing", variant="primary")
             
-            # Add some helpful info
-            gr.Markdown("**Note:** Each ticker makes 2 API calls (1440m and 15m intervals)")
+            gr.Markdown("""
+            **Performance Notes:**
+            - Each ticker = 2 API calls (1440m + 15m intervals)
+            - All requests execute concurrently
+            - No artificial batching delays
+            """)
         
         with gr.Column(scale=2):
-            logs_output = gr.Textbox(label="Logs & Progress", lines=15, interactive=False)
-            json_output = gr.JSON(label="Processed JSON Output")
+            logs_output = gr.Textbox(label="⏱️ Progress & Timing", lines=10, interactive=False)
+            json_output = gr.JSON(label="📊 Processed Results")
 
-    # --- Triggers ---
     start_button.click(
         fn=run_backend_processing,
         inputs=[tickers_input],