diff --git "a/app.py" "b/app.py" --- "a/app.py" +++ "b/app.py" @@ -1,3 +1,4 @@ +import re import pandas as pd import requests import uvicorn @@ -5,7 +6,7 @@ import asyncio from datetime import timedelta from fastapi import FastAPI, Body from fastapi.responses import HTMLResponse, JSONResponse -from typing import Optional, List +from typing import Optional # ========================================== # 1. CONFIGURATION @@ -22,18 +23,8 @@ HEADERS = { app = FastAPI() # ========================================== -# 2. HELPER FUNCTIONS +# 2. SUPABASE HELPERS # ========================================== -def sb_get(endpoint: str, params: dict = None): - try: - url = f"{SB_URL}/rest/v1{endpoint}" - r = requests.get(url, headers=HEADERS, params=params) - r.raise_for_status() - return r.json() - except Exception as e: - print(f"DB Error GET ({endpoint}): {e}") - return[] - def sb_rpc(func_name: str, params: dict = None): if params is None: params = {} @@ -44,10 +35,32 @@ def sb_rpc(func_name: str, params: dict = None): return r.json() except Exception as e: print(f"DB Error RPC ({func_name}): {e}") - return[] + return [] # ========================================== -# 3. API ROUTES - METADATA +# 3. VARIABLE EXTRACTION FROM FORMULAS +# ========================================== +# All valid variable names the user can use in formulas +ALL_FORMULA_VARS = ['CB', 'CS', 'PB', 'PS', 'B', 'S', 'V', 'OI', 'P'] + +def extract_needed_vars(series_list: list) -> set: + """ + Parse each series formula and collect the variable names actually used. + P is always included (it's the price axis reference). + Uses word-boundary matching so 'CB' doesn't match inside 'COMBO' etc. + """ + needed = {'P'} + for s in series_list: + formula = s.get('formula', '') if isinstance(s, dict) else str(s) + f_upper = formula.upper() + for v in ALL_FORMULA_VARS: + # Word boundary: variable must not be preceded or followed by A-Z + if re.search(r'(?= date] - if not valid_expiries: - valid_expiries = all_expiries - + valid_expiries = [e for e in all_expiries if e >= date] or all_expiries response = {"expiries": all_expiries, "current": None, "next": None} - def find_fut_rpc(exp): + def find_ref(exp): inst_data = sb_rpc("get_instruments_for_expiry", {"p_date": date, "p_root": root, "p_expiry": exp}) instruments = [r['instrument_name'] for r in inst_data] for i in instruments: @@ -88,869 +96,1263 @@ def get_auto_config(date: str, root: str): return instruments[0] if instruments else None if len(valid_expiries) > 0: - exp_c = valid_expiries[0] - response["current"] = {"expiry": exp_c, "instrument": find_fut_rpc(exp_c)} - + response["current"] = {"expiry": valid_expiries[0], "instrument": find_ref(valid_expiries[0])} if len(valid_expiries) > 1: - exp_n = valid_expiries[1] - response["next"] = {"expiry": exp_n, "instrument": find_fut_rpc(exp_n)} - + response["next"] = {"expiry": valid_expiries[1], "instrument": find_ref(valid_expiries[1])} return JSONResponse(response) @app.get("/api/expiries") def get_expiries(date: str, root: str): - root = root.upper() - data = sb_rpc("get_expiries_for_root", {"p_date": date, "p_root": root}) - exps =[row['expiry_date'] for row in data] - return JSONResponse(exps) + data = sb_rpc("get_expiries_for_root", {"p_date": date, "p_root": root.upper()}) + return JSONResponse([row['expiry_date'] for row in data]) @app.get("/api/instruments") def get_instruments(date: str, root: str, expiry: str): - root = root.upper() - data = sb_rpc("get_instruments_for_expiry", {"p_date": date, "p_root": root, "p_expiry": expiry}) - instruments =[row['instrument_name'] for row in data] - - spot_fut =[] - options =[] + data = sb_rpc("get_instruments_for_expiry", {"p_date": date, "p_root": root.upper(), "p_expiry": expiry}) + instruments = [row['instrument_name'] for row in data] + spot_fut, options = [], [] for i in instruments: - if i.endswith('CE') or i.endswith('PE'): - options.append(i) - else: - spot_fut.append(i) - + (options if i.endswith('CE') or i.endswith('PE') else spot_fut).append(i) return JSONResponse({"spot_fut": sorted(spot_fut), "options": sorted(options)}) # ========================================== -# 4. API ROUTES - DATA FETCHING (EGRESS OPTIMIZED) +# 5. DATA FETCHING # ========================================== @app.post("/api/fetch_series") async def fetch_series( - date: str = Body(...), - root: str = Body(...), - expiry: str = Body(...), - instrument: str = Body(""), - timeframe: str = Body(...), - start_time: str = Body(...), - end_time: str = Body(...), - mode: str = Body("normal"), - atm_range: int = Body(5), - req_vars: List[str] = Body([]) # EGRESS REDUCTION: Only requested logic passes through + date: str = Body(...), + root: str = Body(...), + expiry: str = Body(...), + instrument: str = Body(""), + timeframe: str = Body(...), + start_time: str = Body(...), + end_time: str = Body(...), + mode: str = Body("normal"), + atm_range: int = Body(5), + series: list = Body(default=[]) ): root = root.upper() - t_start_dt = pd.to_datetime(f"{date} {start_time}:00") - t_end_dt = pd.to_datetime(f"{date} {end_time}:59") - + t_start = pd.to_datetime(f"{date} {start_time}:00") + t_end = pd.to_datetime(f"{date} {end_time}:59") + + # Determine which variables the formulas actually need + needed_vars = extract_needed_vars(series) + + # --------------------------------------------------------------- + # Build RPC params depending on mode + # --------------------------------------------------------------- if mode == "advanced": - inst_data = sb_rpc("get_instruments_for_expiry", {"p_date": date, "p_root": root, "p_expiry": expiry}) + inst_data = sb_rpc("get_instruments_for_expiry", + {"p_date": date, "p_root": root, "p_expiry": expiry}) instruments = [r['instrument_name'] for r in inst_data] - if not instruments: return JSONResponse({"error": f"No data found for expiry {expiry}", "labels":[]}) - - ref_instrument = instrument - 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] - if spot_keys: ref_instrument = spot_keys[0] - elif fut_keys: ref_instrument = fut_keys[0] - else: ref_instrument = instruments[0] - - rpc_name = "get_advanced_chart_data" + if not instruments: + return JSONResponse({"error": f"No data for expiry {expiry}", "labels": [], "P": []}) + + # Pick best reference instrument: prefer INDEX, then FUT, then anything non-option + ref = instrument + if not ref: + for k in instruments: + if k.endswith('-INDEX') or k.endswith('-EQ'): + ref = k; break + if not ref: + for k in instruments: + if 'FUT' in k: + ref = k; break + if not ref: + others = [k for k in instruments if not k.endswith('CE') and not k.endswith('PE')] + ref = others[0] if others else instruments[0] + + rpc_name = "get_advanced_chart_data" base_params = { - "p_root": root, "p_expiry": expiry, "p_ref_instrument": ref_instrument, - "p_atm_range": atm_range, "p_strike_map": {} + "p_root": root, "p_expiry": expiry, + "p_ref_instrument": ref, + "p_atm_range": atm_range } + else: - if not instrument: return JSONResponse({"error": "No Instrument Selected.", "labels":[]}) - rpc_name = "get_normal_chart_data" - base_params = {"p_root": root, "p_expiry": expiry, "p_instrument": instrument} + # Normal mode + if not instrument: + return JSONResponse({"error": "No Instrument Selected.", "labels": [], "P": []}) + rpc_name = "get_normal_chart_data" + base_params = { + "p_root": root, "p_expiry": expiry, + "p_instrument": instrument + } + # --------------------------------------------------------------- + # CHUNKED FETCHING: split into 2-hour windows and run concurrently + # This avoids Supabase RPC timeouts on large ranges + # --------------------------------------------------------------- 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") + params = { + **base_params, + "p_start_time": start_dt.strftime("%Y-%m-%d %H:%M:%S"), + "p_end_time": end_dt.strftime("%Y-%m-%d %H:%M:%S"), + } loop = asyncio.get_event_loop() return await loop.run_in_executor(None, sb_rpc, rpc_name, params) - tasks =[] - curr = t_start_dt - chunk_hours = 2 - while curr <= t_end_dt: - nxt = curr + timedelta(hours=chunk_hours) - timedelta(seconds=1) - if nxt > t_end_dt: nxt = t_end_dt + tasks = [] + curr = t_start + while curr <= t_end: + nxt = min(curr + timedelta(hours=2) - timedelta(seconds=1), t_end) tasks.append(fetch_chunk(curr, nxt)) - curr += timedelta(hours=chunk_hours) - - chunk_results = await asyncio.gather(*tasks) - - all_records =[] - for res in chunk_results: - if isinstance(res, list): all_records.extend(res) - - if not all_records: return JSONResponse({"error": "No valid data found in range.", "labels":[]}) - - df = pd.DataFrame(all_records) - df.rename(columns={"tick_ts": "ts"}, inplace=True) - - df['ts'] = pd.to_datetime(df['ts']) + curr += timedelta(hours=2) + + chunks = await asyncio.gather(*tasks) + all_rows = [row for chunk in chunks if isinstance(chunk, list) for row in chunk] + + if not all_rows: + return JSONResponse({"error": "No data found in range.", "labels": [], "P": []}) + + # --------------------------------------------------------------- + # BUILD DATAFRAME + # --------------------------------------------------------------- + df = pd.DataFrame(all_rows) + df['ts'] = pd.to_datetime(df['tick_ts']) df.drop_duplicates(subset=['ts'], inplace=True) df.set_index('ts', inplace=True) df.sort_index(inplace=True) - - tf_map = {"1min": "1min", "3min": "3min", "5min": "5min", "15min": "15min", "30min": "30min", "1hour": "1h"} + + # --------------------------------------------------------------- + # RENAME COLUMNS: Normal mode only + # SQL returns raw 'b' and 's' for every instrument. + # We rename them to the semantically correct variable name here, + # based on what the instrument actually is. + # CE → b = CB (call buy), s = CS (call sell) + # PE → b = PB (put buy), s = PS (put sell) + # FUT/INDEX/EQ → b = B, s = S + # Advanced mode SQL already returns cb/cs/pb/ps directly. + # --------------------------------------------------------------- + if mode == "normal" and 'b' in df.columns and 's' in df.columns: + inst_upper = instrument.upper() + if inst_upper.endswith('CE'): + df.rename(columns={'b': 'cb', 's': 'cs'}, inplace=True) + elif inst_upper.endswith('PE'): + df.rename(columns={'b': 'pb', 's': 'ps'}, inplace=True) + else: + # FUT, INDEX, EQ — keep as B and S + df.rename(columns={'b': 'b_raw', 's': 's_raw'}, inplace=True) + df.rename(columns={'b_raw': 'b', 's_raw': 's'}, inplace=True) + + # --------------------------------------------------------------- + # RESAMPLE TO REQUESTED TIMEFRAME + # ALL buy/sell/oi values are CUMULATIVE SNAPSHOTS from the exchange + # (tot_buy_qty = total buy orders ever placed, always increasing). + # → Correct aggregation = LAST (last snapshot in the candle) + # Volume (vol_traded_today) is also cumulative → LAST + # --------------------------------------------------------------- + tf_map = {"1min": "1min", "3min": "3min", "5min": "5min", + "15min": "15min", "30min": "30min", "1hour": "1h"} panda_tf = tf_map.get(timeframe, "1min") - - # Strictly take the LAST cumulative Fyers variable output for the candle - agg_dict = { col: "last" for col in["price", "vol", "oi", "cb", "cs", "pb", "ps"] if col in df.columns } - - resampled = df.resample(panda_tf).agg(agg_dict).ffill().fillna(0) - - response_data = {"labels": resampled.index.strftime('%H:%M').tolist(), "P": resampled['price'].tolist() if 'price' in resampled else[]} - - # Maps what frontend asks for (e.g., 'V') to the database columns (e.g., 'vol') - col_map = {"V": "vol", "OI": "oi", "CB": "cb", "CS": "cs", "PB": "pb", "PS": "ps"} - for req in req_vars: - col = col_map.get(req.upper()) - if col and col in resampled.columns: - response_data[req.upper()] = resampled[col].tolist() - - return JSONResponse(response_data) + + agg_rules = {col: "last" for col in ['price', 'b', 's', 'cb', 'cs', 'pb', 'ps', 'vol', 'oi'] + if col in df.columns} + resampled = df.resample(panda_tf).agg(agg_rules).ffill().fillna(0) + + # --------------------------------------------------------------- + # BUILD PAYLOAD: only include variables actually used in formulas + # Map: dataframe column name → API response key → formula variable name + # --------------------------------------------------------------- + col_to_var = { + 'price': 'P', + 'b': 'B', # FUT/EQ normal mode + 's': 'S', # FUT/EQ normal mode + 'cb': 'CB', + 'cs': 'CS', + 'pb': 'PB', + 'ps': 'PS', + 'vol': 'V', + 'oi': 'OI', + } + + response = {"labels": resampled.index.strftime('%H:%M').tolist()} + for col, var in col_to_var.items(): + if col in resampled.columns and var in needed_vars: + response[var] = resampled[col].tolist() + elif var == 'P' and 'price' in resampled.columns: + # Price is always sent regardless + response['P'] = resampled['price'].tolist() + else: + response[var] = [] + + return JSONResponse(response) # ========================================== -# 5. FRONTEND TEMPLATE +# 6. FRONTEND # ========================================== - -HTML_TEMPLATE = """ - +HTML_TEMPLATE = """ - - DepthChain PRO - - - - + +DepthChain PRO + + + +
-
DepthChain PRO
-
- -
- -
-
-
-
- - -
- - +
DepthChain PRO
+
+ +
+ +
-
- -
- - - +
+ -