import os import math import json import asyncio import httpx import pandas as pd import threading import urllib.parse from datetime import datetime, timedelta, timezone from typing import Optional, List, Dict, Any from fastapi import FastAPI, Request from fastapi.responses import HTMLResponse, JSONResponse from supabase import create_client, Client import uvicorn # ========================================== # 1. SUPABASE & APP CONFIGURATION (SECRETS) # ========================================== try: creds = json.loads(os.environ.get("SUPABASE_CREDS", "{}")) SUPABASE_URL = creds.get("url", "") SUPABASE_KEY = creds.get("key", "") except Exception: SUPABASE_URL = os.environ.get("SUPABASE_URL", "") SUPABASE_KEY = os.environ.get("SUPABASE_KEY", "") app = FastAPI() try: supabase: Client = create_client(SUPABASE_URL, SUPABASE_KEY) except Exception as e: print(f"Warning: Supabase client failed to initialize. Check Secrets. Error: {e}") supabase = None # ========================================== # 2. SYSTEM STATE & MASTER DATA CACHE # ========================================== SYSTEM_STATUS = {"ready": False, "message": "Booting..."} NIFTY_MASTER: Dict[str, Dict[str, Dict[float, str]]] = {} EXPIRIES_LIST: List[str] = [] SPOT_SYMBOL = "NSE:NIFTY50-INDEX" def get_fyers_token() -> str: if not supabase: raise Exception("Supabase not initialized.") res = supabase.table('apps').select('*').eq('broker', 'FYERS').eq('is_enabled', True).order('updated_at', desc=True).limit(1).execute() if res.data and res.data[0].get('access_token'): return f"{res.data[0]['app_id']}:{res.data[0]['access_token']}" raise Exception("No active Fyers token found in database.") def update_master_db(): global SYSTEM_STATUS, NIFTY_MASTER, EXPIRIES_LIST try: SYSTEM_STATUS["message"] = "Downloading NSE Master Data..." headers = {"User-Agent": "Mozilla/5.0"} url_nse = "https://public.fyers.in/sym_details/NSE_FO.csv" df = pd.read_csv(url_nse, header=None, on_bad_lines='skip', storage_options=headers) df_nifty = df[(df[13].astype(str) == 'NIFTY') & (df[2] == 14)] temp_master = {} unique_exp = set() for _, row in df_nifty.iterrows(): desc = str(row[1]) exp_epoch = int(row[8]) sym_ticker = str(row[9]) exp_date = datetime.fromtimestamp(exp_epoch, tz=timezone.utc).strftime('%Y-%m-%d') parts = desc.strip().split(' ') if len(parts) >= 2 and parts[-1] in ['CE', 'PE']: try: strike = float(parts[-2].replace(',', '')) opt_type = parts[-1] if exp_date not in temp_master: temp_master[exp_date] = {'CE': {}, 'PE': {}} temp_master[exp_date][opt_type][strike] = sym_ticker unique_exp.add(exp_date) except Exception: continue NIFTY_MASTER = temp_master EXPIRIES_LIST = sorted(list(unique_exp)) SYSTEM_STATUS["ready"] = True SYSTEM_STATUS["message"] = "Ready" except Exception as e: SYSTEM_STATUS["message"] = f"Master Error: {str(e)}" @app.on_event("startup") def startup_event(): threading.Thread(target=update_master_db).start() # ========================================== # 3. CORE LOGIC & THEORETICAL CALCS # ========================================== async def fetch_history(client: httpx.AsyncClient, token: str, symbol: str, date: str, semaphore: asyncio.Semaphore) -> tuple: encoded_symbol = urllib.parse.quote(symbol) url = f"https://api-t1.fyers.in/data/history?symbol={encoded_symbol}&resolution=1&date_format=1&range_from={date}&range_to={date}&cont_flag=0" headers = {"Authorization": token} async with semaphore: await asyncio.sleep(0.35) for attempt in range(4): try: resp = await client.get(url, headers=headers, timeout=15.0) data = resp.json() if data.get('s') == 'error': if "limit" in str(data).lower(): await asyncio.sleep(1.5 * (attempt + 1)) continue return symbol, {} price_map = {} if data.get('s') == 'ok' and 'candles' in data: for candle in data['candles']: price_map[int(candle[0])] = { "o": float(candle[1]), "h": float(candle[2]), "l": float(candle[3]), "c": float(candle[4]) } return symbol, price_map except Exception: await asyncio.sleep(1.0 * (attempt + 1)) return symbol, {} def forward_fill_prices(price_map: dict, ep_list: list) -> dict: filled = {} last_close = 0.0 for ep in ep_list: if ep in price_map: filled[ep] = price_map[ep] last_close = price_map[ep]['c'] else: # Flatten empty gaps so they don't corrupt HH/LL ranges filled[ep] = {"o": last_close, "h": last_close, "l": last_close, "c": last_close} return filled def get_tf_candle(price_map: dict, end_ep: int, tf_mins: int, fallback_val: float) -> dict: epochs = [end_ep - (tf_mins * 60) + (i * 60) for i in range(1, tf_mins + 1)] valid_candles = [] for e in epochs: if e in price_map and price_map[e].get('c', 0) > 0: valid_candles.append(price_map[e]) if not valid_candles: c = price_map.get(end_ep, {}).get("c", fallback_val) return {"o": c, "h": c, "l": c, "c": c} O = valid_candles[0]['o'] H = max(c['h'] for c in valid_candles) L = min(c['l'] for c in valid_candles) C = valid_candles[-1]['c'] return {"o": O, "h": H, "l": L, "c": C} def calc_diff(mode: str, curr: dict, prev: dict, formula: str, c_sign: str) -> float: O, H, L, C = curr.get('o', 0.0), curr.get('h', 0.0), curr.get('l', 0.0), curr.get('c', 0.0) O_prev, H_prev, L_prev, C_prev = prev.get('o', 0.0), prev.get('h', 0.0), prev.get('l', 0.0), prev.get('c', 0.0) if mode == "1": # HH/LL of TF mag = H - L sign_val = 1 if C >= O else -1 return sign_val * mag elif mode == "2": # TF Open-Close absolute return C - O elif mode == "3": # T2 Close - T1 Close return C - C_prev elif mode == "4": # Extremes relative to T1 Close if C > C_prev: val = H elif C < C_prev: val = L else: val = C return val - C_prev elif mode == "5": # Custom Formula allowed_vars = { 'O': O, 'H': H, 'L': L, 'C': C, 'O_prev': O_prev, 'H_prev': H_prev, 'L_prev': L_prev, 'C_prev': C_prev, 'abs': abs, 'max': max, 'min': min } try: res = float(eval(formula, {"__builtins__": None}, allowed_vars)) if c_sign == "candle": mag = abs(res) sign_val = 1 if C >= O else -1 return sign_val * mag return res except Exception: return 0.0 return C - O # ========================================== # 4. API ENDPOINTS # ========================================== @app.get("/api/status") def get_status(): return JSONResponse(SYSTEM_STATUS) @app.get("/api/expiries") def get_expiries(): if not SYSTEM_STATUS["ready"]: return JSONResponse({"s": "error", "message": "Master Data Initializing..."}) return JSONResponse({"s": "ok", "expiries": EXPIRIES_LIST}) @app.get("/api/heatmap") async def generate_heatmap( request: Request, date: str, expiry: str, tfs: str = "1,3,15", atm_range: int = 10, start: str = "09:15", end: str = "15:30", method: str = "1", formula: str = "(H-L)", sign: str = "formula", divisor: str = "true", strike_mode: str = "itm" ): if not SYSTEM_STATUS["ready"]: return JSONResponse({"s": "error", "message": "System Booting... Please wait."}) try: token = get_fyers_token() except Exception as e: return JSONResponse({"s": "error", "message": str(e)}) timeframes = [int(x.strip()) for x in tfs.split(",") if x.strip().isdigit()] use_divisor = (divisor.lower() == "true") ist_tz = timezone(timedelta(hours=5, minutes=30)) start_dt = datetime.strptime(f"{date} {start}:00", "%Y-%m-%d %H:%M:%S").replace(tzinfo=ist_tz) end_dt = datetime.strptime(f"{date} {end}:00", "%Y-%m-%d %H:%M:%S").replace(tzinfo=ist_tz) total_mins = int((end_dt - start_dt).total_seconds() // 60) ep_list = [int((start_dt + timedelta(minutes=i)).timestamp()) for i in range(total_mins + 1)] async with httpx.AsyncClient() as client: _, raw_spot = await fetch_history(client, token, SPOT_SYMBOL, date, asyncio.Semaphore(1)) if not raw_spot: return JSONResponse({"s": "error", "message": f"No Spot data found. Date may be a holiday or wrong format."}) spot_map = forward_fill_prices(raw_spot, ep_list) required_symbols = set() for ep in ep_list: spot_price = spot_map.get(ep, {}).get("c", 0) if spot_price > 0: atm = int(spot_price // 50) * 50 strikes = [atm - (i * 50) for i in range(atm_range - 1, -1, -1)] + [atm + (i * 50) for i in range(1, atm_range + 1)] for k in strikes: if expiry in NIFTY_MASTER: if k in NIFTY_MASTER[expiry]['CE']: required_symbols.add(NIFTY_MASTER[expiry]['CE'][k]) if k in NIFTY_MASTER[expiry]['PE']: required_symbols.add(NIFTY_MASTER[expiry]['PE'][k]) semaphore = asyncio.Semaphore(4) tasks = [fetch_history(client, token, sym, date, semaphore) for sym in required_symbols] results = await asyncio.gather(*tasks) opt_data = {sym: forward_fill_prices(raw, ep_list) for sym, raw in results} matrix = [] for ep in ep_list: time_str = datetime.fromtimestamp(ep, tz=ist_tz).strftime('%H:%M') spot_t2_info = spot_map.get(ep, {"o": 0, "h": 0, "l": 0, "c": 0}) spot_t2_c = spot_t2_info["c"] if spot_t2_c == 0: continue spot_min_diff = spot_t2_c - spot_t2_info["o"] atm = int(spot_t2_c // 50) * 50 strikes = [atm - (i * 50) for i in range(atm_range - 1, -1, -1)] + [atm + (i * 50) for i in range(1, atm_range + 1)] row_data = {"time": time_str, "spot_min_diff": round(spot_min_diff, 2), "tfs": {}} for tf in timeframes: ep_t1 = ep - (tf * 60) if ep_t1 < start_dt.timestamp(): row_data["tfs"][str(tf)] = None continue spot_curr = get_tf_candle(spot_map, ep, tf, spot_t2_c) spot_prev = get_tf_candle(spot_map, ep_t1, tf, spot_t2_c) spot_diff = calc_diff(method, spot_curr, spot_prev, formula, sign) if use_divisor: abs_spot_diff = abs(spot_diff) safe_spot_diff = abs_spot_diff if abs_spot_diff != 0.0 else 0.05 else: safe_spot_diff = 1.0 sum_c_diff, sum_p_diff = 0.0, 0.0 strike_details = [] for k in strikes: # Quadrant Logic Mode ce_valid = True pe_valid = True if strike_mode == "itm": ce_valid = (k <= atm) pe_valid = (k >= atm) elif strike_mode == "otm": ce_valid = (k > atm) pe_valid = (k < atm) ce_sym = NIFTY_MASTER[expiry]['CE'].get(k) if expiry in NIFTY_MASTER else None pe_sym = NIFTY_MASTER[expiry]['PE'].get(k) if expiry in NIFTY_MASTER else None ce_curr = get_tf_candle(opt_data[ce_sym], ep, tf, 0.0) if ce_sym and ce_sym in opt_data else {"o":0,"h":0,"l":0,"c":0} ce_prev = get_tf_candle(opt_data[ce_sym], ep_t1, tf, 0.0) if ce_sym and ce_sym in opt_data else {"o":0,"h":0,"l":0,"c":0} pe_curr = get_tf_candle(opt_data[pe_sym], ep, tf, 0.0) if pe_sym and pe_sym in opt_data else {"o":0,"h":0,"l":0,"c":0} pe_prev = get_tf_candle(opt_data[pe_sym], ep_t1, tf, 0.0) if pe_sym and pe_sym in opt_data else {"o":0,"h":0,"l":0,"c":0} c_diff = calc_diff(method, ce_curr, ce_prev, formula, sign) if ce_valid else 0.0 p_diff = calc_diff(method, pe_curr, pe_prev, formula, sign) if pe_valid else 0.0 c_ratio = c_diff / safe_spot_diff p_ratio = p_diff / safe_spot_diff sum_c_diff += c_diff sum_p_diff += p_diff strike_details.append({ "k": k, "c1": round(ce_prev['c'], 2), "c2": round(ce_curr['c'], 2), "cd": round(c_diff, 2), "cr": round(c_ratio, 4), "cv": ce_valid, "p1": round(pe_prev['c'], 2), "p2": round(pe_curr['c'], 2), "pd": round(p_diff, 2), "pr": round(p_ratio, 4), "pv": pe_valid }) emp_ce_delta = sum_c_diff / safe_spot_diff emp_pe_delta = sum_p_diff / safe_spot_diff net_emp_delta = emp_ce_delta - emp_pe_delta # Identify Anomaly: Spot and Net Delta move in OPPOSITE directions! is_anomaly = (spot_diff > 0 and net_emp_delta < 0) or (spot_diff < 0 and net_emp_delta > 0) row_data["tfs"][str(tf)] = { "tf_name": f"{tf} Min", "method": method, "atm_idx": atm_range - 1, "net_delta": round(net_emp_delta, 4), "ce_delta": round(emp_ce_delta, 4), "pe_delta": round(emp_pe_delta, 4), "t1_spot": round(spot_prev['c'], 2), "t2_spot": round(spot_curr['c'], 2), "spot_diff": round(spot_diff, 2), "ce_diff_total": round(sum_c_diff, 2), "pe_diff_total": round(sum_p_diff, 2), "strike_details": strike_details, "atm_range": atm_range, "is_anomaly": is_anomaly } matrix.append(row_data) return JSONResponse({"s": "ok", "matrix": matrix, "timeframes": timeframes}) # ========================================== # 5. FRONTEND HTML TEMPLATE # ========================================== @app.get("/") def serve_ui(): return HTMLResponse(HTML_TEMPLATE) HTML_TEMPLATE = """ Empirical Delta Terminal - Ultra Compact
TimeSpot 1m ΔMatrix generates here...
""" if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=7860)