Update app.py

app.py

@@ -1,108 +1,283 @@
 import gradio as gr
-import pandas as pd
-import chromadb
-from chromadb.config import Settings
-from chromadb.utils import embedding_functions
+import re
+import asyncio
+import aiohttp
+from datetime import datetime, time
+from typing import Dict, Optional, List
+import pytz
+import tzlocal
+import logging
+import json
+from datetime import date
 
-# Constants
-CSV_PATH = "ime-cdata - Sheet1.csv"  # Replace with your actual CSV file path
-COLLECTION_NAME = "query_embeddings"
-EMBEDDING_MODEL = 'all-MiniLM-L6-v2'
+# --- Basic Setup ---
 
-# Initialize embedding function
-embedding_function = embedding_functions.SentenceTransformerEmbeddingFunction(model_name=EMBEDDING_MODEL)
+# 1. Logger Configuration
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
 
-# Initialize Chroma client
-chroma_client = chromadb.Client(Settings(persist_directory="./chroma_db"))
+# 2. Custom Exception
+class DateTimeValidationError(ValueError):
+    """Custom exception for datetime validation errors."""
+    pass
 
-def load_and_embed_data():
-    # Load CSV
-    df = pd.read_csv(CSV_PATH)
-    
-    # Create or get collection
-    collection = chroma_client.get_or_create_collection(name=COLLECTION_NAME, embedding_function=embedding_function)
-    
-    # Check if collection is empty
-    if collection.count() == 0:
-        print("Embedding and storing data. This may take a while...")
-        # Embed and store only the query column
-        collection.add(
-            documents=df['query'].tolist(),
-            metadatas=df.to_dict('records'),
-            ids=[str(i) for i in range(len(df))]
+# 3. Assume hist_url is globally defined
+hist_url = "https://groww.in/v1/api/charting_service/v2/chart/exchange/NSE/segment/CASH"
+
+
+# --- Date/Time Functions ---
+
+def validate_datetime_format(dt_str: str) -> datetime:
+    """
+    Validate datetime string in strict 'YYYY-MM-DD' format.
+    """
+    if not isinstance(dt_str, str):
+        logger.error(f"Expected string input, got {type(dt_str).__name__}")
+        raise TypeError(f"Input must be a string, got {type(dt_str).__name__}")
+
+    date_pattern = re.compile(r'^\d{4}-\d{2}-\d{2}$')
+    if not date_pattern.match(dt_str):
+        logger.error(f"Invalid date format: '{dt_str}'. Expected 'YYYY-MM-DD'")
+        raise DateTimeValidationError(
+            f"Invalid date format: '{dt_str}'. Expected 'YYYY-MM-DD'"
         )
-        print("Data embedded and stored.")
-    else:
-        print("Data already embedded and stored.")
-    
-    return collection
 
-def search_similar_queries(query, collection, top_k=5):
     try:
-        results = collection.query(
-            query_texts=[query],
-            n_results=top_k,
-            include=["metadatas", "distances"]
+        parsed_date = datetime.strptime(dt_str, '%Y-%m-%d')
+        logger.debug(f"Successfully validated date: {dt_str}")
+        return parsed_date
+    except ValueError as e:
+        logger.error(f"Invalid date value: '{dt_str}' - {str(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."""
+    if timezone:
+        try:
+            return pytz.timezone(timezone)
+        except pytz.UnknownTimeZoneError as e:
+            logger.error(f"Unknown timezone: '{timezone}'")
+            raise ValueError(f"Unknown timezone: '{timezone}'") from e
+    else:
+        local_tz = tzlocal.get_localzone()
+        logger.debug(f"Using local timezone: {local_tz}")
+        return local_tz
+
+def convert_to_unixtimestamp(
+    date_time_str: str,
+    timezone: Optional[str] = None
+) -> int:
+    """
+    Convert datetime string to Unix timestamp in milliseconds with timezone handling.
+    """
+    if not isinstance(date_time_str, str):
+        logger.error(f"Expected string input, got {type(date_time_str).__name__}")
+        raise TypeError(
+            f"DateTime input must be a string in 'YYYY-MM-DD HH:MM' format, "
+            f"got {type(date_time_str).__name__}"
         )
-        
-        formatted_results = []
-        for i in range(len(results['ids'][0])):
-            metadata = results['metadatas'][0][i]
-            result = {
-                'query': metadata.get('query', 'N/A'),
-                'similarity': 1 - results['distances'][0][i],  # Convert distance to similarity
-                'uber_intent': metadata.get('uber_intent', 'N/A'),
-                'common_intent': metadata.get('common_intent', 'N/A'),
-                'sub_common_intent': metadata.get('sub_common_intent', 'N/A'),
-                'fsc': metadata.get('fsc', 'N/A'),
-                'language': metadata.get('language', 'N/A'),
-                'Name': metadata.get('Name', 'N/A')
+
+    try:
+        dt = datetime.strptime(date_time_str, '%Y-%m-%d %H:%M')
+        logger.debug(f"Successfully parsed datetime: {date_time_str}")
+    except ValueError as e:
+        logger.error(f"Invalid datetime format: '{date_time_str}'")
+        raise DateTimeValidationError(
+            f"Invalid datetime format: '{date_time_str}'. Expected 'YYYY-MM-DD HH:MM'"
+        ) from e
+
+    target_tz = _resolve_timezone(timezone)
+
+    if dt.tzinfo is None:
+        localized_dt = target_tz.localize(dt)
+    else:
+        localized_dt = dt.astimezone(target_tz)
+
+    timestamp_ms = int(localized_dt.timestamp() * 1000)
+    logger.debug(f"Converted '{date_time_str}' to timestamp: {timestamp_ms}")
+
+    return timestamp_ms
+
+def get_time_range_in_unix_ms(
+    start_date_str: str,
+    end_date_str: str,
+    timezone: str = 'Asia/Kolkata'
+) -> Dict[str, int]:
+    """
+    Calculates the start and end Unix timestamps in milliseconds for a date range.
+    """
+    start_date = validate_datetime_format(start_date_str)
+    end_date = validate_datetime_format(end_date_str)
+
+    start_datetime = datetime.combine(start_date, time.min)
+    end_datetime = datetime.combine(end_date, time(23, 59))
+
+    start_datetime_str = start_datetime.strftime('%Y-%m-%d %H:%M')
+    end_datetime_str = end_datetime.strftime('%Y-%m-%d %H:%M')
+
+    start_timestamp = convert_to_unixtimestamp(start_datetime_str, timezone)
+    end_timestamp = convert_to_unixtimestamp(end_datetime_str, timezone)
+
+    return {
+        "start_timestamp_ms": start_timestamp,
+        "end_timestamp_ms": end_timestamp,
+    }
+
+
+# --- Asynchronous API Function ---
+
+async def call_price_api_async(
+    session: aiohttp.ClientSession,
+    ticker: str,
+    start: int,
+    end: int,
+    interval: int,
+) -> Dict:
+    """
+    Asynchronously calls the Groww candle API and returns the raw JSON response.
+    """
+    url = f"{hist_url}/{ticker}"
+    params = {
+        "startTimeInMillis": start,
+        "endTimeInMillis": end,
+        "intervalInMinutes": interval,
+    }
+
+    try:
+        async with session.get(url, params=params) as response:
+            response.raise_for_status()
+            json_data = await response.json()
+            return {
+                "ticker": ticker,
+                "interval": interval,
+                "data": json_data,
+                "error": None,
             }
-            formatted_results.append(result)
-        
-        return formatted_results
-    except Exception as e:
-        print(f"Error in search_similar_queries: {str(e)}")
-        return []
+    except aiohttp.ClientError as e:
+        return {
+            "ticker": ticker,
+            "interval": interval,
+            "data": None,
+            "error": str(e),
+        }
+
+
+# --- Main Processing Logic ---
+
+async def main(tickers: List[str], start_time: int, end_time: int, intervals: List[int], progress: gr.Progress):
+    """
+    Main function to run the asynchronous API calls in batches.
+    """
+    results = []
+    batch_size = 30
+    
+    async with aiohttp.ClientSession() as session:
+        for i in progress.tqdm(range(0, len(tickers), batch_size), desc="Processing Batches"):
+            batch_tickers = tickers[i:i+batch_size]
+            tasks = []
+            for ticker in batch_tickers:
+                for interval in intervals:
+                    tasks.append(
+                        call_price_api_async(session, ticker, start_time, end_time, interval)
+                    )
+            
+            batch_results = await asyncio.gather(*tasks)
+            results.extend(batch_results)
+            
+    return results
+
+def process_and_merge_data(results: list) -> str:
+    """
+    Processes and merges the raw API results for different intervals.
+    """
+    merged_data = {}
+
+    for result in results:
+        ticker = result.get("ticker")
+        interval = result.get("interval")
+        data = result.get("data")
+        error = result.get("error")
+
+        if ticker not in merged_data:
+            merged_data[ticker] = {"symbol": ticker}
+
+        if error:
+            merged_data[ticker][f"error_{interval}m"] = error
+            continue
+
+        if data and data.get("candles") and data["candles"]:
+            first_candle = data["candles"][0]
+            
+            if interval == 1440:
+                merged_data[ticker]["day open"] = first_candle[1] if len(first_candle) > 1 else None
+                merged_data[ticker]["day high"] = first_candle[2] if len(first_candle) > 2 else None
+                merged_data[ticker]["day low"] = first_candle[3] if len(first_candle) > 3 else None
+                merged_data[ticker]["day close"] = first_candle[4] if len(first_candle) > 4 else None
+            elif interval == 15:
+                merged_data[ticker]["start open"] = first_candle[1] if len(first_candle) > 1 else None
+                merged_data[ticker]["start high"] = first_candle[2] if len(first_candle) > 2 else None
+                merged_data[ticker]["start low"] = first_candle[3] if len(first_candle) > 3 else None
+                merged_data[ticker]["start close"] = first_candle[4] if len(first_candle) > 4 else None
+        else:
+            merged_data[ticker][f"error_{interval}m"] = "No data or candles found"
+
+    final_results = list(merged_data.values())
+    return json.dumps(final_results, indent=4)
 
-def gradio_interface(query, top_k):
+# --- Gradio Interface ---
+
+async def run_backend_processing(tickers_text: str, progress=gr.Progress(track_tqdm=True)):
+    """
+    The main function to be called by the Gradio interface.
+    """
+    if not tickers_text.strip():
+        return "Please enter at least one ticker.", "{}"
+
+    tickers = [ticker.strip() for ticker in tickers_text.split('\n') if ticker.strip()]
+    
+    today_str = date.today().strftime("%Y-%m-%d")
+    
+    yield f"Starting processing for {len(tickers)} tickers for date: {today_str}", "{}"
+    
     try:
-        results = search_similar_queries(query, collection, top_k)
+        time_range = get_time_range_in_unix_ms(today_str, today_str)
+        start_time = time_range["start_timestamp_ms"]
+        end_time = time_range["end_timestamp_ms"]
+        intervals = [1440, 15]
+
+        yield "Fetching data from API...", "{}"
+        
+        results = await main(tickers, start_time, end_time, intervals, progress)
         
-        if not results:
-            return "No results found or an error occurred."
+        yield "Processing and merging data...", "{}"
         
-        output = ""
-        for i, result in enumerate(results, 1):
-            output += f"Result {i}:\n"
-            output += f"Query: {result['query']}\n"
-            output += f"Similarity: {result['similarity']:.4f}\n"
-            output += f"Uber Intent: {result['uber_intent']}\n"
-            output += f"Common Intent: {result['common_intent']}\n"
-            output += f"Sub-Common Intent: {result['sub_common_intent']}\n"
-            output += f"FSC: {result['fsc']}\n"
-            output += f"Language: {result['language']}\n"
-            output += f"Name: {result['Name']}\n\n"
+        processed_json = process_and_merge_data(results)
+        
+        yield "Processing complete.", processed_json
         
-        return output
     except Exception as e:
-        print(f"Error in gradio_interface: {str(e)}")
-        return f"An error occurred: {str(e)}"
-
-# Load and embed data
-collection = load_and_embed_data()
-
-# Create Gradio interface
-iface = gr.Interface(
-    fn=gradio_interface,
-    inputs=[
-        gr.Textbox(label="Enter your query"),
-        gr.Slider(minimum=1, maximum=10, step=1, label="Top-K results", value=5)
-    ],
-    outputs=gr.Textbox(label="Results"),
-    title="Query Similarity Search",
-    description="Enter a query to find similar queries with associated metadata."
-)
-
-# Launch the interface
-iface.launch()
+        logger.error(f"An error occurred: {e}")
+        yield f"An error occurred: {e}", "{}"
+
+with gr.Blocks() as demo:
+    gr.Markdown("# Backend Processing Logs")
+    
+    with gr.Row():
+        with gr.Column(scale=1):
+            tickers_input = gr.Textbox(lines=10, label="Enter Tickers (one per line)")
+            start_button = gr.Button("Start Processing")
+        
+        with gr.Column(scale=2):
+            logs_output = gr.Textbox(label="Logs", lines=15, interactive=False)
+            json_output = gr.JSON(label="Processed JSON Output")
+
+    start_button.click(
+        fn=run_backend_processing,
+        inputs=[tickers_input],
+        outputs=[logs_output, json_output]
+    )
+
+if __name__ == "__main__":
+    demo.launch()