import pandas as pd import requests import uvicorn import json import time import os import threading from datetime import datetime from fastapi import FastAPI, Cookie, BackgroundTasks, Query from fastapi.responses import HTMLResponse, RedirectResponse, JSONResponse from typing import Optional # ========================================== # 1. CONFIGURATION # ========================================== # REPLACE THESE WITH YOUR UPSTOX CREDENTIALS API_KEY = "13c3ac18-b78f-4e5b-a6b8-d8f66de0b52c" API_SECRET = "qymj4zn0i1" REDIRECT_URI = "https://topsecrettraders-depthchainindiaup.hf.space/callback" CACHE_FILE = "upstox_master_cache.json" app = FastAPI() # ========================================== # 2. DATA LOADER (UPSTOX MASTER) # ========================================== MASTER_DB = {} SEARCH_INDEX = [] SYSTEM_STATUS = { "ready": False, "message": "Booting..." } UPDATE_LOCK = threading.Lock() # Manual mapping for Index Spot Keys (Safety Net) INDEX_MAP = { "NIFTY": "NSE_INDEX|Nifty 50", "BANKNIFTY": "NSE_INDEX|Nifty Bank", "FINNIFTY": "NSE_INDEX|Nifty Fin Service", "MIDCPNIFTY": "NSE_INDEX|NIFTY MID SELECT", "SENSEX": "BSE_INDEX|SENSEX", "BANKEX": "BSE_INDEX|BANKEX", "SENSEX50": "BSE_INDEX|SENSEX50" } def update_master_db(): global SYSTEM_STATUS, MASTER_DB, SEARCH_INDEX if not UPDATE_LOCK.acquire(blocking=False): return try: SYSTEM_STATUS["message"] = "Downloading Upstox Master Data..." print(">>> [BG TASK] STARTING DOWNLOAD (NSE, BSE, MCX)...") # 1. NSE url_nse = "https://assets.upstox.com/market-quote/instruments/exchange/NSE.json.gz" df_nse = pd.read_json(url_nse, compression='gzip') # 2. BSE url_bse = "https://assets.upstox.com/market-quote/instruments/exchange/BSE.json.gz" df_bse = pd.read_json(url_bse, compression='gzip') # 3. MCX url_mcx = "https://assets.upstox.com/market-quote/instruments/exchange/MCX.json.gz" df_mcx = pd.read_json(url_mcx, compression='gzip') SYSTEM_STATUS["message"] = "Processing Data..." # Combine all exchanges df = pd.concat([df_nse, df_bse, df_mcx], ignore_index=True) # Filter relevant segments mask = df['segment'].isin(['NSE_FO', 'BSE_FO', 'MCX_FO', 'NSE_INDEX', 'BSE_INDEX', 'NSE_EQ']) df = df[mask] temp_db = {} search_list = [] # --------------------------------------------------------- # STEP 1: Build the SPOT LOOKUP Dictionary # Map: "RELIANCE" -> "NSE_EQ|INE002A01018" # --------------------------------------------------------- spot_lookup = {} for _, row in df.iterrows(): seg = row['segment'] # For Stocks (Equity): Use 'trading_symbol' (e.g., "RELIANCE") if seg == 'NSE_EQ': sym = str(row['trading_symbol']).strip().upper() key = row['instrument_key'] spot_lookup[sym] = key # For Indices: Use 'trading_symbol' (e.g., "Nifty 50") elif seg == 'NSE_INDEX' or seg == 'BSE_INDEX': sym = str(row['trading_symbol']).strip().upper() key = row['instrument_key'] spot_lookup[sym] = key # Also map 'name' just in case name = str(row['name']).strip().upper() spot_lookup[name] = key # --------------------------------------------------------- # STEP 2: Build the DERIVATIVES Chain # --------------------------------------------------------- for _, row in df.iterrows(): seg = row['segment'] # Skip non-derivatives if seg not in ['NSE_FO', 'BSE_FO', 'MCX_FO']: continue sym = row['trading_symbol'] # e.g. RELIANCE24JANFUT key = row['instrument_key'] # ----------------------------------------------------- # CRITICAL FIX: IDENTIFY THE ROOT SYMBOL # Priority: underlying_symbol > asset_symbol > name # ----------------------------------------------------- root = "" # 1. Try 'underlying_symbol' (This gives "RELIANCE" correctly for stocks) us = row.get('underlying_symbol') if pd.notna(us) and str(us).strip() != "": root = str(us).strip().upper() # 2. Fallback to 'asset_symbol' if underlying is empty if not root: asym = row.get('asset_symbol') if pd.notna(asym) and str(asym).strip() != "": root = str(asym).strip().upper() # 3. Last resort: 'name' (Though usually Long Name for stocks) if not root: nm = row.get('name') if pd.notna(nm): root = str(nm).strip().upper() # Normalize (remove spaces just in case) clean_root = root.replace(' ', '') # Initialize DB Entry if clean_root not in temp_db: # Find Spot Key spot_key = "" # A: Manual Map (Indices) if clean_root in INDEX_MAP: spot_key = INDEX_MAP[clean_root] # B: Lookup using raw ROOT (e.g. "RELIANCE") matches "NSE_EQ" trading_symbol elif root in spot_lookup: spot_key = spot_lookup[root] # C: Lookup using cleaned ROOT elif clean_root in spot_lookup: spot_key = spot_lookup[clean_root] # Determine Exchange Code exch_code = seg.split('_')[0] # Use Short Name for display if possible, else Full Name # If root is "RELIANCE", display is likely "RELIANCE" # If row['name'] is "RELIANCE INDUSTRIES LTD", use that for nice display display_name = row['name'] if pd.notna(row['name']) else root temp_db[clean_root] = { "spot": spot_key, "exch": exch_code, "name": display_name, "items": [] } inst_type = row.get('instrument_type', '') # Parse Expiry exp = 0 raw_exp = row.get('expiry') if pd.notna(raw_exp): try: if isinstance(raw_exp, (int, float)): exp = int(raw_exp / 1000) elif isinstance(raw_exp, str): dt = datetime.strptime(raw_exp, "%Y-%m-%d") exp = int(dt.replace(tzinfo=None).timestamp()) except: pass strike = float(row.get('strike_price', 0.0)) # Map Types t_mapped = "FUT" if inst_type == "CE": t_mapped = "CE" elif inst_type == "PE": t_mapped = "PE" elif inst_type in ["FUTIDX", "FUTSTK", "FUTCOM", "FUTBLN", "FUTENR"]: t_mapped = "FUT" temp_db[clean_root]["items"].append({ "s": key, "dis": sym, "e": exp, "k": strike, "t": t_mapped }) # --------------------------------------------------------- # STEP 3: Finalize # --------------------------------------------------------- final_db = {} for root, data in temp_db.items(): if not data["items"]: continue sorted_items = sorted(data["items"], key=lambda x: (x['e'], x['k'])) final_db[root] = { "spot": data["spot"], "exch": data["exch"], "items": sorted_items } # Add to Search # We use the full name in display so user sees "RELIANCE INDUSTRIES LTD" # But the 'root' key is "RELIANCE" search_list.append({ "root": root, "name": data["name"], "exch": data["exch"], "display": f"{data['name']} ({root})" }) MASTER_DB = final_db SEARCH_INDEX = sorted(search_list, key=lambda x: x['root']) with open(CACHE_FILE, "w") as f: json.dump({"db": final_db, "idx": SEARCH_INDEX}, f) SYSTEM_STATUS["ready"] = True SYSTEM_STATUS["message"] = "Ready" print(f">>> [BG TASK] COMPLETE. Loaded {len(SEARCH_INDEX)} roots. Fixed Underlying Symbol Logic.") except Exception as e: print(f">>> [BG TASK] ERROR: {e}") SYSTEM_STATUS["message"] = f"Error: {str(e)}" finally: UPDATE_LOCK.release() @app.on_event("startup") def startup_event(): global MASTER_DB, SEARCH_INDEX if os.path.exists(CACHE_FILE): try: print(">>> STARTUP: Loading Cache...") with open(CACHE_FILE, "r") as f: data = json.load(f) MASTER_DB = data["db"] SEARCH_INDEX = data["idx"] SYSTEM_STATUS["ready"] = True SYSTEM_STATUS["message"] = "Ready (Cached)" except: pass if not SYSTEM_STATUS["ready"]: thread = threading.Thread(target=update_master_db) thread.start() # ========================================== # 3. ROUTES # ========================================== @app.get("/") def home(access_token: Optional[str] = Cookie(None)): if not access_token: auth_url = ( f"https://api.upstox.com/v2/login/authorization/dialog?" f"response_type=code&client_id={API_KEY}&redirect_uri={REDIRECT_URI}" ) return HTMLResponse(f"""
{json.dumps(json_resp, indent=2)}")
@app.get("/logout")
def logout():
response = RedirectResponse(url="/")
response.delete_cookie("access_token")
return response
@app.get("/api/status")
def get_status():
return JSONResponse(SYSTEM_STATUS)
@app.get("/api/search")
def search_symbol(q: str = Query(..., min_length=1)):
if not SYSTEM_STATUS["ready"]: return JSONResponse([])
query = q.upper()
results = []
# 1. Exact Starts With Root (e.g. "REL")
results += [x for x in SEARCH_INDEX if x['root'].startswith(query)]
# 2. Contains in Name (e.g. "RELIANCE") - limit to 10
if len(results) < 20:
results += [x for x in SEARCH_INDEX if query in x['name'].upper() and x not in results][:10]
return JSONResponse(results[:30])
@app.get("/api/chain")
def get_chain(symbol: str):
root = symbol.upper()
if root in MASTER_DB:
return JSONResponse(MASTER_DB[root])
return JSONResponse({"error": "Symbol not found"}, status_code=404)
@app.get("/api/refresh")
def force_refresh(background_tasks: BackgroundTasks):
SYSTEM_STATUS["ready"] = False
SYSTEM_STATUS["message"] = "Refreshing..."
background_tasks.add_task(update_master_db)
return JSONResponse({"s": "ok"})
# ==========================================
# 4. FRONTEND
# ==========================================
HTML_TEMPLATE = """