import gradio as gr import asyncio import json import re import logging import time import atexit from typing import List, Dict, Optional, Union from datetime import datetime, time as dt_time import pytz import tzlocal import aiohttp from fastapi import FastAPI, HTTPException from pydantic import BaseModel, validator import uvicorn import threading import weakref # Configure enhanced logging with timestamps logging.basicConfig( level=logging.INFO, format='%(asctime)s.%(msecs)03d - %(name)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) logger = logging.getLogger(__name__) # --- Enhanced Session Management --- class SessionManager: """ Thread-safe session manager with graceful rotation to prevent race conditions. """ def __init__(self): self._session: aiohttp.ClientSession = None self._session_lock = asyncio.Lock() # Lock is created once and tied to the event loop. self._creation_time = None self._request_count = 0 self.max_session_age = 300 # 5 minutes self.max_requests_per_session = 1000 async def _graceful_close(self, session_to_close: aiohttp.ClientSession, delay: int = 5): """Waits for a delay before closing a stale session to allow in-flight requests to complete.""" if session_to_close and not session_to_close.closed: logger.info(f"โณ Waiting {delay}s before closing stale session (ID: {id(session_to_close)})...") await asyncio.sleep(delay) # **SYNTAX ERROR FIX**: Added the required colon ':' after 'try' try: logger.info(f"๐Ÿงน Gracefully closing stale session (ID: {id(session_to_close)}).") await session_to_close.close() except Exception as e: logger.warning(f"โš ๏ธ Error during graceful close of stale session: {e}") async def get_session(self) -> aiohttp.ClientSession: """ Get or create a session with graceful rotation to prevent race conditions. """ async with self._session_lock: now = time.time() # Use >= for precision on max requests. needs_renewal = ( self._session is None or self._session.closed or (self._creation_time and now - self._creation_time > self.max_session_age) or self._request_count >= self.max_requests_per_session ) if needs_renewal: old_session = self._session # *** RACE CONDITION FIX: Schedule the old session's closure instead of awaiting it. *** if old_session and not old_session.closed: logger.info(f"๐Ÿ”„ Scheduling closure of old session after {self._request_count} requests.") loop = asyncio.get_running_loop() loop.create_task(self._graceful_close(old_session)) # --- Create the new session immediately --- try: connector = aiohttp.TCPConnector( limit=100, limit_per_host=50, ttl_dns_cache=300, use_dns_cache=True, keepalive_timeout=30, enable_cleanup_closed=True ) timeout = aiohttp.ClientTimeout(total=60, connect=10, sock_read=30) headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', 'Accept': 'application/json', 'Accept-Encoding': 'gzip, deflate', 'Connection': 'keep-alive', 'Cache-Control': 'no-cache' } self._session = aiohttp.ClientSession( connector=connector, timeout=timeout, headers=headers ) self._creation_time = now self._request_count = 0 logger.info(f"๐Ÿš€ Created new session at {datetime.fromtimestamp(now)}") except Exception as e: logger.error(f"Failed to create new session: {e}") raise return self._session def increment_request_count(self): """Increment the request count. Called after a session is successfully retrieved.""" self._request_count += 1 async def close(self): """Clean shutdown of the current active session.""" if self._session and not self._session.closed: try: logger.info(f"๐Ÿ”’ Closing active session after {self._request_count} requests.") await self._session.close() await asyncio.sleep(0.1) # Short delay for cleanup except Exception as e: logger.warning(f"Error during final session cleanup: {e}") self._session = None # Global session manager per event loop to handle multiple threads/loops _session_managers = weakref.WeakKeyDictionary() def get_session_manager(): """Get or create session manager for current event loop.""" try: loop = asyncio.get_running_loop() if loop not in _session_managers: _session_managers[loop] = SessionManager() logger.debug(f"๐Ÿ“ Created new SessionManager for event loop {id(loop)}") return _session_managers[loop] except RuntimeError: # No event loop running logger.warning("โš ๏ธ No event loop running, cannot get session manager.") return None # --- Enhanced Data Models --- class StockDataRequest(BaseModel): tickers: List[str] start_date: str end_date: str interval: int = 15 timezone: str = "Asia/Kolkata" batch_size: int = 50 batch_delay: float = 0.5 max_concurrent: int = 50 @validator('tickers') def validate_tickers(cls, v): if not v: raise ValueError("Tickers list cannot be empty") return [ticker.strip().upper() for ticker in v] @validator('interval') def validate_interval(cls, v): if v <= 0: raise ValueError("Interval must be positive") return v @validator('batch_size') def validate_batch_size(cls, v): if v <= 0 or v > 100: raise ValueError("Batch size must be between 1 and 100") return v @validator('max_concurrent') def validate_max_concurrent(cls, v): if v <= 0 or v > 100: raise ValueError("Max concurrent must be between 1 and 100") return v # --- Utility Functions (same as before) --- class DateTimeValidationError(Exception): pass def validate_datetime_format(dt_str: str) -> datetime: """Validate date in strict 'YYYY-MM-DD' format.""" 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: parsed_date = datetime.strptime(dt_str, '%Y-%m-%d') today = datetime.now().date() if parsed_date.date() > today: raise DateTimeValidationError( f"Future date provided: '{dt_str}'. Please provide a past or current date." ) return parsed_date except ValueError as e: raise DateTimeValidationError( f"Invalid date value: '{dt_str}'. Please provide a valid calendar date." ) from e def _resolve_timezone(timezone: Optional[str]) -> pytz.BaseTzInfo: """Resolve timezone string to pytz timezone object.""" try: if timezone: return pytz.timezone(timezone) else: return tzlocal.get_localzone() except pytz.exceptions.UnknownTimeZoneError: logger.warning(f"Unknown timezone '{timezone}', falling back to Asia/Kolkata") return pytz.timezone('Asia/Kolkata') def convert_to_unixtimestamp(date_time_str: str, timezone: Optional[str] = None) -> int: """Convert 'YYYY-MM-DD HH:MM' string to Unix ms timestamp.""" dt = datetime.strptime(date_time_str, '%Y-%m-%d %H:%M') target_tz = _resolve_timezone(timezone) try: if dt.tzinfo is None: localized_dt = target_tz.localize(dt) else: localized_dt = dt.astimezone(target_tz) return int(localized_dt.timestamp() * 1000) except Exception as e: logger.error(f"Error converting datetime to timestamp: {e}") raise def get_time_range_in_unix_ms(start_date_str: str, end_date_str: str, timezone: str = 'Asia/Kolkata') -> Dict[str, int]: """Convert start/end date into full-day unix ms timestamps.""" start_date = validate_datetime_format(start_date_str) end_date = validate_datetime_format(end_date_str) if start_date > end_date: raise DateTimeValidationError( f"Start date '{start_date_str}' cannot be after end date '{end_date_str}'" ) start_datetime = datetime.combine(start_date, dt_time.min) end_datetime = datetime.combine(end_date, dt_time(23, 59)) start_ts = convert_to_unixtimestamp(start_datetime.strftime('%Y-%m-%d %H:%M'), timezone) end_ts = convert_to_unixtimestamp(end_datetime.strftime('%Y-%m-%d %H:%M'), timezone) return {"start_timestamp_ms": start_ts, "end_timestamp_ms": end_ts} # --- Optimized API Functions --- HIST_URL = "https://groww.in/v1/api/charting_service/v2/chart/exchange/NSE/segment/CASH" async def call_price_api_optimized( ticker: str, start: int, end: int, interval: int, timeout: int = 30, request_id: str = "unknown" ) -> Dict: """Optimized API call with proper session handling and error recovery.""" start_time = time.time() url = f"{HIST_URL}/{ticker}" params = { "startTimeInMillis": start, "endTimeInMillis": end, "intervalInMinutes": interval } try: # Get session manager for current event loop session_mgr = get_session_manager() if session_mgr is None: raise RuntimeError("No event loop available") session = await session_mgr.get_session() session_mgr.increment_request_count() # Log request start logger.debug(f"๐Ÿš€ [{request_id}] Starting request for {ticker}") async with session.get(url, params=params) as response: response_time = time.time() - start_time if response.status == 200: json_data = await response.json() logger.debug(f"โœ… [{request_id}] {ticker}: {response_time*1000:.1f}ms - SUCCESS") return { "ticker": ticker, "data": json_data, "error": None, "status": "success", "response_time_ms": round(response_time * 1000, 1), "request_id": request_id } else: logger.warning(f"โŒ [{request_id}] {ticker}: {response_time*1000:.1f}ms - HTTP {response.status}") return { "ticker": ticker, "data": None, "error": f"HTTP {response.status}: {response.reason}", "status": "failed", "response_time_ms": round(response_time * 1000, 1), "request_id": request_id } except asyncio.CancelledError: response_time = time.time() - start_time logger.error(f"๐Ÿšซ [{request_id}] {ticker}: {response_time*1000:.1f}ms - CANCELLED") raise # Re-raise cancellation except asyncio.TimeoutError: response_time = time.time() - start_time logger.error(f"โฐ [{request_id}] {ticker}: {response_time*1000:.1f}ms - TIMEOUT") return { "ticker": ticker, "data": None, "error": "Request timeout", "status": "timeout", "response_time_ms": round(response_time * 1000, 1), "request_id": request_id } except Exception as e: response_time = time.time() - start_time error_msg = str(e) # Handle specific error types if "Event loop is closed" in error_msg: logger.error(f"๐Ÿ’ฅ [{request_id}] {ticker}: EVENT LOOP CLOSED - attempting recovery") # Try to create a new session manager try: session_mgr = SessionManager() # Create fresh instance session = await session_mgr.get_session() # Retry the request once async with session.get(url, params=params) as response: if response.status == 200: json_data = await response.json() response_time = time.time() - start_time logger.info(f"๐Ÿ”„ [{request_id}] {ticker}: {response_time*1000:.1f}ms - RECOVERED") return { "ticker": ticker, "data": json_data, "error": None, "status": "success", "response_time_ms": round(response_time * 1000, 1), "request_id": request_id } except Exception as retry_e: logger.error(f"๐Ÿ’ฅ [{request_id}] {ticker}: Recovery failed: {retry_e}") logger.error(f"๐Ÿ’ฅ [{request_id}] {ticker}: {response_time*1000:.1f}ms - ERROR: {error_msg}") return { "ticker": ticker, "data": None, "error": error_msg, "status": "error", "response_time_ms": round(response_time * 1000, 1), "request_id": request_id } async def fetch_stock_data_batch_optimized( tickers: List[str], start_time: int, end_time: int, interval: int, batch_size: int = 50, batch_delay: float = 0.5, max_concurrent: int = 50 ) -> List[Dict]: """Highly optimized batch processing with proper event loop and session handling.""" overall_start = time.time() request_id = f"batch_{int(time.time())}" # Split tickers into batches ticker_batches = [tickers[i:i + batch_size] for i in range(0, len(tickers), batch_size)] all_results = [] logger.info(f"๐ŸŽฏ [{request_id}] Starting batch processing: {len(tickers)} tickers in {len(ticker_batches)} batches of {batch_size}") logger.info(f"๐Ÿ“Š [{request_id}] Config: max_concurrent={max_concurrent}, batch_delay={batch_delay}s") # Get or create session manager for this event loop session_mgr = get_session_manager() if session_mgr is None: raise RuntimeError("No event loop available for batch processing") # Pre-warm session logger.info(f"๐Ÿ”ฅ [{request_id}] Pre-warming session...") try: await session_mgr.get_session() logger.info(f"โœ… [{request_id}] Session pre-warmed successfully") except Exception as e: logger.error(f"โŒ [{request_id}] Session pre-warming failed: {e}") raise for batch_idx, ticker_batch in enumerate(ticker_batches): batch_start = time.time() batch_request_id = f"{request_id}_b{batch_idx+1}" logger.info(f"๐Ÿš€ [{batch_request_id}] Processing batch {batch_idx + 1}/{len(ticker_batches)} with {len(ticker_batch)} tickers") # Create semaphore to limit concurrent requests within batch semaphore = asyncio.Semaphore(max_concurrent) async def bounded_fetch(ticker, idx): async with semaphore: tick_request_id = f"{batch_request_id}_t{idx+1}" try: return await call_price_api_optimized( ticker, start_time, end_time, interval, 30, tick_request_id ) except Exception as e: logger.error(f"๐Ÿ’ฅ [{tick_request_id}] Bounded fetch error for {ticker}: {e}") return { "ticker": ticker, "data": None, "error": f"Bounded fetch error: {str(e)}", "status": "error", "response_time_ms": 0, "request_id": tick_request_id } # Process current batch with error handling try: tasks = [bounded_fetch(ticker, idx) for idx, ticker in enumerate(ticker_batch)] batch_results = await asyncio.gather(*tasks, return_exceptions=True) except Exception as e: logger.error(f"๐Ÿ’ฅ [{batch_request_id}] Batch gather failed: {e}") # Create error results for entire batch batch_results = [Exception(f"Batch gather failed: {e}") for _ in ticker_batch] # Process results processed_batch_results = [] successful_in_batch = 0 for i, result in enumerate(batch_results): if isinstance(result, Exception): logger.error(f"๐Ÿ’ฅ [{batch_request_id}] Exception for {ticker_batch[i]}: {str(result)}") processed_batch_results.append({ "ticker": ticker_batch[i], "data": None, "error": str(result), "status": "exception", "batch": batch_idx + 1, "response_time_ms": 0, "request_id": f"{batch_request_id}_t{i+1}" }) else: result["batch"] = batch_idx + 1 processed_batch_results.append(result) if result.get("status") == "success": successful_in_batch += 1 all_results.extend(processed_batch_results) batch_duration = time.time() - batch_start avg_response_time = sum(r.get("response_time_ms", 0) for r in processed_batch_results) / len(processed_batch_results) logger.info(f"โœ… [{batch_request_id}] Completed in {batch_duration:.2f}s | Success: {successful_in_batch}/{len(ticker_batch)} | Avg: {avg_response_time:.1f}ms") # Add delay between batches (except for the last batch) if batch_idx < len(ticker_batches) - 1 and batch_delay > 0: logger.info(f"โณ [{batch_request_id}] Waiting {batch_delay}s before next batch...") await asyncio.sleep(batch_delay) overall_duration = time.time() - overall_start total_successful = len([r for r in all_results if r.get('status') == 'success']) success_rate = (total_successful / len(tickers)) * 100 if len(tickers) > 0 else 0 logger.info(f"๐Ÿ [{request_id}] COMPLETED: {overall_duration:.2f}s total | {total_successful}/{len(tickers)} successful ({success_rate:.1f}%)") logger.info(f"๐Ÿ“ˆ [{request_id}] Performance: {len(tickers)/overall_duration:.1f} tickers/sec") return all_results # --- Core Processing Function --- def process_stock_request( tickers: Union[str, List[str]], start_date: str, end_date: str, interval: int = 15, timezone: str = "Asia/Kolkata", batch_size: int = 50, batch_delay: float = 0.5, max_concurrent: int = 50 ) -> Dict: """Legacy function - now delegates to the enhanced loop-safe version.""" logger.info("๐Ÿ”„ Using legacy process_stock_request, delegating to enhanced version") return process_stock_request_with_new_loop( tickers, start_date, end_date, interval, timezone, batch_size, batch_delay, max_concurrent ) # --- FastAPI Application --- def safe_json_serialize(obj): """Safely serialize any object to JSON string.""" def default_serializer(o): if isinstance(o, (datetime, dt_time)): return o.isoformat() elif hasattr(o, '__dict__'): return o.__dict__ elif hasattr(o, 'to_dict'): return o.to_dict() else: return str(o) try: return json.dumps(obj, indent=2, default=default_serializer, ensure_ascii=False) except Exception: return json.dumps(str(obj), indent=2) api_app = FastAPI(title="Optimized Groww Stock Data API", version="2.0.0") @api_app.post("/fetch-stock-data") async def fetch_stock_data_endpoint(request: StockDataRequest): """Optimized API endpoint with enhanced performance monitoring.""" try: # Use the new loop-safe version for API calls result = process_stock_request_with_new_loop( request.tickers, request.start_date, request.end_date, request.interval, request.timezone, request.batch_size, request.batch_delay, request.max_concurrent ) result["timestamp"] = datetime.now().isoformat() serializable_result = safe_json_serialize(result) return json.loads(serializable_result) except Exception as e: logger.error(f"API endpoint error: {e}") return { "success": False, "data": None, "error": str(e), "timestamp": datetime.now().isoformat(), "processing_summary": { "total_tickers": 0, "successful": 0, "failed": 0, "success_rate": "0%", "total_duration_seconds": 0, "batch_processing_used": True }, "request_info": {} } @api_app.get("/health") async def health_check(): """Health check endpoint.""" return {"status": "healthy", "timestamp": datetime.now().isoformat()} @api_app.get("/session-stats") async def session_stats(): """Get session statistics for monitoring.""" try: session_mgr = get_session_manager() if session_mgr and session_mgr._session and not session_mgr._session.closed: return { "session_active": True, "session_age_seconds": time.time() - session_mgr._creation_time if session_mgr._creation_time else 0, "request_count": session_mgr._request_count, "loop_id": session_mgr._loop_id, "timestamp": datetime.now().isoformat() } else: return { "session_active": False, "session_age_seconds": 0, "request_count": 0, "loop_id": None, "timestamp": datetime.now().isoformat() } except Exception as e: return { "error": str(e), "session_active": False, "timestamp": datetime.now().isoformat() } # --- Gradio Interface --- def execute_stock_request( ticker_input: str, start_date: str, end_date: str, interval: int, batch_size: int = 50, batch_delay: float = 0.5, max_concurrent: int = 50 ) -> str: """Enhanced wrapper function for Gradio interface with loop safety.""" try: # Use the new loop-safe version result = process_stock_request_with_new_loop( ticker_input, start_date, end_date, interval, "Asia/Kolkata", batch_size, batch_delay, max_concurrent ) return safe_json_serialize(result) except Exception as e: logger.error(f"Error in execute_stock_request: {e}") error_result = { "success": False, "error": str(e), "timestamp": datetime.now().isoformat(), "processing_summary": { "total_duration_seconds": 0, "throughput_tickers_per_second": 0, "error_type": "gradio_wrapper_error" } } return safe_json_serialize(error_result) def create_gradio_interface(): """Create optimized Gradio interface with performance controls.""" with gr.Blocks(title="Optimized Groww Stock Data Fetcher") as demo: gr.Markdown(""" # โšก Optimized Groww Stock Data Fetcher v2.0 **High-Performance Features:** - ๐Ÿš€ **Persistent HTTP Sessions**: Reuses connections for 5min/1000 requests - ๐Ÿ“Š **Detailed Performance Monitoring**: Response times, throughput metrics - ๐ŸŽฏ **Optimized Batch Processing**: Smart batching with timing controls - โšก **Enhanced Async Processing**: Up to 50 concurrent requests per batch - ๐Ÿ“ˆ **Real-time Statistics**: Success rates, timing analysis **Expected Performance:** ~187 tickers/second (750 tickers in ~4 seconds) """) with gr.Row(): with gr.Column(): ticker_box = gr.Textbox( label="Stock Tickers", placeholder='["RELIANCE","TCS","INFY"] or RELIANCE,TCS,INFY', value='["RELIANCE","TCS","INFY"]', lines=3 ) with gr.Row(): start_box = gr.Textbox( label="Start Date (YYYY-MM-DD)", placeholder="2025-08-01", value="2025-08-01" ) end_box = gr.Textbox( label="End Date (YYYY-MM-DD)", placeholder="2025-08-10", value="2025-08-10" ) with gr.Row(): interval_box = gr.Number( label="Interval (minutes)", value=15, minimum=1, maximum=1440 ) batch_size_box = gr.Number( label="Batch Size", value=50, minimum=1, maximum=100, info="Tickers per batch" ) with gr.Row(): batch_delay_box = gr.Number( label="Batch Delay (seconds)", value=0.5, minimum=0, maximum=10, step=0.1, info="Delay between batches" ) max_concurrent_box = gr.Number( label="Max Concurrent", value=50, minimum=1, maximum=100, info="Concurrent requests per batch" ) fetch_button = gr.Button("๐Ÿš€ Fetch Data (Optimized)", variant="primary", size="lg") gr.Markdown(""" **Performance Tuning:** - **Batch Size**: 50 (optimal for API rate limits) - **Batch Delay**: 0.5s (prevents rate limiting) - **Max Concurrent**: 50 (parallel requests per batch) - **Session Reuse**: Connections kept alive for 5 minutes """) with gr.Column(): output_box = gr.Textbox( label="API Response with Performance Metrics", lines=25, max_lines=40, show_copy_button=True, container=True ) gr.Markdown(""" ### Performance Monitoring The response now includes detailed timing metrics: ```json { "processing_summary": { "total_duration_seconds": 4.23, "throughput_tickers_per_second": 177.3, "avg_response_time_ms": 95.4, "max_response_time_ms": 234.1, "min_response_time_ms": 67.8, "success_rate": "98.75%" } } ``` ### API Usage ```bash curl -X POST "http://localhost:8000/fetch-stock-data" \\ -H "Content-Type: application/json" \\ -d '{ "tickers": ["RELIANCE", "TCS", ...], "start_date": "2025-08-01", "end_date": "2025-08-10", "batch_size": 50, "batch_delay": 0.5, "max_concurrent": 50 }' ``` """) fetch_button.click( fn=execute_stock_request, inputs=[ticker_box, start_box, end_box, interval_box, batch_size_box, batch_delay_box, max_concurrent_box], outputs=output_box ) return demo # --- Main Execution --- def run_api_server(host="0.0.0.0", port=8000): """Run the optimized FastAPI server.""" uvicorn.run(api_app, host=host, port=port, log_level="info") def run_gradio_interface(share=False): """Run the optimized Gradio interface.""" demo = create_gradio_interface() demo.launch(share=share, server_name="0.0.0.0") # --- Enhanced Process Management --- def process_stock_request_with_new_loop( tickers: Union[str, List[str]], start_date: str, end_date: str, interval: int = 15, timezone: str = "Asia/Kolkata", batch_size: int = 50, batch_delay: float = 0.5, max_concurrent: int = 50 ) -> Dict: """Process stock request with a fresh event loop to avoid loop closure issues.""" request_start = time.time() request_id = f"req_{int(request_start)}" logger.info(f"๐Ÿš€ [{request_id}] Starting stock request with fresh event loop") try: # Handle tickers input if isinstance(tickers, str): try: tickers_list = json.loads(tickers) except json.JSONDecodeError: tickers_list = [t.strip().upper() for t in tickers.split(',')] else: tickers_list = [t.strip().upper() for t in tickers] if not tickers_list: raise ValueError("No tickers provided") logger.info(f"๐Ÿ“ [{request_id}] Processing {len(tickers_list)} tickers") # Convert dates to timestamps ts_range = get_time_range_in_unix_ms(start_date, end_date, timezone) start_ts, end_ts = ts_range["start_timestamp_ms"], ts_range["end_timestamp_ms"] # Create a new event loop for this request to avoid closure issues try: # Try to get existing loop first loop = asyncio.get_event_loop() if loop.is_closed(): raise RuntimeError("Event loop is closed") except RuntimeError: # Create new loop if none exists or current is closed loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) logger.info(f"๐Ÿ”„ [{request_id}] Created new event loop") try: # Run the batch processing logger.info(f"โšก [{request_id}] Using optimized batch processing") results = loop.run_until_complete(fetch_stock_data_batch_optimized( tickers_list, start_ts, end_ts, interval, batch_size, batch_delay, max_concurrent )) finally: # Clean up session for this loop if we created it if request_id in [f"req_{int(request_start)}"]: # Only clean if we created the loop try: session_mgr = get_session_manager() if session_mgr: loop.run_until_complete(session_mgr.close()) except Exception as e: logger.warning(f"Error cleaning up session: {e}") # Generate enhanced statistics total_tickers = len(tickers_list) successful = len([r for r in results if r.get('status') == 'success']) failed = len([r for r in results if r.get('status') in ['failed', 'error', 'timeout', 'exception']]) # Calculate timing statistics response_times = [r.get('response_time_ms', 0) for r in results if r.get('response_time_ms', 0) > 0] avg_response_time = sum(response_times) / len(response_times) if response_times else 0 max_response_time = max(response_times) if response_times else 0 min_response_time = min(response_times) if response_times else 0 total_duration = time.time() - request_start throughput = total_tickers / total_duration if total_duration > 0 else 0 processing_summary = { "total_tickers": total_tickers, "successful": successful, "failed": failed, "success_rate": f"{(successful/total_tickers*100):.2f}%" if total_tickers > 0 else "0%", "total_duration_seconds": round(total_duration, 2), "throughput_tickers_per_second": round(throughput, 1), "avg_response_time_ms": round(avg_response_time, 1), "max_response_time_ms": round(max_response_time, 1), "min_response_time_ms": round(min_response_time, 1), "batch_processing_used": True, "batch_size": batch_size, "batch_delay": batch_delay, "max_concurrent": max_concurrent, "request_id": request_id } logger.info(f"โœ… [{request_id}] Request completed successfully in {total_duration:.2f}s") return { "success": True, "data": results, "error": None, "timestamp": datetime.now().isoformat(), "processing_summary": processing_summary, "request_info": { "tickers": tickers_list[:10] if len(tickers_list) > 10 else tickers_list, "total_tickers": len(tickers_list), "start_date": start_date, "end_date": end_date, "interval": interval, "timezone": timezone, "start_timestamp_ms": start_ts, "end_timestamp_ms": end_ts, "request_id": request_id } } except Exception as e: total_duration = time.time() - request_start logger.error(f"โŒ [{request_id}] Error after {total_duration:.2f}s: {e}") return { "success": False, "data": None, "error": str(e), "timestamp": datetime.now().isoformat(), "processing_summary": { "total_tickers": len(tickers_list) if 'tickers_list' in locals() else 0, "successful": 0, "failed": 0, "success_rate": "0%", "total_duration_seconds": round(total_duration, 2), "throughput_tickers_per_second": 0, "batch_processing_used": True, "request_id": request_id }, "request_info": { "tickers": tickers if isinstance(tickers, list) else [tickers], "start_date": start_date, "end_date": end_date, "interval": interval, "timezone": timezone, "request_id": request_id } } # --- Cleanup Handler --- import atexit async def cleanup_all_sessions(): """Cleanup all sessions across all event loops.""" logger.info("๐Ÿงน Starting session cleanup...") try: for loop, session_mgr in list(_session_managers.items()): if not loop.is_closed(): try: await session_mgr.close() logger.info(f"โœ… Cleaned up session for loop {id(loop)}") except Exception as e: logger.warning(f"โš ๏ธ Error cleaning up session for loop {id(loop)}: {e}") except Exception as e: logger.warning(f"โš ๏ธ Error during session cleanup: {e}") def cleanup_session(): """Cleanup session on exit.""" try: # Check if there's a running event loop try: loop = asyncio.get_running_loop() # If we're in a running loop, create a task if not loop.is_closed(): loop.create_task(cleanup_all_sessions()) except RuntimeError: # No running loop, create one for cleanup try: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) loop.run_until_complete(cleanup_all_sessions()) loop.close() except Exception as e: logger.warning(f"โš ๏ธ Error during final cleanup: {e}") except Exception as e: logger.warning(f"โš ๏ธ Error during cleanup: {e}") atexit.register(cleanup_session) def cleanup_session(): """Cleanup session on exit.""" if session_manager._session and not session_manager._session.closed: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) loop.run_until_complete(session_manager.close()) loop.close() atexit.register(cleanup_session) if __name__ == "__main__": import sys # Set logging level based on environment if "--debug" in sys.argv: logging.getLogger().setLevel(logging.DEBUG) logger.info("๐Ÿ› Debug logging enabled") if len(sys.argv) > 1 and sys.argv[1] == "api": # Run only API server print("๐Ÿš€ Starting optimized API server...") run_api_server() elif len(sys.argv) > 1 and sys.argv[1] == "both": # Run both API and Gradio in separate threads print("๐Ÿš€ Starting both API server and Gradio interface...") api_thread = threading.Thread(target=run_api_server, daemon=True) api_thread.start() print("โณ Waiting 2 seconds for API server to start...") time.sleep(2) run_gradio_interface() else: # Default: Run only Gradio print("๐Ÿš€ Starting optimized Gradio interface...") run_gradio_interface()