#!/usr/bin/env python3 """ Multi-Instrument GEX Dashboard with in-browser token input. Extends gex_app.py to support any Sensibull-available instrument (NIFTY, BANKNIFTY, SENSEX, FINNIFTY, MIDCPNIFTY, stocks like RELIANCE, etc.). Run directly: python GEX/gex_app_multi.py """ import os import json import logging from datetime import datetime, timezone, timedelta import dash from dash import html, dcc, Output, Input, State, ALL import requests import plotly.graph_objects as go import numpy as np import pandas as pd from scipy.stats import norm # Import reusable pieces from gex_dashboard from gex_dashboard import ( SensibullFetcher, _current_checkpoint, _time_to_expiry, implied_vol_vec, _bsm_gamma_vec, build_key_levels_panel, RISK_FREE_RATE, REFRESH_MS, SCALE, IST, BG, GREEN, RED, GOLD, GRAY, PURPLE, DIMWHITE, PCT_RANGE, ) log = logging.getLogger("GEX") # ── Symbol configuration ───────────────────────────────────────────── # label: display name, yahoo: Yahoo Finance ticker for spot price, # lot_size_fallback: used only if API doesn't return lot_size SYMBOLS = { 'NIFTY': {'label': 'NIFTY 50', 'yahoo': '^NSEI', 'lot_size_fallback': 75}, 'BANKNIFTY': {'label': 'BANK NIFTY', 'yahoo': '^NSEBANK', 'lot_size_fallback': 30}, 'SENSEX': {'label': 'SENSEX', 'yahoo': '^BSESN', 'lot_size_fallback': 20}, 'FINNIFTY': {'label': 'FIN NIFTY', 'yahoo': None, 'lot_size_fallback': 65}, 'MIDCPNIFTY': {'label': 'MIDCAP NIFTY', 'yahoo': None, 'lot_size_fallback': 120}, 'CRUDEOIL': {'label': 'CRUDE OIL', 'yahoo': None, 'lot_size_fallback': 100, 'exchange': 'MCX'}, 'GOLD': {'label': 'GOLD', 'yahoo': None, 'lot_size_fallback': 100, 'exchange': 'MCX'}, 'SILVER': {'label': 'SILVER', 'yahoo': None, 'lot_size_fallback': 30, 'exchange': 'MCX'}, 'NATURALGAS': {'label': 'NATURAL GAS', 'yahoo': None, 'lot_size_fallback': 1250, 'exchange': 'MCX'}, } DEFAULT_SYMBOL = 'NIFTY' # Quick-select buttons shown in the header QUICK_SYMBOLS = ['NIFTY', 'BANKNIFTY', 'SENSEX', 'CRUDEOIL', 'GOLD'] def _detect_lot_size(instruments, symbol): """Try to extract lot_size from Sensibull instrument data, fallback to config. For MCX instruments, the effective lot size is lot_size * multiplier (e.g. CRUDEOIL: lot_size=1, multiplier=100 → effective=100 barrels).""" for inst in instruments: if inst.get('instrument_type') in ('CE', 'PE'): ls = inst.get('lot_size') mult = inst.get('multiplier', 1) if ls and ls > 0: effective = int(ls * (mult if mult and mult > 0 else 1)) if effective > 0: return effective return SYMBOLS.get(symbol, {}).get('lot_size_fallback', 75) def _detect_strike_gap(df): """Detect the most common strike gap from a sorted DataFrame of strikes.""" strikes = sorted(df['Strike'].unique()) if len(strikes) < 2: return 50 diffs = [strikes[i+1] - strikes[i] for i in range(min(len(strikes) - 1, 20))] # Most common gap from collections import Counter gap = Counter(diffs).most_common(1)[0][0] return max(gap, 1) def process_instruments_multi(instruments, symbol, expiry=None, spot_override=None): """Like gex_dashboard.process_instruments but with dynamic lot_size and spot. spot_override: if provided, use this as spot price (from Sensibull API).""" if not instruments: return None, 0, None, [], 0 futs = sorted([i for i in instruments if i['instrument_type'] == 'FUT'], key=lambda x: x['expiry']) futures_price = futs[0]['last_price'] if futs else 0 if futures_price <= 0 and not spot_override: return None, 0, None, [], 0 # Priority: Sensibull spot API > futures price (for MCX, futures IS the reference) if spot_override and spot_override > 0: spot = spot_override else: spot = futures_price lot_size = _detect_lot_size(instruments, symbol) expiry_set = sorted(set(i['expiry'] for i in instruments if i['instrument_type'] in ('CE', 'PE'))) if expiry is None and expiry_set: expiry = expiry_set[0] rows = [] for inst in instruments: if inst['instrument_type'] not in ('CE', 'PE'): continue if inst['expiry'] != expiry: continue rows.append({ 'Strike': inst['strike'], 'Type': inst['instrument_type'], 'OI': inst.get('oi', 0) or 0, 'LTP': inst.get('last_price', 0) or 0, }) if not rows: return None, spot, expiry, expiry_set, lot_size df = pd.DataFrame(rows) T = _time_to_expiry(expiry) calls = df[df['Type'] == 'CE'].rename(columns={'OI': 'Call_OI', 'LTP': 'Call_LTP'})[['Strike', 'Call_OI', 'Call_LTP']] puts = df[df['Type'] == 'PE'].rename(columns={'OI': 'Put_OI', 'LTP': 'Put_LTP'})[['Strike', 'Put_OI', 'Put_LTP']] merged = pd.merge(calls, puts, on='Strike', how='outer').fillna(0) merged = merged.sort_values('Strike').reset_index(drop=True) K = merged['Strike'].values.astype(float) call_iv = implied_vol_vec(spot, K, T, RISK_FREE_RATE, merged['Call_LTP'].values.astype(float), is_call=True) put_iv = implied_vol_vec(spot, K, T, RISK_FREE_RATE, merged['Put_LTP'].values.astype(float), is_call=False) avg_iv = np.where((call_iv > 0) & (put_iv > 0), (call_iv + put_iv) / 2, np.where(call_iv > 0, call_iv, put_iv)) gamma = _bsm_gamma_vec(spot, K, T, avg_iv) gamma = np.round(gamma, 4) merged['Call_Gamma'] = gamma merged['Put_Gamma'] = gamma spot_sq = spot ** 2 merged['Call_GEX'] = gamma * merged['Call_OI'] * spot_sq * lot_size merged['Put_GEX'] = -gamma * merged['Put_OI'] * spot_sq * lot_size merged['Net_GEX'] = merged['Call_GEX'] + merged['Put_GEX'] merged['Net_GEX_sc'] = merged['Net_GEX'] / SCALE merged['Call_GEX_sc'] = merged['Call_GEX'] / SCALE merged['Put_GEX_sc'] = merged['Put_GEX'] / SCALE return merged, spot, expiry, expiry_set, lot_size # ── Overridden key levels with dynamic strike gap ──────────────────── def _compute_key_levels_multi(plot_df, spot, strike_gap): """Like gex_dashboard._compute_key_levels but rounds gamma flip to the actual strike gap instead of hardcoded 50.""" pos = plot_df[plot_df['Net_GEX_sc'] > 0].sort_values('Net_GEX_sc', ascending=False) neg = plot_df[plot_df['Net_GEX_sc'] < 0].sort_values('Net_GEX_sc', ascending=True) resistances = [{'strike': int(r['Strike']), 'gex': r['Net_GEX_sc']} for _, r in pos.head(4).iterrows()] supports = [{'strike': int(r['Strike']), 'gex': r['Net_GEX_sc']} for _, r in neg.head(4).iterrows()] sdf = plot_df.sort_values('Strike').reset_index(drop=True) gv, sv = sdf['Net_GEX_sc'].values, sdf['Strike'].values gamma_flip = None for i in range(len(gv) - 1): if gv[i] * gv[i + 1] < 0: s1, s2, g1, g2 = sv[i], sv[i+1], gv[i], gv[i+1] cross = s1 + (s2 - s1) * abs(g1) / (abs(g1) + abs(g2)) if gamma_flip is None or abs(cross - spot) < abs(gamma_flip - spot): gamma_flip = cross if gamma_flip is not None: gamma_flip = round(gamma_flip / strike_gap) * strike_gap return resistances, supports, gamma_flip # ── Overridden build_figure with dynamic dtick ────────────────────── def build_figure_multi(df, spot, strike_gap, snap_now=None, snap_prev=None): """Like gex_dashboard.build_figure but with dynamic y-axis dtick and gamma flip rounding based on actual strike gap.""" lower = spot * (1 - PCT_RANGE) upper = spot * (1 + PCT_RANGE) plot_df = df[(df['Strike'] >= lower) & (df['Strike'] <= upper)].copy() plot_df = plot_df.sort_values('Strike').reset_index(drop=True) if plot_df.empty: fig = go.Figure() fig.update_layout(paper_bgcolor=BG, plot_bgcolor=BG) return fig, [], [], None strikes = plot_df['Strike'].values net_gex = plot_df['Net_GEX_sc'].values max_pos = max(net_gex.max(), 0.01) min_neg = min(net_gex.min(), -0.01) resistances, supports, gamma_flip = _compute_key_levels_multi(plot_df, spot, strike_gap) # Bar colours — gradient colors = [] for v in net_gex: if v >= 0: t = min(v / max_pos, 1.0) r = int((0.06 + 0.04 * (1 - t)) * 255) g = int((0.38 + 0.62 * t) * 255) b = int((0.22 + 0.15 * (1 - t)) * 255) else: t = min(abs(v) / abs(min_neg), 1.0) r = int((0.38 + 0.62 * t) * 255) g = int((0.06 + 0.08 * (1 - t)) * 255) b = int((0.12 + 0.08 * (1 - t)) * 255) colors.append(f'rgba({r},{g},{b},0.92)') bar_w = (strikes[1] - strikes[0]) * 0.76 if len(strikes) > 1 else strike_gap * 0.76 fig = go.Figure() # Zone shading fig.add_vrect(x0=0, x1=max_pos * 1.15, fillcolor='rgba(0,255,136,0.03)', line_width=0, layer='below') fig.add_vrect(x0=-abs(min_neg) * 1.15, x1=0, fillcolor='rgba(255,68,68,0.03)', line_width=0, layer='below') # Main bars fig.add_trace(go.Bar( y=strikes, x=net_gex, orientation='h', marker=dict(color=colors, line_width=0), width=bar_w, customdata=np.column_stack([ plot_df['Call_GEX_sc'].values, plot_df['Put_GEX_sc'].values, plot_df['Call_OI'].values, plot_df['Put_OI'].values, ]), hovertemplate=( 'Strike %{y:,.0f}
' 'Net GEX: %{x:,.0f} Cr
' 'Call GEX: +%{customdata[0]:,.0f} Cr
' 'Put GEX: %{customdata[1]:,.0f} Cr
' 'Call OI: %{customdata[2]:,.0f}
' 'Put OI: %{customdata[3]:,.0f}' ), showlegend=False, )) # Delta overlays (hourly checkpoint comparison) if snap_now and snap_prev: delta_threshold = (max_pos + abs(min_neg)) * 0.01 gain_y, gain_x, gain_base, gain_colors, gain_custom = [], [], [], [], [] for _, row in plot_df.iterrows(): strike = row['Strike'] key = str(int(strike)) curr = snap_now.get(key) prev = snap_prev.get(key) if curr is None or prev is None: continue delta = curr - prev if abs(delta) < delta_threshold: continue if abs(curr) > abs(prev): gain_y.append(strike) gain_x.append(delta) gain_base.append(prev) gain_colors.append('rgba(100,255,200,0.55)' if curr >= 0 else 'rgba(255,130,130,0.55)') gain_custom.append([prev, curr]) else: x0, x1 = min(curr, prev), max(curr, prev) fig.add_shape( type='rect', x0=x0, x1=x1, y0=strike - bar_w / 2, y1=strike + bar_w / 2, line=dict(color='rgba(255,255,255,0.3)', width=1, dash='dot'), fillcolor='rgba(255,255,255,0.03)', ) if gain_y: fig.add_trace(go.Bar( y=gain_y, x=gain_x, base=gain_base, orientation='h', marker=dict(color=gain_colors, line=dict(color='rgba(255,255,255,0.8)', width=1.5)), width=bar_w, showlegend=False, customdata=gain_custom, hovertemplate='Strike %{y:,.0f}
' '\u0394 GEX: %{x:+,.1f} Cr
' '%{customdata[0]:,.1f} \u2192 %{customdata[1]:,.1f} Cr' '', )) # Zero line fig.add_vline(x=0, line_color='rgba(255,255,255,0.2)', line_width=0.8) # Spot line fig.add_hline(y=spot, line_dash='dash', line_color=GOLD, line_width=2.5, opacity=0.85) fig.add_annotation(x=max_pos * 0.85, y=spot, text=f'SPOT {spot:,.0f}', font=dict(size=11, color=GOLD), bgcolor='#14142a', bordercolor=GOLD, borderwidth=2, borderpad=4, showarrow=False, xanchor='right', yanchor='middle') # Gamma flip line if gamma_flip is not None: fig.add_hline(y=gamma_flip, line_dash='dot', line_color=PURPLE, line_width=1.8, opacity=0.7) fig.add_annotation(x=-abs(min_neg) * 0.85, y=gamma_flip, text=f'GAMMA FLIP {gamma_flip:,.0f}', font=dict(size=9, color=PURPLE), bgcolor='#180828', bordercolor=PURPLE, borderwidth=1.5, borderpad=3, showarrow=False, xanchor='left', yanchor='middle') # Call Wall annotation if resistances: cw = resistances[0] fig.add_annotation( x=cw['gex'], y=cw['strike'], text=f'CALL WALL {cw["strike"]}', font=dict(size=11, color=GREEN), bgcolor='#082818', bordercolor=GREEN, borderwidth=2, borderpad=4, showarrow=True, arrowhead=2, arrowsize=1.5, arrowwidth=2, arrowcolor=GREEN, ax=-60, ay=-40) # Put Wall annotation if supports: pw = supports[0] fig.add_annotation( x=pw['gex'], y=pw['strike'], text=f'PUT WALL {pw["strike"]}', font=dict(size=11, color=RED), bgcolor='#280808', bordercolor=RED, borderwidth=2, borderpad=4, showarrow=True, arrowhead=2, arrowsize=1.5, arrowwidth=2, arrowcolor=RED, ax=-60, ay=40) # R2-R4 / S2-S4 diamond markers for i, lv in enumerate(resistances[1:4], start=2): fig.add_trace(go.Scatter( x=[lv['gex'] * 0.35], y=[lv['strike']], mode='markers+text', marker=dict(symbol='diamond', size=6, color=GREEN, line=dict(color='white', width=0.5)), text=[f'R{i}'], textposition='middle left', textfont=dict(size=8, color=GREEN), showlegend=False, hoverinfo='skip')) for i, lv in enumerate(supports[1:4], start=2): fig.add_trace(go.Scatter( x=[lv['gex'] * 0.35], y=[lv['strike']], mode='markers+text', marker=dict(symbol='diamond', size=6, color=RED, line=dict(color='white', width=0.5)), text=[f'S{i}'], textposition='middle left', textfont=dict(size=8, color=RED), showlegend=False, hoverinfo='skip')) # Zone labels fig.add_annotation(x=max_pos * 0.5, y=upper, text='POSITIVE GEX (Mean-Reverting)', font=dict(size=9, color='rgba(0,255,136,0.3)'), showarrow=False, yanchor='top') fig.add_annotation(x=-abs(min_neg) * 0.5, y=upper, text='NEGATIVE GEX (Trending)', font=dict(size=9, color='rgba(255,68,68,0.3)'), showarrow=False, yanchor='top') # Dynamic y-axis dtick: aim for ~8-15 ticks in view, snapped to strike gap visible_range = upper - lower ideal_tick = visible_range / 10 dtick = max(strike_gap, round(ideal_tick / strike_gap) * strike_gap) fig.update_layout( paper_bgcolor=BG, plot_bgcolor=BG, margin=dict(l=80, r=10, t=10, b=55), height=820, xaxis=dict( title='Net Gamma Exposure (GEX) [\u20b9 Cr]', title_font=dict(size=13, color=DIMWHITE), tickfont=dict(size=9.5, color='#777777'), tickformat=',', gridcolor='rgba(255,255,255,0.15)', gridwidth=0.4, zeroline=False, range=[-abs(min_neg) * 1.12, max_pos * 1.12], ), yaxis=dict( title='Strike Price', title_font=dict(size=13, color=DIMWHITE), tickfont=dict(size=9.5, color='#777777'), tickformat=',', gridcolor='rgba(255,255,255,0.10)', gridwidth=0.3, dtick=dtick, range=[lower, upper], ), barmode='overlay', bargap=0, hoverlabel=dict(bgcolor='#14142a', bordercolor='#444466', font_size=12, font_color='#eeeeee'), ) return fig, resistances, supports, gamma_flip # ── PCR chart builder ───────────────────────────────────────────────── def _build_pcr_figure(readings): """Build an intraday PCR line chart from accumulated readings.""" fig = go.Figure() if not readings: fig.update_layout( paper_bgcolor=BG, plot_bgcolor=BG, height=250, margin=dict(l=60, r=20, t=30, b=30), xaxis=dict(visible=False), yaxis=dict(visible=False), annotations=[dict(text='PCR chart — waiting for data...', x=0.5, y=0.5, xref='paper', yref='paper', showarrow=False, font=dict(size=14, color='#555'))]) return fig times = [r['time'] for r in readings] pcrs = [r['pcr'] for r in readings] call_ois = [r['call_oi'] for r in readings] put_ois = [r['put_oi'] for r in readings] y_min = min(pcrs) - 0.05 y_max = max(pcrs) + 0.05 # Ensure the 1.0 line is always visible y_min = min(y_min, 0.92) y_max = max(y_max, 1.08) # Green shading above PCR=1.0 (put-heavy / bullish) fig.add_hrect(y0=1.0, y1=y_max, fillcolor='rgba(0,255,136,0.04)', line_width=0, layer='below') # Red shading below PCR=1.0 (call-heavy / bearish) fig.add_hrect(y0=y_min, y1=1.0, fillcolor='rgba(255,68,68,0.04)', line_width=0, layer='below') # Dashed reference line at PCR = 1.0 fig.add_hline(y=1.0, line_dash='dash', line_color='rgba(255,255,255,0.25)', line_width=1.2) fig.add_annotation(x=0, y=1.0, text='PCR = 1.0', xref='paper', font=dict(size=9, color='rgba(255,255,255,0.4)'), showarrow=False, xanchor='left', yanchor='bottom', yshift=2, xshift=4) # PCR line fig.add_trace(go.Scatter( x=times, y=pcrs, mode='lines+markers', line=dict(color=PURPLE, width=2.5), marker=dict(size=6, color=PURPLE, line=dict(color='white', width=1)), customdata=list(zip(call_ois, put_ois)), hovertemplate=( '%{x}
' 'PCR: %{y:.4f}
' 'Call OI: %{customdata[0]:,.0f}
' 'Put OI: %{customdata[1]:,.0f}' ), showlegend=False, )) # Zone labels fig.add_annotation(x=1, y=y_max, text='Put Heavy (Bullish)', xref='paper', font=dict(size=9, color='rgba(0,255,136,0.35)'), showarrow=False, xanchor='right', yanchor='top', yshift=-2) fig.add_annotation(x=1, y=y_min, text='Call Heavy (Bearish)', xref='paper', font=dict(size=9, color='rgba(255,68,68,0.35)'), showarrow=False, xanchor='right', yanchor='bottom', yshift=2) fig.update_layout( paper_bgcolor=BG, plot_bgcolor=BG, height=250, margin=dict(l=60, r=20, t=30, b=30), xaxis=dict( title='Time (IST)', title_font=dict(size=11, color=DIMWHITE), tickfont=dict(size=9, color='#777777'), gridcolor='rgba(255,255,255,0.08)', gridwidth=0.3, ), yaxis=dict( title='Put-Call Ratio (OI)', title_font=dict(size=11, color=DIMWHITE), tickfont=dict(size=9, color='#777777'), tickformat='.2f', gridcolor='rgba(255,255,255,0.08)', gridwidth=0.3, range=[y_min, y_max], ), hoverlabel=dict(bgcolor='#14142a', bordercolor='#444466', font_size=12, font_color='#eeeeee'), ) return fig # ── App ─────────────────────────────────────────────────────────────── fetcher = SensibullFetcher(None) # Track which symbol the fetcher's cached data belongs to, # to prevent cross-symbol data contamination. _fetcher_last_symbol = {'sym': None} app = dash.Dash(__name__, title='GEX Dashboard', update_title=None, suppress_callback_exceptions=True) app.index_string = ''' {%metas%}{%title%}{%favicon%}{%css%} {%app_entry%}''' def _market_badge(): now = datetime.now(IST) is_open = now.weekday() < 5 and ( now.replace(hour=9, minute=15, second=0, microsecond=0) <= now <= now.replace(hour=15, minute=30, second=0, microsecond=0)) color = GREEN if is_open else RED label = 'Market Open' if is_open else 'Market Closed' return html.Span([ html.Span('\u25cf ', style={'color': color, 'fontSize': '12px'}), html.Span(label, style={'color': color, 'fontSize': '11px', 'fontWeight': 'bold'})]) # ── Token input page ───────────────────────────────────────────────── def _token_page(): return html.Div([ html.Div([ html.H1('GEX Dashboard', style={ 'fontSize': '28px', 'fontWeight': 'bold', 'color': '#fff', 'textAlign': 'center', 'marginBottom': '8px', 'letterSpacing': '2px'}), html.Div('Sensibull Token Required', style={ 'fontSize': '14px', 'color': GRAY, 'textAlign': 'center', 'marginBottom': '32px'}), html.Div([ html.Div('How to get your token:', style={ 'fontSize': '12px', 'color': '#bbb', 'marginBottom': '8px', 'fontWeight': 'bold'}), html.Ol([ html.Li('Log into web.sensibull.com in your browser'), html.Li('Open DevTools (F12) → Application → Cookies'), html.Li('Find api.sensibull.com → copy access_token value'), html.Li('Paste it below'), ], style={'fontSize': '11px', 'color': '#999', 'paddingLeft': '20px', 'lineHeight': '1.8'}), ], style={'backgroundColor': '#0c0c24', 'border': '1px solid #222244', 'borderRadius': '6px', 'padding': '16px', 'marginBottom': '24px'}), dcc.Input(id='token-input', type='text', placeholder='Paste your access_token here...', style={'width': '100%', 'padding': '12px', 'fontSize': '14px', 'backgroundColor': '#0c0c24', 'border': '1px solid #333366', 'borderRadius': '6px', 'color': '#fff', 'outline': 'none', 'boxSizing': 'border-box', 'marginBottom': '16px'}), html.Button('Connect', id='token-submit', n_clicks=0, style={ 'width': '100%', 'padding': '12px', 'fontSize': '14px', 'fontWeight': 'bold', 'backgroundColor': GREEN, 'color': '#000', 'border': 'none', 'borderRadius': '6px', 'cursor': 'pointer', 'letterSpacing': '1px'}), html.Div(id='token-error', style={ 'fontSize': '12px', 'color': RED, 'textAlign': 'center', 'marginTop': '12px', 'minHeight': '18px'}), ], style={'width': '420px', 'padding': '40px', 'backgroundColor': BG, 'border': '1px solid #222244', 'borderRadius': '12px'}), ], style={'display': 'flex', 'justifyContent': 'center', 'alignItems': 'center', 'minHeight': '100vh', 'backgroundColor': BG}) # ── Dashboard page ─────────────────────────────────────────────────── def _quick_btn(sym, is_default=False): """Small quick-select button for a popular symbol.""" return html.Button(sym, id={'type': 'quick-sym', 'sym': sym}, n_clicks=0, style={ 'padding': '2px 8px', 'fontSize': '10px', 'fontWeight': 'bold', 'backgroundColor': '#1a1a3e' if is_default else 'transparent', 'color': GOLD if is_default else '#666', 'border': f'1px solid {GOLD}' if is_default else '1px solid #333', 'borderRadius': '3px', 'cursor': 'pointer', 'marginRight': '3px', }) def _dashboard_page(): return html.Div([ dcc.Store(id='active-symbol', data=DEFAULT_SYMBOL), html.Div([ html.Div([ html.H1(id='dashboard-title', style={ 'fontSize': '22px', 'fontWeight': 'bold', 'color': '#fff', 'margin': '0', 'letterSpacing': '1px'}), html.Div(id='subtitle', style={'fontSize': '11px', 'color': GRAY, 'marginTop': '3px'}), ]), html.Div([ # Symbol text input + Go button html.Div([ dcc.Input(id='symbol-input', type='text', value=DEFAULT_SYMBOL, n_submit=0, placeholder='e.g. RELIANCE', debounce=True, style={'width': '120px', 'padding': '5px 8px', 'fontSize': '12px', 'backgroundColor': '#0c0c24', 'border': '1px solid #333366', 'borderRadius': '4px 0 0 4px', 'color': '#fff', 'outline': 'none', 'textTransform': 'uppercase'}), html.Button('Go', id='symbol-go-btn', n_clicks=0, style={ 'padding': '5px 10px', 'fontSize': '12px', 'fontWeight': 'bold', 'backgroundColor': GOLD, 'color': '#000', 'border': 'none', 'borderRadius': '0 4px 4px 0', 'cursor': 'pointer'}), ], style={'display': 'flex', 'alignItems': 'center'}), # Quick-select buttons html.Div([ _quick_btn(s, s == DEFAULT_SYMBOL) for s in QUICK_SYMBOLS ], style={'display': 'flex', 'alignItems': 'center', 'marginLeft': '8px'}), html.Div([ html.Label('Expiry:', style={'color': '#888', 'fontSize': '11px', 'marginRight': '5px', 'marginLeft': '12px'}), dcc.Dropdown(id='expiry-dd', style={'width': '160px'}, clearable=False), ], style={'display': 'flex', 'alignItems': 'center'}), html.Div(id='market-badge', style={'marginLeft': '12px'}), html.Div(id='source-badge', style={'marginLeft': '8px'}), html.Div(id='symbol-error', style={'marginLeft': '8px', 'fontSize': '10px', 'color': RED, 'minWidth': '0'}), html.Button('\u21bb Token', id='change-token-btn', n_clicks=0, title='Change token', style={ 'marginLeft': '12px', 'padding': '3px 10px', 'fontSize': '11px', 'backgroundColor': 'transparent', 'color': '#555', 'border': '1px solid #333', 'borderRadius': '4px', 'cursor': 'pointer'}), ], style={'display': 'flex', 'alignItems': 'center'}), ], style={'display': 'flex', 'justifyContent': 'space-between', 'alignItems': 'flex-start', 'padding': '12px 20px 4px'}), html.Div([ html.Div([ dcc.Loading(dcc.Graph(id='gex-chart', config={ 'displayModeBar': False, 'displaylogo': False}), type='dot', color=GOLD), ], style={'flex': '2.4', 'minWidth': '0'}), html.Div(id='key-levels-panel', style={'flex': '1', 'minWidth': '280px', 'maxWidth': '340px', 'padding': '0 0 0 6px'}), ], style={'display': 'flex', 'padding': '0 12px', 'gap': '2px'}), # PCR intraday chart — full width, compact html.Div([ dcc.Graph(id='pcr-chart', config={ 'displayModeBar': False, 'displaylogo': False}, style={'height': '250px'}), ], style={'padding': '0 12px'}), html.Div([ html.Span('Auto-refreshes every 60s | Data: Sensibull | Greeks: Black-Scholes', style={'color': '#444'}), ], style={'textAlign': 'center', 'fontSize': '10px', 'padding': '6px', 'borderTop': '1px solid #222244'}), dcc.Interval(id='timer', interval=REFRESH_MS, n_intervals=0), dcc.Store(id='data-store'), dcc.Store(id='hourly-gex-store'), dcc.Store(id='pcr-history'), ]) # ── App layout ─────────────────────────────────────────────────────── app.layout = html.Div([ dcc.Store(id='token-store', storage_type='local'), html.Div(id='token-page-container', children=_token_page()), html.Div(id='dashboard-page-container', children=_dashboard_page(), style={'display': 'none'}), ], style={'backgroundColor': BG, 'minHeight': '100vh', 'fontFamily': "'SF Mono','Fira Code','Consolas','Monaco',monospace", 'color': '#fff'}) # ══════════════════════════════════════════════════════════════════════ # CALLBACKS # ══════════════════════════════════════════════════════════════════════ @app.callback( [Output('token-store', 'data'), Output('token-error', 'children')], Input('token-submit', 'n_clicks'), State('token-input', 'value'), prevent_initial_call=True, ) def submit_token(n_clicks, token_value): if not token_value or not token_value.strip(): return dash.no_update, 'Please paste your token' token = token_value.strip() try: r = requests.get( f"{SensibullFetcher.API_BASE}/instruments/{DEFAULT_SYMBOL}", headers=SensibullFetcher.HEADERS, cookies={'access_token': token}, timeout=10, ) data = r.json() instruments = data.get('data', []) if r.status_code in (401, 403) or not instruments: return dash.no_update, f'Invalid or expired token (HTTP {r.status_code})' today_ist = datetime.now(IST).strftime('%Y-%m-%d') return {'token': token, 'date': today_ist}, '' except Exception as e: return dash.no_update, f'Connection error: {e}' @app.callback( Output('token-store', 'clear_data'), Input('change-token-btn', 'n_clicks'), prevent_initial_call=True, ) def change_token(n_clicks): return True @app.callback( [Output('token-page-container', 'style'), Output('dashboard-page-container', 'style')], Input('token-store', 'data'), ) def route_page(token_data): show = {} hide = {'display': 'none'} if token_data and token_data.get('token'): today_ist = datetime.now(IST).strftime('%Y-%m-%d') if token_data.get('date', '') == today_ist: fetcher.token = token_data['token'] fetcher.token_valid = True return hide, show return show, hide @app.callback( [Output('active-symbol', 'data'), Output('symbol-input', 'value'), Output('symbol-error', 'children')], [Input('symbol-go-btn', 'n_clicks'), Input('symbol-input', 'n_submit'), Input({'type': 'quick-sym', 'sym': ALL}, 'n_clicks')], State('symbol-input', 'value'), prevent_initial_call=True, ) def set_active_symbol(go_clicks, n_submit, quick_clicks, typed_value): ctx = dash.callback_context if not ctx.triggered: raise dash.exceptions.PreventUpdate trigger = ctx.triggered[0]['prop_id'] # Quick-select button clicked — these are known-good symbols, # but still guard against missing token. if 'quick-sym' in trigger: btn_id = json.loads(trigger.split('.')[0]) sym = btn_id['sym'].upper().strip() if not fetcher.token: return dash.no_update, sym, 'No token — connect first' return sym, sym, '' # Go button or Enter in text input if not typed_value or not typed_value.strip(): return dash.no_update, dash.no_update, 'Enter a symbol' sym = typed_value.strip().upper() if not fetcher.token: return dash.no_update, dash.no_update, 'No token — connect first' # Validate via Sensibull (try NSE first, then MCX metacache) try: r = requests.get( f"{SensibullFetcher.API_BASE}/instruments/{sym}", headers=SensibullFetcher.HEADERS, cookies={'access_token': fetcher.token}, timeout=10, ) data = r.json().get('data', []) if r.status_code in (401, 403): return dash.no_update, dash.no_update, 'Token expired' if not data: # Check MCX metacache for the symbol metacache = fetcher._fetch_metacache() if sym not in metacache: return dash.no_update, dash.no_update, f'"{sym}" not found' except Exception: return dash.no_update, dash.no_update, 'Connection error' return sym, sym, '' @app.callback( Output('dashboard-title', 'children'), Input('active-symbol', 'data'), ) def update_title(symbol): label = SYMBOLS.get(symbol, {}).get('label', symbol) return f'{label} Dealer Gamma Exposure (GEX)' @app.callback( [Output('data-store', 'data'), Output('expiry-dd', 'options'), Output('expiry-dd', 'value'), Output('source-badge', 'children')], [Input('timer', 'n_intervals'), Input('active-symbol', 'data')], State('expiry-dd', 'value'), ) def fetch_data(n, symbol, current_expiry): symbol = symbol or DEFAULT_SYMBOL # Clear stale cache if symbol changed — prevents cross-symbol contamination. if symbol != _fetcher_last_symbol['sym']: fetcher.last_data = None _fetcher_last_symbol['sym'] = symbol instruments = fetcher.fetch(symbol) if instruments: # Fetch spot price from Sensibull (works for NSE indices; MCX returns None) spot_price = fetcher.fetch_spot(symbol) expiry_set = sorted(set(i['expiry'] for i in instruments if i['instrument_type'] in ('CE', 'PE'))) options = [{'label': datetime.strptime(e, '%Y-%m-%d').strftime('%d-%b-%Y'), 'value': e} for e in expiry_set[:8]] # Reset expiry when symbol changes ctx = dash.callback_context symbol_changed = any('active-symbol' in t['prop_id'] for t in ctx.triggered) if symbol_changed or current_expiry not in expiry_set: value = expiry_set[0] if expiry_set else None else: value = current_expiry # Compute PCR for the weekly (nearest) expiry weekly_expiry = expiry_set[0] if expiry_set else None call_oi_total = sum(i.get('oi', 0) or 0 for i in instruments if i['instrument_type'] == 'CE' and i['expiry'] == weekly_expiry) put_oi_total = sum(i.get('oi', 0) or 0 for i in instruments if i['instrument_type'] == 'PE' and i['expiry'] == weekly_expiry) pcr = round(put_oi_total / max(call_oi_total, 1), 4) badge = html.Span('API', style={ 'fontSize': '9px', 'color': GREEN, 'border': f'1px solid {GREEN}', 'borderRadius': '3px', 'padding': '1px 5px'}) return { 'source': 'api', 'instruments': instruments, 'symbol': symbol, 'spot_price': spot_price, 'pcr': pcr, 'call_oi_total': call_oi_total, 'put_oi_total': put_oi_total, 'weekly_expiry': weekly_expiry, }, options, value, badge raise dash.exceptions.PreventUpdate @app.callback( [Output('gex-chart', 'figure'), Output('key-levels-panel', 'children'), Output('subtitle', 'children'), Output('market-badge', 'children'), Output('hourly-gex-store', 'data')], [Input('data-store', 'data'), Input('expiry-dd', 'value')], State('hourly-gex-store', 'data'), ) def render_dashboard(stored, selected_expiry, hourly_state): if not stored: empty = go.Figure() empty.update_layout(paper_bgcolor=BG, plot_bgcolor=BG, xaxis=dict(visible=False), yaxis=dict(visible=False), annotations=[dict(text='Waiting for data...', x=0.5, y=0.5, xref='paper', yref='paper', showarrow=False, font=dict(size=20, color='#555'))]) return empty, html.Div('Loading...', style={'color': '#555'}), '', _market_badge(), dash.no_update symbol = stored.get('symbol', DEFAULT_SYMBOL) if stored['source'] == 'api': df, spot, expiry, _, lot_size = process_instruments_multi( stored['instruments'], symbol, selected_expiry, spot_override=stored.get('spot_price')) else: raise dash.exceptions.PreventUpdate if df is None or df.empty: empty = go.Figure() empty.update_layout(paper_bgcolor=BG, plot_bgcolor=BG) return empty, html.Div('No data', style={'color': '#555'}), '', _market_badge(), dash.no_update label = SYMBOLS.get(symbol, {}).get('label', symbol) strike_gap = _detect_strike_gap(df) # ── Hourly checkpoint logic ── hourly = hourly_state or {} checkpoint = _current_checkpoint() store_key = f"{symbol}|{selected_expiry or expiry}" same_key = hourly.get('_key') == store_key new_hourly = dash.no_update snap_now = hourly.get('current') if same_key else None snap_prev = hourly.get('previous') if same_key else None if checkpoint and (checkpoint != hourly.get('_last_checkpoint') or not same_key): today = datetime.now(IST).strftime('%Y-%m-%d') prev_cp = hourly.get('_last_checkpoint', '') carry_prev = hourly.get('current', {}) if (same_key and prev_cp.startswith(today)) else {} current_gex = {str(int(row['Strike'])): row['Net_GEX_sc'] for _, row in df.iterrows()} new_hourly = { '_last_checkpoint': checkpoint, '_key': store_key, 'previous': carry_prev, 'current': current_gex, } snap_now = current_gex snap_prev = carry_prev or None fig, res, sup, gf = build_figure_multi(df, spot, strike_gap, snap_now=snap_now, snap_prev=snap_prev) panel = build_key_levels_panel(res, sup, spot, gf) total_gex = df['Net_GEX_sc'].sum() regime = 'Positive (Mean-Reverting)' if total_gex > 0 else 'Negative (Trending)' regime_color = GREEN if total_gex > 0 else RED now_str = datetime.now(IST).strftime('%H:%M:%S IST') subtitle = html.Span([ html.Span(f'{label} | Expiry: {expiry} | Spot: {spot:,.0f} | Lot: {lot_size} | {now_str} | '), html.Span(f'Net GEX: {total_gex:,.0f} Cr ({regime})', style={'color': regime_color, 'fontWeight': 'bold'}), ]) return fig, panel, subtitle, _market_badge(), new_hourly @app.callback( [Output('pcr-chart', 'figure'), Output('pcr-history', 'data')], [Input('data-store', 'data')], State('pcr-history', 'data'), ) def update_pcr_chart(stored, pcr_state): if not stored or 'pcr' not in stored: return _build_pcr_figure([]), dash.no_update # pcr_state is a dict keyed by symbol, so each instrument's # readings persist when switching between symbols. # Structure: { "_day": "2026-03-06", "NIFTY": [...], "SENSEX": [...], ... } pcr_state = pcr_state or {} symbol = stored.get('symbol', DEFAULT_SYMBOL) today = datetime.now(IST).strftime('%Y-%m-%d') time_label = datetime.now(IST).strftime('%H:%M') # New trading day — clear all history if pcr_state.get('_day') != today: pcr_state = {'_day': today} readings = pcr_state.get(symbol, []) # Append new reading (avoid duplicate timestamps) new_reading = { 'time': time_label, 'pcr': stored['pcr'], 'call_oi': stored['call_oi_total'], 'put_oi': stored['put_oi_total'], } if not readings or readings[-1]['time'] != time_label: readings.append(new_reading) else: readings[-1] = new_reading pcr_state[symbol] = readings fig = _build_pcr_figure(readings) return fig, pcr_state if __name__ == '__main__': import webbrowser, threading, socket def _find_free_port(start=8051, end=8060): for p in range(start, end): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: if s.connect_ex(('127.0.0.1', p)) != 0: return p return start port = int(os.environ.get('PORT', 0)) or _find_free_port() url = f"http://127.0.0.1:{port}" threading.Timer(1.5, webbrowser.open, args=[url]).start() print(f"\n Multi-Instrument GEX Dashboard -> {url}") print(" Press Ctrl+C to quit\n") app.run(debug=False, port=port, host='0.0.0.0')