diff --git "a/app.py" "b/app.py" --- "a/app.py" +++ "b/app.py" @@ -2,10 +2,11 @@ import pandas as pd import requests import uvicorn import asyncio +import re from datetime import timedelta from fastapi import FastAPI, Body from fastapi.responses import HTMLResponse, JSONResponse -from typing import Optional, List +from typing import Optional, Set, List, Dict # ========================================== # 1. CONFIGURATION @@ -35,6 +36,7 @@ def sb_get(endpoint: str, params: dict = None): return [] def sb_rpc(func_name: str, params: dict = None): + """Executes a Supabase Postgres Function (RPC) for Zero Egress""" if params is None: params = {} try: @@ -46,8 +48,119 @@ def sb_rpc(func_name: str, params: dict = None): print(f"DB Error RPC ({func_name}): {e}") return [] +def extract_variables_from_formulas(series_list: list) -> Set[str]: + """Parse formulas to extract required variables.""" + required = {'price'} + variable_map = { + '$P': 'price', '$V': 'volume', '$OI': 'oi', + '$B': 'buy_qty', '$S': 'sell_qty', + '$CB': 'cb', '$CS': 'cs', '$PB': 'pb', '$PS': 'ps' + } + for s in series_list: + formula = s.get('formula', '') + for var, col in variable_map.items(): + if var in formula: + required.add(col) + return required + +# ========================================== +# 3. ADVANCED MODE - STRIKE FILTERING LOGIC +# ========================================== +def process_advanced_data(raw_records: List[Dict], atm_range: int = 5) -> List[Dict]: + """ + Process raw strike-wise data for Advanced Mode. + + Logic: + - For each timestamp, calculate ATM (strike closest to ref_price) + - Filter: 5 ITM + 5 OTM (where ATM is included in 5 OTM) + - For CE: ITM = strikes BELOW ATM, OTM = ATM + strikes ABOVE ATM + - For PE: ITM = strikes ABOVE ATM, OTM = ATM + strikes BELOW ATM + - Aggregate CB, CS, PB, PS from filtered strikes + """ + if not raw_records: + return [] + + # Convert to DataFrame + df = pd.DataFrame(raw_records) + df['tick_ts'] = pd.to_datetime(df['tick_ts']) + + results = [] + + # Group by timestamp + for ts, group in df.groupby('tick_ts'): + ref_price = group['ref_price'].iloc[0] + ref_b = group['ref_b'].iloc[0] + ref_s = group['ref_s'].iloc[0] + + if ref_price <= 0: + continue + + # Get unique strikes and find ATM (closest to ref_price) + unique_strikes = group['strike_price'].unique() + atm_strike = min(unique_strikes, key=lambda x: abs(x - ref_price)) + + # Separate CE and PE data + ce_data = group[group['opt_type'] == 'CE'].copy() + pe_data = group[group['opt_type'] == 'PE'].copy() + + # For CE: + # ITM = strikes BELOW ATM (call has intrinsic value when spot > strike) + # OTM = ATM and strikes ABOVE ATM + ce_itm = ce_data[ce_data['strike_price'] < atm_strike].nlargest(atm_range, 'strike_price') + ce_otm = ce_data[ce_data['strike_price'] >= atm_strike].nsmallest(atm_range, 'strike_price') + ce_filtered = pd.concat([ce_itm, ce_otm]).drop_duplicates() + + # For PE: + # ITM = strikes ABOVE ATM (put has intrinsic value when spot < strike) + # OTM = ATM and strikes BELOW ATM + pe_itm = pe_data[pe_data['strike_price'] > atm_strike].nsmallest(atm_range, 'strike_price') + pe_otm = pe_data[pe_data['strike_price'] <= atm_strike].nlargest(atm_range, 'strike_price') + pe_filtered = pd.concat([pe_itm, pe_otm]).drop_duplicates() + + # Aggregate using LAST value (take the last non-zero value, or last value) + def get_last_value(series): + """Get last valid (non-zero if possible) value from series""" + non_zero = series[series > 0] + if len(non_zero) > 0: + return non_zero.iloc[-1] + return series.iloc[-1] if len(series) > 0 else 0 + + # Sum across filtered strikes for each metric + # CB = Call Buy = sum of 'b' for CE + cb = ce_filtered['b'].sum() if len(ce_filtered) > 0 else 0 + # CS = Call Sell = sum of 's' for CE + cs = ce_filtered['s'].sum() if len(ce_filtered) > 0 else 0 + # PB = Put Buy = sum of 'b' for PE + pb = pe_filtered['b'].sum() if len(pe_filtered) > 0 else 0 + # PS = Put Sell = sum of 's' for PE + ps = pe_filtered['s'].sum() if len(pe_filtered) > 0 else 0 + + # Volume and OI - sum all + vol = (ce_filtered['v'].sum() + pe_filtered['v'].sum()) if (len(ce_filtered) > 0 or len(pe_filtered) > 0) else 0 + oi = (ce_filtered['o'].sum() + pe_filtered['o'].sum()) if (len(ce_filtered) > 0 or len(pe_filtered) > 0) else 0 + + results.append({ + 'tick_ts': ts, + 'price': ref_price, + 'b_qty': ref_b, + 's_qty': ref_s, + 'volume': vol, + 'oi': oi, + 'cb': cb, + 'cs': cs, + 'pb': pb, + 'ps': ps, + 'atm_strike': atm_strike, + 'strikes_included': { + 'ce': ce_filtered['strike_price'].tolist(), + 'pe': pe_filtered['strike_price'].tolist() + } + }) + + return results + # ========================================== -# 3. API ROUTES - METADATA +# 4. API ROUTES - METADATA # ========================================== @app.get("/") @@ -73,7 +186,7 @@ def get_auto_config(date: str, root: str): root = root.upper() 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] if not valid_expiries: valid_expiries = all_expiries @@ -109,7 +222,7 @@ 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 = [] for i in instruments: @@ -117,11 +230,11 @@ def get_instruments(date: str, root: str, expiry: str): options.append(i) else: spot_fut.append(i) - + return JSONResponse({"spot_fut": sorted(spot_fut), "options": sorted(options)}) # ========================================== -# 4. API ROUTES - DATA FETCHING +# 5. API ROUTES - DATA FETCHING (RPC DRIVEN WITH CHUNKING) # ========================================== @app.post("/api/fetch_series") @@ -130,35 +243,34 @@ async def fetch_series( root: str = Body(...), expiry: str = Body(...), instrument: str = Body(""), - timeframe: str = Body(...), + timeframe: str = Body(...), start_time: str = Body(...), end_time: str = Body(...), mode: str = Body("normal"), atm_range: int = Body(5), - required_vars: List[str] = Body(default=None) + 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") - - # Default: all variables if not specified - if required_vars is None: - required_vars = ["P", "B", "S", "CB", "CS", "PB", "PS", "V", "OI"] - - # Always include P (price) - required_vars = list(set(required_vars) | {"P"}) - + + # Extract required variables from formulas + required_cols = extract_variables_from_formulas(series) + if mode == "advanced": + # 1. Fetch available instruments 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": [], "P": []}) - + return JSONResponse({"error": f"No data found for expiry {expiry}", "labels":[], "P":[]}) + + # 2. Intelligent Default Selection for reference instrument 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: @@ -167,19 +279,18 @@ async def fetch_series( 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] - rpc_name = "get_advanced_chart_data" + rpc_name = "get_advanced_raw_data" base_params = { "p_root": root, "p_expiry": expiry, - "p_ref_instrument": ref_instrument, - "p_atm_range": atm_range, - "p_strike_map": {} + "p_ref_instrument": ref_instrument } - + else: + # Normal fetching if not instrument: - return JSONResponse({"error": "No Instrument Selected.", "labels": [], "P": []}) - + return JSONResponse({"error": "No Instrument Selected.", "labels": [], "P":[]}) + rpc_name = "get_normal_chart_data" base_params = { "p_root": root, @@ -188,96 +299,110 @@ async def fetch_series( } # ========================================================= - # CHUNKING: 2-hour chunks fetched in parallel + # CHUNKING LOGIC: Bypass Supabase DB/Timeout Limits # ========================================================= 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() 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=2) - timedelta(seconds=1) + 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=2) - + 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": [], "P": []}) + return JSONResponse({"error": "No valid data found in range.", "labels":[], "P":[]}) # ========================================================= # DATA PROCESSING # ========================================================= - df = pd.DataFrame(all_records) - - df.rename(columns={ - "tick_ts": "ts", - "b_qty": "B", - "s_qty": "S", - "vol": "V", - "oi": "OI", - "cb": "CB", - "cs": "CS", - "pb": "PB", - "ps": "PS", - "price": "P" - }, inplace=True) - + if mode == "advanced": + # Process raw strike-wise data + processed_records = process_advanced_data(all_records, atm_range) + if not processed_records: + return JSONResponse({"error": "No valid data after processing.", "labels":[], "P":[]}) + df = pd.DataFrame(processed_records) + else: + df = pd.DataFrame(all_records) + + # Standardize column names + if 'tick_ts' in df.columns: + df.rename(columns={"tick_ts": "ts"}, inplace=True) + if 'b_qty' not in df.columns and 'b' in df.columns: + df.rename(columns={"b": "buy_qty"}, inplace=True) + elif 'b_qty' in df.columns: + df.rename(columns={"b_qty": "buy_qty"}, inplace=True) + if 's_qty' not in df.columns and 's' in df.columns: + df.rename(columns={"s": "sell_qty"}, inplace=True) + elif 's_qty' in df.columns: + df.rename(columns={"s_qty": "sell_qty"}, inplace=True) + if 'vol' in df.columns: + df.rename(columns={"vol": "volume"}, inplace=True) + df['ts'] = pd.to_datetime(df['ts']) df.drop_duplicates(subset=['ts'], inplace=True) df.set_index('ts', inplace=True) df.sort_index(inplace=True) - - # Resample + + # Resample Timeframes tf_map = { - "1min": "1min", "3min": "3min", "5min": "5min", + "1min": "1min", "3min": "3min", "5min": "5min", "15min": "15min", "30min": "30min", "1hour": "1h" } panda_tf = tf_map.get(timeframe, "1min") - - # KEY FIX: Use 'last' for quantity fields — these are cumulative running totals. - # The LAST value in any period is the most current/accurate snapshot. + + # CORRECTED: Use LAST value for all quantity columns (not mean) agg_dict = { - "P": "last", - "V": "sum", # Volume: summing new volume per candle makes sense - "OI": "last", # OI: last value - "B": "last", # Buy qty cumulative total → last value - "S": "last", # Sell qty cumulative total → last value - "CB": "last", # Call Buy → last value - "CS": "last", # Call Sell → last value - "PB": "last", # Put Buy → last value - "PS": "last", # Put Sell → last value + "price": "last", + "volume": "sum", + "oi": "last", + "buy_qty": "last", + "sell_qty": "last", + "cb": "last", + "cs": "last", + "pb": "last", + "ps": "last" } - - existing_agg = {k: v for k, v in agg_dict.items() if k in df.columns} - resampled = df.resample(panda_tf).agg(existing_agg).ffill().fillna(0) - - # Build response: only include requested variables (reduces egress) + + existing_cols = {k: v for k, v in agg_dict.items() if k in df.columns} + resampled = df.resample(panda_tf).agg(existing_cols).ffill().fillna(0) + response_data = { - "labels": resampled.index.strftime('%H:%M').tolist() + "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 [], + "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 [] } - - for var_key in ["P", "B", "S", "V", "OI", "CB", "CS", "PB", "PS"]: - if var_key in required_vars and var_key in resampled.columns: - response_data[var_key] = [round(float(v), 2) for v in resampled[var_key].tolist()] - # Omit entirely if not requested — JS handles missing keys gracefully - + return JSONResponse(response_data) # ========================================== -# 5. FRONTEND TEMPLATE +# 6. FRONTEND TEMPLATE # ========================================== HTML_TEMPLATE = """ @@ -290,31 +415,40 @@ HTML_TEMPLATE = """ @@ -465,15 +588,15 @@ HTML_TEMPLATE = """
DepthChain PRO
- -
- +
+ -
+
-
+
@@ -487,7 +610,7 @@ HTML_TEMPLATE = """
- +