app.py

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()