Update app.py

app.py

@@ -9,7 +9,6 @@ import tzlocal
 import logging
 import json
 import ast
-from concurrent.futures import ThreadPoolExecutor
 
 # --- Basic Setup ---
 
@@ -73,9 +72,9 @@ 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}
 
 
-# --- OPTIMIZED Async Functions ---
+# --- OPTIMIZED API Functions - Interval-Based Batching ---
 
-async def fetch_single_request(session: aiohttp.ClientSession, ticker: str, start: int, end: int, interval: int) -> Dict:
+async def fetch_single_ticker_interval(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}
@@ -85,162 +84,183 @@ async def fetch_single_request(session: aiohttp.ClientSession, ticker: str, star
             json_data = await response.json()
             return {"ticker": ticker, "interval": interval, "data": json_data, "error": None}
     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 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."""
+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 = []
     
-    # Create ALL tasks at once
-    all_tasks = []
-    for ticker in tickers:
-        for interval in intervals:
-            all_tasks.append((ticker, start_time, end_time, interval))
-    
-    logger.info(f"Creating {len(all_tasks)} concurrent requests...")
-    
-    # Configure session for high concurrency
-    connector = aiohttp.TCPConnector(
-        limit=100,  # Max concurrent connections
-        limit_per_host=50,  # Max per host
-        keepalive_timeout=30
-    )
-    
-    timeout = aiohttp.ClientTimeout(total=30, connect=10)
-    
-    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
-        # Create all coroutines
+    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")
+        
+        # Create tasks for this batch
         tasks = [
-            fetch_single_request(session, ticker, start, end, interval)
-            for ticker, start, end, interval in all_tasks
+            fetch_single_ticker_interval(session, ticker, start_time, end_time, interval)
+            for ticker in batch_tickers
         ]
         
-        # Execute ALL requests concurrently
-        logger.info("Executing all requests concurrently...")
-        start_time_fetch = datetime.now()
-        
-        results = await asyncio.gather(*tasks, return_exceptions=True)
+        # Execute batch concurrently
+        batch_results = await asyncio.gather(*tasks)
+        results.extend(batch_results)
         
-        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")
+        # Small delay between batches to be API-friendly
+        if i + batch_size < len(tickers):
+            await asyncio.sleep(0.05)
+    
+    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 = []
+    
+    # Configure session for optimal performance
+    connector = aiohttp.TCPConnector(limit=50, limit_per_host=30)
+    timeout = aiohttp.ClientTimeout(total=20, connect=5)
+    
+    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
         
-        return valid_results
+        # 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)
+            
+            interval_duration = (datetime.now() - interval_start).total_seconds()
+            logger.info(f"Completed {interval}m interval in {interval_duration:.2f}s - Got {len(interval_results)} results")
+            
+            all_results.extend(interval_results)
+    
+    return all_results
 
-def process_data_sync(results: List[Dict]) -> str:
-    """Process data synchronously - this is fast."""
+def merge_data_fast(results: List[Dict]) -> str:
+    """Fast data merging without unnecessary overhead."""
     merged_data = {}
     
     for result in results:
-        ticker = result.get("ticker")
-        interval = result.get("interval")
-        data = result.get("data")
-        error = result.get("error")
+        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
-            
-        if data and data.get("candles") and data["candles"]:
-            first_candle = data["candles"][0]
-            if len(first_candle) > 4:
+        
+        # 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][f"{prefix} open"] = first_candle[1]
-                merged_data[ticker][f"{prefix} high"] = first_candle[2]
-                merged_data[ticker][f"{prefix} low"] = first_candle[3]
-                merged_data[ticker][f"{prefix} close"] = first_candle[4]
+                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 data or candles found"
+            merged_data[ticker][f"error_{interval}m"] = "No candles data"
     
-    return json.dumps(list(merged_data.values()), indent=4)
+    return json.dumps(list(merged_data.values()), indent=2)
 
 
-# --- STREAMLINED Gradio Backend Function ---
+# --- MAIN Gradio Backend Function ---
 
 async def run_backend_processing(tickers_text: str, progress=gr.Progress(track_tqdm=True)):
-    """Streamlined backend processing with minimal yields."""
+    """Main processing function - optimized for speed."""
+    
+    # Input validation
     if not tickers_text.strip():
-        yield "Error: Input is empty. Please provide a list of tickers.", "{}"
+        yield "❌ Error: Input is empty", "{}"
         return
 
     try:
         parsed_input = ast.literal_eval(tickers_text)
         if not isinstance(parsed_input, list):
-            raise TypeError("Input must be a list.")
-        tickers = [str(item).strip().upper() for item in parsed_input]
+            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: The provided list is empty.", "{}"
+            yield "❌ Error: No valid tickers found", "{}"
             return
-    except (ValueError, SyntaxError, TypeError):
-        yield 'Invalid input format. Please provide a list of strings, e.g., ["RELIANCE", "INFY"]', "{}"
+    except:
+        yield '❌ Invalid format. Use: ["TICKER1", "TICKER2"]', "{}"
         return
 
-    # Get timestamps
+    # 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 = [1440, 15]
+    intervals = [15, 1440]  # Process 15m first, then 1440m
     
     total_requests = len(tickers) * len(intervals)
-    yield f"Fetching data for {len(tickers)} tickers ({total_requests} total requests)...", "{}"
     
     try:
-        # Time the entire operation
         overall_start = datetime.now()
         
-        # Fetch all data with maximum concurrency
-        results = await fetch_all_data_optimized(tickers, start_time, end_time, intervals)
+        # Update progress
+        yield f"🚀 Starting: {len(tickers)} tickers × {len(intervals)} intervals = {total_requests} requests", "{}"
+        
+        # Fetch all data using interval-based batching
+        results = await fetch_all_data_by_intervals(tickers, start_time, end_time, intervals)
         
-        # Process data (this is fast)
-        processed_json = process_data_sync(results)
+        # Merge data quickly
+        merge_start = datetime.now()
+        final_json = merge_data_fast(results)
+        merge_time = (datetime.now() - merge_start).total_seconds()
         
         total_time = (datetime.now() - overall_start).total_seconds()
         
-        yield f"✅ Complete! Processed {len(results)} responses in {total_time:.2f} seconds", processed_json
+        success_count = len([r for r in results if r.get("error") is None])
+        yield f"✅ Complete! {success_count}/{total_requests} successful in {total_time:.2f}s (merge: {merge_time:.3f}s)", final_json
         
     except Exception as e:
-        logger.error(f"Error: {e}")
-        yield f"An error occurred: {e}", "{}"
+        logger.error(f"Processing error: {e}")
+        yield f"❌ Error: {str(e)}", "{}"
 
 
 # --- Gradio UI ---
 
-with gr.Blocks(theme=gr.themes.Soft()) as demo:
-    gr.Markdown("## ⚡ High-Performance Stock Data Processor")
-    gr.Markdown("Optimized for maximum concurrency - fetches all data simultaneously!")
+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=3, 
-                label='Enter Tickers as List', 
-                value='["RELIANCE", "INFY", "TCS", "HDFCBANK", "ICICIBANK"]',
-                placeholder='["TICKER1", "TICKER2", "TICKER3"]'
+                lines=4, 
+                label='📈 Stock Tickers (Python List Format)', 
+                value='["RELIANCE", "INFY", "TCS"]',
+                placeholder='["TICKER1", "TICKER2", "TICKER3", ...]'
             )
-            start_button = gr.Button("🚀 Start Processing", variant="primary")
+            start_button = gr.Button("🚀 Process Data", variant="primary", size="lg")
             
-            gr.Markdown("""
-            **Performance Notes:**
-            - Each ticker = 2 API calls (1440m + 15m intervals)
-            - All requests execute concurrently
-            - No artificial batching delays
-            """)
+            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="⏱️ Progress & Timing", lines=10, interactive=False)
-            json_output = gr.JSON(label="📊 Processed Results")
+            logs_output = gr.Textbox(label="📊 Processing Logs", lines=8, interactive=False)
+            json_output = gr.JSON(label="📋 Final Results", show_label=True)
 
+    # Event handlers
     start_button.click(
         fn=run_backend_processing,
         inputs=[tickers_input],
         outputs=[logs_output, json_output]
     )
+    
     tickers_input.submit(
         fn=run_backend_processing,
         inputs=[tickers_input],
@@ -248,4 +268,4 @@ with gr.Blocks(theme=gr.themes.Soft()) as demo:
     )
 
 if __name__ == "__main__":
-    demo.launch()
+    demo.launch(share=False, server_name="0.0.0.0")