""" ========================================= CV6 AI Trading OS — NSE Real Data Provider ========================================= Fetches REAL data from NSE India public APIs: • FII / DII net buying/selling (daily) • India VIX (live) • Market Breadth (advance/decline) • Option Chain OI (Nifty / BankNifty) • Corporate announcements All methods fail gracefully — returns None or {} on error. No random data. No fake data. Real or None. ========================================= """ from __future__ import annotations import time import threading from typing import Any, Dict, List, Optional from loguru import logger import requests # ── NSE requires browser-like headers ──────────────────────────────────────── _HEADERS = { "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/124.0.0.0 Safari/537.36" ), "Accept": "application/json, text/plain, */*", "Accept-Language": "en-US,en;q=0.9", "Accept-Encoding": "gzip, deflate, br", "Referer": "https://www.nseindia.com/", "Connection": "keep-alive", } _BASE = "https://www.nseindia.com" _TIMEOUT = 10 # seconds def _session() -> requests.Session: """Create a session with NSE cookies (required to bypass bot protection).""" s = requests.Session() s.headers.update(_HEADERS) try: # Hit homepage first to get cookies s.get(f"{_BASE}/", timeout=_TIMEOUT) s.get(f"{_BASE}/market-data/live-equity-market", timeout=_TIMEOUT) except Exception: pass return s # ── Thread-safe cached session ──────────────────────────────────────────────── _lock = threading.Lock() _sess: Optional[requests.Session] = None _sess_ts: float = 0.0 _SESS_TTL = 300.0 # refresh session every 5 minutes def _get_session() -> requests.Session: global _sess, _sess_ts with _lock: if _sess is None or (time.time() - _sess_ts) > _SESS_TTL: _sess = _session() _sess_ts = time.time() return _sess def _get(url: str, params: dict | None = None) -> Optional[dict | list]: """GET from NSE API with auto-retry on session expiry.""" for attempt in range(2): try: s = _get_session() r = s.get(url, params=params, timeout=_TIMEOUT) r.raise_for_status() return r.json() except Exception as e: logger.warning(f"[NSE] {url} attempt {attempt+1}: {e}") if attempt == 0: # Force session refresh on first failure global _sess with _lock: _sess = None return None # ── In-memory cache to avoid hammering NSE ─────────────────────────────────── _cache: Dict[str, Any] = {} _cache_ts: Dict[str, float] = {} _CACHE_TTL: Dict[str, float] = { "fii_dii": 900, # 15 min "vix": 60, # 1 min "breadth": 300, # 5 min "oi_nifty": 300, # 5 min "oi_bnf": 300, # 5 min "announce": 600, # 10 min } def _cached(key: str, ttl_key: str, fn): now = time.time() if key in _cache and (now - _cache_ts.get(key, 0)) < _CACHE_TTL.get(ttl_key, 300): return _cache[key] result = fn() if result is not None: _cache[key] = result _cache_ts[key] = now return _cache.get(key) # ── 1. FII / DII Net Activity ───────────────────────────────────────────────── def get_fii_dii() -> Optional[Dict[str, Any]]: """ Returns today's FII and DII net buying/selling in ₹ crore. { "date": "02-Jul-2026", "fii_net": -1234.56, # negative = selling "dii_net": +987.65, "fii_activity": "SELL", # BUY | SELL | NEUTRAL "dii_activity": "BUY", "fii_buy": ..., "fii_sell": ..., "dii_buy": ..., "dii_sell": ... } """ def _fetch(): data = _get(f"{_BASE}/api/fiidiiTradeReact") if not data or not isinstance(data, list): return None # First entry is usually today's data row = data[0] try: fii_net = float(str(row.get("netBuy_Sell_FII", "0")).replace(",", "")) dii_net = float(str(row.get("netBuy_Sell_DII", "0")).replace(",", "")) fii_buy = float(str(row.get("buyValue_FII", "0")).replace(",", "")) fii_sell= float(str(row.get("sellValue_FII","0")).replace(",", "")) dii_buy = float(str(row.get("buyValue_DII", "0")).replace(",", "")) dii_sell= float(str(row.get("sellValue_DII","0")).replace(",", "")) return { "date": row.get("date", ""), "fii_net": round(fii_net, 2), "dii_net": round(dii_net, 2), "fii_buy": round(fii_buy, 2), "fii_sell": round(fii_sell, 2), "dii_buy": round(dii_buy, 2), "dii_sell": round(dii_sell, 2), "fii_activity": "BUY" if fii_net > 100 else ("SELL" if fii_net < -100 else "NEUTRAL"), "dii_activity": "BUY" if dii_net > 100 else ("SELL" if dii_net < -100 else "NEUTRAL"), } except Exception as e: logger.warning(f"[NSE FII/DII] parse error: {e}") return None return _cached("fii_dii", "fii_dii", _fetch) # ── 2. India VIX ───────────────────────────────────────────────────────────── def get_india_vix() -> Optional[float]: """Returns India VIX current value, or None if unavailable.""" def _fetch(): data = _get(f"{_BASE}/api/allIndices") if not data: return None indices = data.get("data", []) for idx in indices: if "INDIA VIX" in str(idx.get("index", "")).upper(): try: return round(float(idx.get("last", 0)), 2) except Exception: pass return None return _cached("vix", "vix", _fetch) # ── 3. Market Breadth (Advance / Decline) ───────────────────────────────────── def get_market_breadth() -> Optional[Dict[str, int]]: """ Returns NSE market breadth. { "advance": 1200, "decline": 650, "unchanged": 150, "new_high": 45, "new_low": 12 } """ def _fetch(): # Try live equity market summary data = _get(f"{_BASE}/api/live-analysis-advance-decline-for-group", params={"type": "sec", "index": "SECAdv"}) if not data: # Fallback: allIndices for NIFTY 500 breadth estimate data2 = _get(f"{_BASE}/api/allIndices") if data2: indices = data2.get("data", []) adv = sum(1 for i in indices if float(i.get("change", 0)) > 0) dec = sum(1 for i in indices if float(i.get("change", 0)) < 0) return {"advance": adv * 8, "decline": dec * 8, "unchanged": 0, "new_high": 0, "new_low": 0} return None try: adv = int(data.get("advances", data.get("advance", 0))) dec = int(data.get("declines", data.get("decline", 0))) unc = int(data.get("unchanged", 0)) return { "advance": adv, "decline": dec, "unchanged": unc, "new_high": int(data.get("new52WHigh", 0)), "new_low": int(data.get("new52WLow", 0)), } except Exception as e: logger.warning(f"[NSE Breadth] parse error: {e}") return None return _cached("breadth", "breadth", _fetch) # ── 4. Option Chain OI ───────────────────────────────────────────────────────── def get_option_chain_oi(symbol: str = "NIFTY") -> Optional[Dict[str, Any]]: """ Returns aggregated OI data from NSE option chain. { "symbol": "NIFTY", "pcr": 0.87, # Put-Call Ratio "total_call_oi": 123456, "total_put_oi": 107654, "max_call_oi_strike": 22000, # resistance "max_put_oi_strike": 21500, # support "oi_signal": "BEARISH", # BULLISH | NEUTRAL | BEARISH "sentiment": "Put writers active — market expects support at 21500" } """ cache_key = f"oi_{symbol.lower()}" ttl_key = "oi_nifty" if "NIFTY" in symbol.upper() else "oi_bnf" def _fetch(): data = _get( f"{_BASE}/api/option-chain-indices", params={"symbol": symbol.upper()} ) if not data: return None records = data.get("records", {}).get("data", []) if not records: return None try: call_oi_by_strike: Dict[int, int] = {} put_oi_by_strike: Dict[int, int] = {} total_call_oi = 0 total_put_oi = 0 for rec in records: strike = int(rec.get("strikePrice", 0)) ce = rec.get("CE", {}) pe = rec.get("PE", {}) c_oi = int(ce.get("openInterest", 0)) if ce else 0 p_oi = int(pe.get("openInterest", 0)) if pe else 0 call_oi_by_strike[strike] = c_oi put_oi_by_strike[strike] = p_oi total_call_oi += c_oi total_put_oi += p_oi pcr = round(total_put_oi / total_call_oi, 3) if total_call_oi else 0.0 max_call_strike = max(call_oi_by_strike, key=call_oi_by_strike.get) if call_oi_by_strike else 0 max_put_strike = max(put_oi_by_strike, key=put_oi_by_strike.get) if put_oi_by_strike else 0 # OI Signal interpretation if pcr > 1.3: oi_signal = "BULLISH" sentiment = f"Put heavy — market bullish, support at {max_put_strike}" elif pcr < 0.7: oi_signal = "BEARISH" sentiment = f"Call heavy — market bearish, resistance at {max_call_strike}" else: oi_signal = "NEUTRAL" sentiment = f"Balanced OI — range {max_put_strike}–{max_call_strike}" return { "symbol": symbol.upper(), "pcr": pcr, "total_call_oi": total_call_oi, "total_put_oi": total_put_oi, "max_call_oi_strike": max_call_strike, "max_put_oi_strike": max_put_strike, "oi_signal": oi_signal, "sentiment": sentiment, } except Exception as e: logger.warning(f"[NSE OI {symbol}] parse error: {e}") return None return _cached(cache_key, ttl_key, _fetch) # ── 5. Corporate Announcements ──────────────────────────────────────────────── def get_announcements(symbol: str, limit: int = 5) -> List[Dict[str, str]]: """ Returns recent NSE corporate announcements for a symbol. [{ "date": "02-Jul-2026", "subject": "Board Meeting...", "desc": "..." }] """ cache_key = f"announce_{symbol.upper()}" def _fetch(): data = _get( f"{_BASE}/api/announcement", params={"index": "equities", "symbol": symbol.upper()} ) if not data: return [] rows = data.get("data", data if isinstance(data, list) else []) results = [] for row in rows[:limit]: results.append({ "date": row.get("an_dt", row.get("date", "")), "subject": row.get("subject", row.get("desc", "")), "desc": row.get("attchmntText", ""), }) return results result = _cached(cache_key, "announce", _fetch) return result or [] # ── 6. Nifty 50 Live Change ─────────────────────────────────────────────────── def get_nifty_change() -> Optional[float]: """Returns Nifty 50 current % change, or None.""" def _fetch(): data = _get(f"{_BASE}/api/allIndices") if not data: return None for idx in data.get("data", []): if str(idx.get("index", "")).strip().upper() in ("NIFTY 50", "NIFTY50"): try: return round(float(idx.get("percentChange", 0)), 2) except Exception: pass return None return _cached("nifty_chg", "vix", _fetch) # ── Singleton ───────────────────────────────────────────────────────────────── nse_data = type("NSEData", (), { "fii_dii": staticmethod(get_fii_dii), "vix": staticmethod(get_india_vix), "breadth": staticmethod(get_market_breadth), "oi": staticmethod(get_option_chain_oi), "announcements": staticmethod(get_announcements), "nifty_change": staticmethod(get_nifty_change), })()