Update app.py

app.py

@@ -1,14 +1,11 @@
 import pandas as pd
 import requests
 import uvicorn
-import json
 import asyncio
-import re
-from datetime import datetime, timedelta
-from fastapi import FastAPI, Body, HTTPException, Query
+from datetime import timedelta
+from fastapi import FastAPI, Body
 from fastapi.responses import HTMLResponse, JSONResponse
-from fastapi.staticfiles import StaticFiles
-from typing import Optional, List, Dict, Any
+from typing import Optional
 
 # ==========================================
 # 1. CONFIGURATION
@@ -37,8 +34,10 @@ def sb_get(endpoint: str, params: dict = None):
         print(f"DB Error GET ({endpoint}): {e}")
         return[]
 
-def sb_rpc(func_name: str, params: dict = {}):
+def sb_rpc(func_name: str, params: dict = None):
     """Executes a Supabase Postgres Function (RPC) for Zero Egress"""
+    if params is None:
+        params = {}
     try:
         url = f"{SB_URL}/rest/v1/rpc/{func_name}"
         r = requests.post(url, headers=HEADERS, json=params)
@@ -76,7 +75,7 @@ def get_auto_config(date: str, root: str):
     exps_data = sb_rpc("get_expiries_for_root", {"p_date": date, "p_root": root})
     all_expiries = [row['expiry_date'] for row in exps_data if row['expiry_date'] != 'MARKET']
     
-    valid_expiries =[e for e in all_expiries if e >= date]
+    valid_expiries = [e for e in all_expiries if e >= date]
     if not valid_expiries:
         valid_expiries = all_expiries
 
@@ -123,11 +122,11 @@ def get_instruments(date: str, root: str, expiry: str):
     return JSONResponse({"spot_fut": sorted(spot_fut), "options": sorted(options)})
 
 # ==========================================
-# 4. API ROUTES - DATA FETCHING (RPC DRIVEN)
+# 4. API ROUTES - DATA FETCHING (RPC DRIVEN WITH CHUNKING)
 # ==========================================
 
 @app.post("/api/fetch_series")
-def fetch_series(
+async def fetch_series(
     date: str = Body(...),
     root: str = Body(...),
     expiry: str = Body(...),
@@ -139,8 +138,8 @@ def fetch_series(
     atm_range: int = Body(5)
 ):
     root = root.upper()
-    t_start = f"{date} {start_time}:00"
-    t_end = f"{date} {end_time}:59"
+    t_start_dt = pd.to_datetime(f"{date} {start_time}:00")
+    t_end_dt = pd.to_datetime(f"{date} {end_time}:59")
     
     if mode == "advanced":
         # 1. Fetch available instruments (Lightweight metadata fetch)
@@ -150,59 +149,83 @@ def fetch_series(
         if not instruments:
             return JSONResponse({"error": f"No data found for expiry {expiry}", "labels":[], "P":[]})
         
-        # 2. Intelligent Default Selection (Spot > Future > First Available)
+        # 2. Intelligent Default Selection
         ref_instrument = instrument
-        
-        # If user left instrument blank, find the best reference (Index or Future)
         if not ref_instrument:
-            spot_keys = [k for k in instruments if k.endswith('-INDEX') or k.endswith('-EQ')]
-            fut_keys = [k for k in instruments if 'FUT' in k]
+            spot_keys =[k for k in instruments if k.endswith('-INDEX') or k.endswith('-EQ')]
+            fut_keys =[k for k in instruments if 'FUT' in k]
             
             if spot_keys:
                 ref_instrument = spot_keys[0]
             elif fut_keys:
                 ref_instrument = fut_keys[0]
             else:
-                # Fallback: Find first key that is NOT an option
-                others = [k for k in instruments if not k.endswith('CE') and not k.endswith('PE')]
+                others =[k for k in instruments if not k.endswith('CE') and not k.endswith('PE')]
                 ref_instrument = others[0] if others else instruments[0]
 
-        # 3. Call Postgres (Logic is now 100% inside DB)
-        # We pass empty dict for p_strike_map as it is handled in SQL now
-        params = {
+        rpc_name = "get_advanced_chart_data"
+        base_params = {
             "p_root": root,
             "p_expiry": expiry,
-            "p_start_time": t_start,
-            "p_end_time": t_end,
             "p_ref_instrument": ref_instrument,
             "p_atm_range": atm_range,
-            "p_strike_map": {} 
+            "p_strike_map": {}  # Sent but ignored by SQL safely
         }
-        records = sb_rpc("get_advanced_chart_data", params)
     
     else:
-        # Normal fetching (One specific instrument)
+        # Normal fetching
         if not instrument:
             return JSONResponse({"error": "No Instrument Selected.", "labels": [], "P":[]})
             
-        params = {
+        rpc_name = "get_normal_chart_data"
+        base_params = {
             "p_root": root,
             "p_expiry": expiry,
-            "p_start_time": t_start,
-            "p_end_time": t_end,
             "p_instrument": instrument
         }
-        records = sb_rpc("get_normal_chart_data", params)
 
-    if not records or (isinstance(records, dict) and records.get("error")):
-         # Handle case where Time range exists but maybe reference instrument had no ticks
-         return JSONResponse({"error": "No valid data found in range.", "labels":[], "P":[]})
+    # =========================================================
+    # CHUNKING LOGIC: Bypass Supabase DB/Timeout Limits
+    # Fetches 2-hour chunks concurrently!
+    # =========================================================
+    async def fetch_chunk(start_dt, end_dt):
+        params = base_params.copy()
+        params["p_start_time"] = start_dt.strftime("%Y-%m-%d %H:%M:%S")
+        params["p_end_time"] = end_dt.strftime("%Y-%m-%d %H:%M:%S")
+        
+        loop = asyncio.get_event_loop()
+        # Execute synchronous requests.post in a background thread to allow parallel execution
+        return await loop.run_in_executor(None, sb_rpc, rpc_name, params)
+
+    tasks =[]
+    curr = t_start_dt
+    chunk_hours = 2  # Safe size for Postgres processing limit
+
+    while curr <= t_end_dt:
+        nxt = curr + timedelta(hours=chunk_hours) - timedelta(seconds=1)
+        if nxt > t_end_dt:
+            nxt = t_end_dt
+        
+        tasks.append(fetch_chunk(curr, nxt))
+        curr += timedelta(hours=chunk_hours)
+        
+    # Execute all chunks in parallel
+    chunk_results = await asyncio.gather(*tasks)
+    
+    all_records =[]
+    for res in chunk_results:
+        if isinstance(res, list):
+            all_records.extend(res)
 
-    df = pd.DataFrame(records)
-    if df.empty:
-        return JSONResponse({"error": "Dataframe Empty", "labels": [], "P":[]})
+    if not all_records:
+         return JSONResponse({"error": "No valid data found in range.", "labels":[], "P":[]})
 
-    # Rename columns to match standard
+    # =========================================================
+    # DATA PROCESSING & AGGREGATION
+    # =========================================================
+    df = pd.DataFrame(all_records)
+    
+    # Standardize output
     df.rename(columns={
         "tick_ts": "ts",
         "b_qty": "buy_qty",
@@ -211,9 +234,13 @@ def fetch_series(
     }, inplace=True)
     
     df['ts'] = pd.to_datetime(df['ts'])
+    
+    # Deduplicate boundaries just in case overlapping seconds occurred
+    df.drop_duplicates(subset=['ts'], inplace=True)
     df.set_index('ts', inplace=True)
+    df.sort_index(inplace=True)
     
-    # Resample Logic
+    # Resample Timeframes accurately in Pandas 
     tf_map = {
         "1min": "1min", "3min": "3min", "5min": "5min", 
         "15min": "15min", "30min": "30min", "1hour": "1h"
@@ -221,7 +248,7 @@ def fetch_series(
     panda_tf = tf_map.get(timeframe, "1min")
     
     agg_dict = {
-        "price": "last", "volume": "sum", "oi": "last", # Vol should be sum for aggregations usually
+        "price": "last", "volume": "sum", "oi": "last",
         "buy_qty": "mean", "sell_qty": "mean",
         "cb": "mean", "cs": "mean", "pb": "mean", "ps": "mean"
     }
@@ -234,14 +261,14 @@ def fetch_series(
     response_data = {
         "labels": resampled.index.strftime('%H:%M').tolist(),
         "P": resampled['price'].tolist() if 'price' in resampled else [],
-        "B": resampled['buy_qty'].tolist() if 'buy_qty' in resampled else [],
-        "S": resampled['sell_qty'].tolist() if 'sell_qty' in resampled else [],
+        "B": resampled['buy_qty'].tolist() if 'buy_qty' in resampled else[],
+        "S": resampled['sell_qty'].tolist() if 'sell_qty' in resampled else[],
         "V": resampled['volume'].tolist() if 'volume' in resampled else [],
         "OI": resampled['oi'].tolist() if 'oi' in resampled else [],
         "CB": resampled['cb'].tolist() if 'cb' in resampled else [],
         "CS": resampled['cs'].tolist() if 'cs' in resampled else [],
         "PB": resampled['pb'].tolist() if 'pb' in resampled else [],
-        "PS": resampled['ps'].tolist() if 'ps' in resampled else []
+        "PS": resampled['ps'].tolist() if 'ps' in resampled else[]
     }
     
     return JSONResponse(response_data)
@@ -499,7 +526,7 @@ HTML_TEMPLATE = """
                         </div>
                     </div>
 
-                    <!-- EXPLICIT INSTRUMENT SELECTION (VISIBLE IN BOTH MODES NOW) -->
+                    <!-- EXPLICIT INSTRUMENT SELECTION -->
                     <div class="form-section" id="cfgInstrumentContainer">
                         <label id="lblInst">Instrument</label>
                         <div class="input-wrapper">
@@ -627,7 +654,7 @@ const CustomDD = {
         let html = '';
         let currentGroup = null;
         
-        let pool =[...this.data[id]];
+        let pool = [...this.data[id]];
         
         if(id === 'cfgRoot' && q === '') {
             const recentItems = this.recent.cfgRoot.map(r => ({val: r, group: 'Recent Search'}));
@@ -930,10 +957,10 @@ const ChartLogic = {
                     
                     // Frontend JS evaluates the line using the variables passed from the backend
                     let f = s.formula
-                        .replace(/\$CB/g, CB).replace(/\$CS/g, CS)
-                        .replace(/\$PB/g, PB).replace(/\$PS/g, PS)
-                        .replace(/\$OI/g, OI).replace(/\$P/g, P)
-                        .replace(/\$V/g, V).replace(/\$B/g, B).replace(/\$S/g, S);
+                        .replace(/\$CB/g, str(CB)).replace(/\$CS/g, str(CS))
+                        .replace(/\$PB/g, str(PB)).replace(/\$PS/g, str(PS))
+                        .replace(/\$OI/g, str(OI)).replace(/\$P/g, str(P))
+                        .replace(/\$V/g, str(V)).replace(/\$B/g, str(B)).replace(/\$S/g, str(S));
                         
                     try { return new Function('return '+f)(); } catch{ return null; }
                 });
@@ -1324,10 +1351,16 @@ const Config = {
     }
 };
 
+// Helper inside JS for string replacement in formulas
+function str(val) {
+    return val !== null && val !== undefined ? val.toString() : "0";
+}
+
 window.onload = () => App.init();
 </script>
 </body>
-</html>"""
+</html>
+"""
 
 if __name__ == "__main__":
     uvicorn.run(app, host="0.0.0.0", port=7860)