Update app.py

app.py

@@ -1,991 +1,290 @@
-import gradio as gr
+# @title Function To Get PreOpenDetails
+# Install required packages if not installed
+# !pip install aiohttp nest-asyncio pandas gradio pytz requests
+
 import asyncio
+import aiohttp
 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 gradio as gr
+import pandas as pd
+import time
 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
+import requests
+import pytz
+from datetime import datetime
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
+
+import nest_asyncio
+nest_asyncio.apply()
+
+# ==============================================================================
+# 1. TELEGRAM CONFIGURATION
+# ==============================================================================
+BOT_TOKEN = "8220537137:AAGtz1bBsHzMhbxtzMWNLmFbGPHIErcys9o"
+CHAT_ID = "1342204098"
+
+# ==============================================================================
+# 2. YOUR PROVIDED FUNCTIONS (UNCHANGED)
+# ==============================================================================
+
+class NSEAsyncAPI:
+    # ... (Your NSEAsyncAPI class code remains exactly the same) ...
+    def __init__(self, debug: bool = False):
+        self.debug = debug
+        self.base_url = "https://www.nseindia.com"
+        self.headers = {
+            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36',
+            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9',
+            'Accept-Language': 'en-US,en;q=0.9',
+            'Accept-Encoding': 'gzip, deflate, br',
         }
 
-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
+    async def _initialize_session(self, session: aiohttp.ClientSession, symbol: str) -> bool:
+        """Initialize session by visiting the quote page to get cookies"""
+        try:
+            init_url = f"{self.base_url}/get-quotes/equity?symbol={symbol}&series=EQ"
+            if self.debug: print(f"Initializing session for {symbol}: {init_url}")
+            async with session.get(init_url, headers=self.headers, timeout=15) as response:
+                if response.status == 200:
+                    if self.debug: print(f"✅ Session initialized for {symbol}.")
+                    return True
+                else:
+                    if self.debug: print(f"❌ Failed to initialize session for {symbol}: {response.status}")
+                    return False
+        except Exception as e:
+            if self.debug: print(f"❌ Session initialization error for {symbol}: {e}")
+            return False
+
+    async def _get_quote_data(self, session: aiohttp.ClientSession, symbol: str) -> Optional[Dict]:
+        """Get quote data for a single symbol"""
         try:
-            tasks = [bounded_fetch(ticker, idx) for idx, ticker in enumerate(ticker_batch)]
-            batch_results = await asyncio.gather(*tasks, return_exceptions=True)
+            api_url = f"{self.base_url}/api/quote-equity?symbol={symbol}&series=EQ"
+            api_headers = self.headers.copy()
+            api_headers['Accept'] = 'application/json'
+            if self.debug: print(f"Fetching data for {symbol}: {api_url}")
+            async with session.get(api_url, headers=api_headers, timeout=15) as response:
+                if response.status == 200:
+                    data = await response.json()
+                    if self.debug: print(f"✅ Data received for {symbol}")
+                    return data
+                else:
+                    if self.debug: print(f"❌ API request failed for {symbol}: {response.status}")
+                    return None
         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 self.debug: print(f"❌ API request error for {symbol}: {e}")
+            return None
+
+    async def _fetch_single_symbol(self, symbol: str) -> Dict:
+        """Fetch data for a single symbol with session initialization"""
+        timeout = aiohttp.ClientTimeout(total=30)
+        async with aiohttp.ClientSession(timeout=timeout, cookie_jar=aiohttp.CookieJar()) as session:
+            init_success = await self._initialize_session(session, symbol)
+            if not init_success:
+                return {"symbol": symbol, "data": None, "error": "Failed to initialize session"}
+            data = await self._get_quote_data(session, symbol)
+            if data:
+                return {"symbol": symbol, "data": data, "error": None}
+            else:
+                return {"symbol": symbol, "data": None, "error": "Failed to fetch data"}
+
+    async def get_quotes_async(self, symbols: List[str]) -> List[Dict]:
+        """Asynchronously fetch quotes for multiple symbols"""
+        if self.debug: print(f"Starting async fetch for {len(symbols)} symbols: {symbols}")
+        tasks = [self._fetch_single_symbol(symbol) for symbol in symbols]
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+        processed_results = []
+        for i, result in enumerate(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}"
-                })
+                processed_results.append({"symbol": symbols[i], "data": None, "error": str(result)})
             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
+                processed_results.append(result)
+        if self.debug:
+            success_count = sum(1 for r in processed_results if r["data"] is not None)
+            print(f"Completed: {success_count}/{len(symbols)} successful")
+        return processed_results
+
+    def get_quote(self, symbol: str) -> Optional[Dict]:
+        """Get quote for a single symbol (Jupyter-friendly)"""
+        async def _fetch():
+            result = await self._fetch_single_symbol(symbol)
+            return result["data"]
+        return asyncio.run(_fetch())
+
+    def get_quotes(self, symbols: Union[str, List[str]]) -> Union[Dict, List[Dict]]:
+        """Get quotes for single symbol or multiple symbols (Jupyter-friendly)"""
+        if isinstance(symbols, str): return self.get_quote(symbols)
+        results = asyncio.run(self.get_quotes_async(symbols))
+        return [{"symbol": r["symbol"], "data": r["data"], "error": r["error"]} for r in results]
+
+def get_nse_quotes(symbols: List[str], debug: bool = False) -> List[Dict]:
+    api = NSEAsyncAPI(debug=debug)
+    return api.get_quotes(symbols)
+
+def predict_preopen_sentiment(obj):
+      def _extract_data_and_symbol(item):
+          if isinstance(item, dict) and "data" in item:
+              symbol = item.get("symbol") or item.get("data", {}).get("metadata", {}).get("symbol") or "UNKNOWN"
+              data = item["data"]
+              return data, symbol
+          if isinstance(item, dict):
+              symbol = item.get("metadata", {}).get("symbol", "UNKNOWN")
+              return item, symbol
+          return {}, "UNKNOWN"
+
+      def analyze_single(item):
+          data, symbol = _extract_data_and_symbol(item)
+          pre = data.get("preOpenMarket", {}) if data else {}
+          result = { "symbol": symbol, "dominant": "Unknown", "TotalBuyOrder": 0, "TotalSellOrder": 0, "Multiplier": 0.0, "PrevClose": 0.0, "IEP": 0.0 }
+          if not pre: return result
+          buy_qty, sell_qty = pre.get("totalBuyQuantity", 0), pre.get("totalSellQuantity", 0)
+          result.update({"TotalBuyOrder": buy_qty, "TotalSellOrder": sell_qty, "PrevClose": pre.get("prevClose", 0.0), "IEP": pre.get("IEP", 0.0)})
+          dominant, multiplier = "Balanced", 0.0
+          if buy_qty > sell_qty:
+              dominant = "Demand"
+              if sell_qty > 0: multiplier = (buy_qty / sell_qty)
+          elif sell_qty > buy_qty:
+              dominant = "Supply"
+              if buy_qty > 0: multiplier = (-sell_qty / buy_qty)
+          result.update({"dominant": dominant, "Multiplier": round(multiplier, 2)})
+          return result
+
+      return [analyze_single(item) for item in obj] if isinstance(obj, list) else analyze_single(obj)
+
+# ==============================================================================
+# 3. TELEGRAM SENDER FUNCTION
+# ==============================================================================
+def send_telegram_message(text: str):
+    """Send a text message to the configured Telegram chat with retry logic."""
+    session = requests.Session()
+    retry_strategy = Retry(
+        total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504]
     )
-
-# --- 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."""
+    adapter = HTTPAdapter(max_retries=retry_strategy)
+    session.mount("https://", adapter)
+    api_url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"
+    payload = {"chat_id": CHAT_ID, "text": text, "parse_mode": "Markdown"}
     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()
-            }
+        response = session.post(api_url, data=payload, timeout=10)
+        response.raise_for_status()
+        print("Telegram message sent successfully.")
+    except requests.exceptions.RequestException as e:
+        print(f"Failed to send Telegram message after several retries: {e}")
+
+# ==============================================================================
+# 4. APPLICATION LOGIC
+# ==============================================================================
+ALL_SYMBOLS = ["360ONE", "3MINDIA", "ABB", "ACC", "AIAENG", "APLAPOLLO", "AUBANK", "AARTIIND", "AAVAS", "ABBOTTINDIA", "ADANIENT", "ADANIGREEN", "ADANIPORTS", "ADANIPOWER", "ADANITOTAL", "AWL", "ABCAPITAL", "ABFRL", "AEGISCHEM", "AETHER", "AFFLE", "AJANTPHARM", "APLLTD", "ALKEM", "ALKYLAMINE", "ALLCARGO", "ALOKINDS", "AMARAJABAT", "AMBER", "AMBUJACEM", "ANANDRATHI", "ANGELONE", "ANURAS", "APARINDS", "APOLLOHOSP", "APOLLOTYRE", "APTUS", "ARCHEAN", "ASAHIINDIA", "ASHOKLEY", "ASIANPAINT", "ASTERDM", "ASTRAZEN", "ASTRAL", "ATUL", "AUROPHARMA", "AVANTIFEED", "DMART", "AXISBANK", "BEML", "BLS", "BSE", "BAJAJ-AUTO", "BAJFINANCE", "BAJAJFINSV", "BAJAJHLDNG", "BALAMINES", "BALKRISIND", "BALRAMCHIN", "BANDHANBNK", "BATAINDIA", "BAYERCROP", "BERGEPAINT", "BDL", "BEL", "BHARATFORG", "BHEL", "BPCL", "BHARTIARTL", "BIKAJI", "BIOCON", "BIRLACORPN", "BSOFT", "BLUEDART", "BLUESTARCO", "BBTC", "BORORENEW", "BOSCHLTD", "BRIGADE", "BRITANNIA", "MAPMYINDIA", "CCL", "CESC", "CGPOWER", "CIEINDIA", "CRISIL", "CSBBANK", "CAMPUS", "CANFINHOME", "CAPLIPOINT", "CGCL", "CARBORUNIV", "CASTROLIND", "CEATLTD", "CDSL", "CENTRALBK", "CENTURYPLY", "CERA", "CHALET", "CHAMBLFERT", "CHEMPLASTS", "CHOLAFIN", "CHOLAHLDNG", "CIPLA", "CLEAN", "COALINDIA", "COCHINSHIP", "COFORGE", "COLPAL", "CONCOR", "COROMANDEL", "CREDITACC", "CROMPTON", "CUB", "CUMMINSIND", "CYIENT", "DCMSHRIRAM", "DABUR", "DALBHARAT", "DEEPAKNTR", "DELHIVERY", "DEVYANI", "DHANI", "DBL", "DIVISLAB", "DIXON", "LALPATHLAB", "DRREDDY", "ECLERX", "EIDPARRY", "EIHOTEL", "EQUITASBNK", "ERIS", "ESCORTS", "EXIDEIND", "FSL", "FACT", "FDC", "FEDERALBNK", "FINEORG", "FINCABLES", "FINPIPE", "FLUOROCHEM", "FORTIS", "GAEL", "GAIL", "GALAXYSURF", "GARFIBRES", "GESHIP", "GICRE", "GLAND", "GLAXO", "GLENMARK", "GMDCLTD", "GNFC", "GODFRYPHLP", "GODREJCP", "GODREJIND", "GODREJPROP", "GRANULES", "GRAPHITE", "GRASIM", "GESHIP", "GRINDWELL", "GUJALKALI", "GUJGASLTD", "GSFC", "GSPL", "HAVELLS", "HCLTECH", "HDFCAMC", "HDFCBANK", "HDFCLIFE", "HFCL", "HLEGLAS", "HAPPSTMNDS", "HAL", "HEMIPROP", "HEROMOTOCO", "HINDALCO", "HINDCOPPER", "HINDPETRO", "HINDUNILVR", "HINDZINC", "POWERINDIA", "HOMEFIRST", "HONAUT", "HUDCO", "IBULHSGFIN", "ICICIBANK", "ICICIGI", "ICICIPRULI", "IDBI", "IDFCFIRSTB", "IDFC", "IFBIND", "INDIACEM", "IBREALEST", "INDIAMART", "INDIANB", "IEX", "INDHOTEL", "INDIGO", "IOB", "IOC", "IRCTC", "IRB", "IRCON", "IREDA", "ITC", "ITDC", "ITI", "INDUSINDBK", "NAUKRI", "INFY", "INGERRAND", "INTELLECT", "IOB", "IPCALAB", "JBCHEPHARM", "JKCEMENT", "JKLAKSHMI", "JKPAPER", "JMFINANCIL", "JSWENERGY", "JSWSTEEL", "JAGRAN", "JAICORPLTD", "JSL", "JINDALSTEL", "JIOFIN", "JUBLFOOD", "JUBLINGT", "JUBLPHARMA", "JUSTDIAL", "JYOTHYLAB", "KPRMILL", "KEI", "KNRCON", "KPITTECH", "KRBL", "KAJARIACER", "KALPATPOWR", "KALYANKJIL", "KANSAINER", "KARURVYSYA", "KEC", "KIMS", "KOTAKBANK", "L&TFH", "LTTS", "LICHSGFIN", "LAURUSLABS", "LAXMIMACH", "LTIM", "LT", "LICI", "LINDEINDIA", "LUPIN", "LUXIND", "LXCHEM", "MMTC", "MOIL", "MRF", "MTARTECH", "MGL", "MAHABANK", "M&M", "MAHINDCIE", "M&MFIN", "MAHLIFE", "MANAPPURAM", "MRPL", "MARICO", "MARUTI", "MASTEK", "MAXHEALTH", "MAZFAB", "MEDANTA", "METROBRAND", "METROPOLIS", "MFSL", "MOTILALOFS", "MPHASIS", "MUMBAIAIRPORT", "MUTHOOTFIN", "NATCOPHARM", "NBCC", "NCC", "NHPC", "NLCINDIA", "NMDC", "NTPC", "NATIONALUM", "NAVINFLUOR", "NAZARA", "NESTLEIND", "NETWORK18", "NEWGEN", "NH", "NIACL", "NILKAMAL", "NOCIL", "NUVAMA", "OBEROIRLTY", "ONGC", "OIL", "PAYTM", "OFSS", "ORIENTELEC", "PCBL", "PIIND", "PNB", "PNBHOUSING", "PNCINFRA", "PFC", "PAGEIND", "PATANJALI", "PERSISTENT", "PETRONET", "PFIZER", "PHOENIXLTD", "PIDILITIND", "PEL", "POLYMED", "POLYCAB", "POONAWALLA", "POWERGRID", "PRAJIND", "PRESTIGE", "PRINCEPIPE", "PRSMJOHNSN", "PRUDENT", "PSYCHE", "PUJAPNP", "QUESS", "RBLBANK", "RECLTD", "RVNL", "RAILTEL", "RITES", "RADICO", "RAIN", "RAJESHEXPO", "RALLIS", "RCF", "RATNAMANI", "RAYMOND", "REDINGTON", "RELAXO", "RELIANCE", "RINFRA", "RENUKA", "RHIM", "RKFORG", "ROYALCHID", "SAIL", "SBIN", "SJVN", "SKFINDIA", "SRF", "SANOFI", "SAPPHIRE", "SAREGAMA", "SCHAEFFLER", "SENATE", "SEQUENT", "SFL", "SHOPERSTOP", "SHREECEM", "SHRIRAMFIN", "SIEMENS", "SIGNATURE", "SILV", "SOBHA", "SOLARINDS", "SONACOMS", "SONATSOFTW", "SPARC", "SPICEJET", "STARHEALTH", "SBILIFE", "SWSOLAR", "STEELXIND", "STERTOOLS", "STLTECH", "SUDARSCHEM", "SUMICHEM", "SUMIT", "SUNDARMFIN", "SUNDRMFAST", "SUNPHARMA", "SUNTV", "SUPRAJIT", "SUPREMEIND", "SUVENPHAR", "SYMPHONY", "SYNGENE", "TCIEXP", "TCNSBRANDS", "TTKPRESTIG", "TV18BRDCST", "TVSMOTOR", "TANLA", "TATACHEM", "TATACOMM", "TATACONSUM", "TATAELXSI", "TATAINVEST", "TATAMTRDVR", "TATAMOTORS", "TATAPOWER", "TATASTEEL", "TCS", "TECHM", "TEJASNET", "THERMAX", "TIMKEN", "TITAN", "TORNTPHARM", "TORNTPOWER", "TRENT", "TRIDENT", "TRIVENI", "TRU", "UBL", "UCOBANK", "ULTRACEMCO", "UNIONBANK", "UPL", "UTIAMC", "VAIBHAVGBL", "VGUARD", "VARROC", "VBL", "VEDL", "VENKEYS", "VIJAYA", "VOLTAS", "WELCORP", "WELSPUNIND", "WESTLIFE", "WHIRLPOOL", "WIPRO", "WOCKPHARMA", "YESBANK", "ZEEL", "ZENSARTECH", "ZFCVINDIA", "ZOMATO", "ZYDUSLIFE"]
+
+def create_batches(data: list, batch_size: int):
+    for i in range(0, len(data), batch_size): yield data[i:i + batch_size]
+
+def analyze_symbols(symbols: List[str]):
+    if not symbols: return pd.DataFrame()
+    all_quotes = []
+    symbol_batches = list(create_batches(symbols, 75))
+    print(f"Processing {len(symbols)} symbols in {len(symbol_batches)} batches.")
+    for i, batch in enumerate(symbol_batches):
+        print(f"Fetching batch {i+1}/{len(symbol_batches)}...")
+        quotes = get_nse_quotes(batch, debug=False)
+        all_quotes.extend(quotes)
+    print("All data fetched. Analyzing sentiment...")
+    analysis_results = predict_preopen_sentiment(all_quotes)
+    return pd.DataFrame(analysis_results)
+
+def manual_pull_handler(selected_symbols: List[str], select_all: bool):
+    symbols_to_process = ALL_SYMBOLS if select_all else selected_symbols
+    if not symbols_to_process:
+        return pd.DataFrame(), "Please select symbols or check 'Select All'."
+    df = analyze_symbols(symbols_to_process)
+    return df, f"Successfully analyzed {len(df)} symbols."
+
+# ==============================================================================
+# 5. SCHEDULING LOGIC
+# ==============================================================================
+def scheduled_job():
+    print("\n" + "="*50)
+    print("RUNNING SCHEDULED JOB at 09:10 AM IST...")
+    df = analyze_symbols(ALL_SYMBOLS)
+    print("SCHEDULED JOB FINISHED. Data:")
+    print(df)
+
+    if df is not None and not df.empty:
+        significant_movers = df[df['Multiplier'].abs() >= 15].copy()
+        significant_movers.sort_values(by='Multiplier', ascending=False, inplace=True)
+        if not significant_movers.empty:
+            print(f"Found {len(significant_movers)} significant symbols. Formatting for Telegram.")
+            message_header = "🚀 *Daily Pre-Open Movers (|Multiplier| >= 15)*\n"
+            all_cards = []
+            for i, row in enumerate(significant_movers.itertuples()):
+                dom_text = { "Demand": "🟢 **Dominant: Demand**", "Supply": "🔴 **Dominant: Supply**" }.get(row.dominant, "⚪️ **Dominant: Balanced**")
+                mult_emoji = "📈" if row.Multiplier > 0 else "📉"
+                card = f"""*-----------------------------------*
+{i+1}️⃣ ***{row.symbol}***
+{dom_text}
+💰 Prev Close: `{row.PrevClose:.2f}`
+⚖️ Open Price (IEP): `{row.IEP:.2f}`
+🛒 Buy Qty: `{row.TotalBuyOrder:,}`
+🛑 Sell Qty: `{row.TotalSellOrder:,}`
+{mult_emoji} Multiplier: `{row.Multiplier:.2f}`"""
+                all_cards.append(card)
+            message = message_header + "\n".join(all_cards)
+            send_telegram_message(message)
         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
-    import threading
-    import time
-
-    # Set logging level based on environment
-    if "--debug" in sys.argv:
-        logging.getLogger().setLevel(logging.DEBUG)
-        logger.info("🐛 Debug logging enabled")
-
-    # --- Thread-safe function to run the API server ---
-    def run_api():
-        print("🚀 Starting optimized API server on http://0.0.0.0:8000")
-        run_api_server(host="0.0.0.0", port=8000)
-
-    # --- Main function to start Gradio ---
-    def run_gradio():
-        print("🚀 Starting optimized Gradio interface...")
-        # The demo.launch() will block the main thread
-        run_gradio_interface()
-
-    # --- Logic to run API, Gradio, or both ---
-    if len(sys.argv) > 1 and sys.argv[1] == "api":
-        # Run only the API server
-        run_api()
-
-    elif len(sys.argv) > 1 and sys.argv[1] == "gradio":
-        # Run only the Gradio interface
-        run_gradio()
-
+            print("No symbols found with |Multiplier| >= 15 today.")
     else:
-        # Default behavior: Run both API and Gradio
-        print("🚀 Starting both API server and Gradio interface...")
-
-        # Run the API server in a daemon thread
-        # A daemon thread will exit when the main program exits.
-        api_thread = threading.Thread(target=run_api, daemon=True)
-        api_thread.start()
+        print("Analysis returned no data. Skipping Telegram notification.")
+    print("="*50 + "\n")
+
+def run_scheduler():
+    """Checks the time in IST every minute and runs the job at 09:10."""
+    target_tz = pytz.timezone("Asia/Kolkata")
+    target_hour, target_minute = 22, 29
+    job_has_run_today = False
+    print(f"Scheduler started. Will run job daily at {target_hour:02d}:{target_minute:02d} IST.")
+
+    while True:
+        now_ist = datetime.now(target_tz)
+        if now_ist.hour == target_hour and now_ist.minute == target_minute and not job_has_run_today:
+            scheduled_job()
+            job_has_run_today = True
+        if now_ist.hour == 0 and now_ist.minute == 0:
+            job_has_run_today = False
+        time.sleep(60)
+
+# ==============================================================================
+# 6. GRADIO USER INTERFACE (WITH BUTTON MOVED TO TOP)
+# ==============================================================================
+with gr.Blocks(title="NSE Pre-Open Market Analyzer") as demo:
+    gr.Markdown("# NSE Pre-Open Market Sentiment Analyzer")
+    gr.Markdown("Select symbols manually or choose 'Select All' to analyze pre-open market data. The data is also fetched automatically every day at 09:10 AM IST.")
+    with gr.Row():
+        with gr.Column(scale=1):
+            
+            # !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! #
+            # BUTTON IS NOW THE FIRST ELEMENT IN THIS COLUMN      #
+            # !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! #
+            manual_pull_button = gr.Button("Manual Pull Data", variant="primary")
+            
+            select_all_checkbox = gr.Checkbox(label="Select All Symbols", value=False)
+            symbol_selector = gr.CheckboxGroup(choices=ALL_SYMBOLS, label="Select Symbols", interactive=True)
+            
+            def update_symbol_selector(select_all):
+                return gr.CheckboxGroup(value=ALL_SYMBOLS, interactive=False) if select_all else gr.CheckboxGroup(value=[], interactive=True)
+            
+            select_all_checkbox.change(fn=update_symbol_selector, inputs=select_all_checkbox, outputs=symbol_selector)
+            
+            status_textbox = gr.Textbox(label="Status", interactive=False)
 
-        # Give the API server a moment to initialize
-        print("⏳ Waiting 2 seconds for API server to start...")
-        time.sleep(2)
+        with gr.Column(scale=3):
+            output_dataframe = gr.DataFrame(label="Analysis Results")
+            
+    manual_pull_button.click(fn=manual_pull_handler, inputs=[symbol_selector, select_all_checkbox], outputs=[output_dataframe, status_textbox])
 
-        # Run the Gradio interface in the main thread
-        run_gradio()
+if __name__ == "__main__":
+    scheduler_thread = threading.Thread(target=run_scheduler)
+    scheduler_thread.daemon = True
+    scheduler_thread.start()
+    print("Starting Gradio application...")
+    demo.launch()