""" sector_pulse.py — NSE sector heatmap and rotation detector. Tracks 10 NSE sector indices via yfinance and detects leading/lagging sectors. Cache: in-memory dict, 5-min TTL (same pattern as news_sentiment.py). Usage: from sector_pulse import get_sector_pulse pulse = get_sector_pulse() # pulse["rotation_signal"] → "DEFENSIVE" | "CYCLICAL" | "GROWTH" | "MIXED" # pulse["leading_sectors"] → ["BANK", "IT"] # pulse["lagging_sectors"] → ["METAL", "REALTY"] Run standalone to test: python sector_pulse.py """ from __future__ import annotations import logging import time from datetime import datetime from typing import Optional import requests as _requests # ── NSE SECTOR INDICES (Yahoo Finance tickers) ──────────────────────────────── _SECTORS = [ {"name": "BANK", "ticker": "^NSEBANK", "label": "Nifty Bank"}, {"name": "IT", "ticker": "^CNXIT", "label": "Nifty IT"}, {"name": "PHARMA", "ticker": "^CNXPHARMA", "label": "Nifty Pharma"}, {"name": "FMCG", "ticker": "^CNXFMCG", "label": "Nifty FMCG"}, {"name": "AUTO", "ticker": "^CNXAUTO", "label": "Nifty Auto"}, {"name": "METAL", "ticker": "^CNXMETAL", "label": "Nifty Metal"}, {"name": "REALTY", "ticker": "^CNXREALTY", "label": "Nifty Realty"}, {"name": "ENERGY", "ticker": "^CNXENERGY", "label": "Nifty Energy"}, {"name": "FINANCE", "ticker": "^CNXFINANCE", "label": "Nifty Financial Services"}, {"name": "INFRA", "ticker": "^CNXINFRA", "label": "Nifty Infra"}, ] # ── SECTOR ROTATION CLASSIFICATION ─────────────────────────────────────────── # Defensive: FMCG, PHARMA (outperform in risk-off environments) # Cyclical: METAL, ENERGY, AUTO (outperform in economic expansion) # Growth: IT, BANK, FINANCE (outperform in low-rate / high-growth) _DEFENSIVE = {"FMCG", "PHARMA"} _CYCLICAL = {"METAL", "ENERGY", "AUTO"} _GROWTH = {"IT", "BANK", "FINANCE"} # ── STOCK → SECTOR MAP (module-level so it can be inverted for sector→stocks) ────────────────── # Large-cap constituents of the 10 tracked NSE sector indices. This is deliberately a curated # large-cap list (the NSE constituent API is bot-blocked / unreliable from datacenter IPs), used # both for the per-stock sector tag AND, inverted, as the candidate pool for the sector-driven # Top Picks scan. TICKER_SECTOR_MAP: dict[str, str] = { # Banking "HDFCBANK": "BANK", "ICICIBANK": "BANK", "KOTAKBANK": "BANK", "AXISBANK": "BANK", "SBIN": "BANK", "INDUSINDBK": "BANK", "BANKBARODA": "BANK", "IDFCFIRSTB": "BANK", "AUBANK": "BANK", "PNB": "BANK", "CANBK": "BANK", "FEDERALBNK": "BANK", # IT/Technology "TCS": "IT", "INFY": "IT", "WIPRO": "IT", "HCLTECH": "IT", "TECHM": "IT", "LTIM": "IT", "PERSISTENT": "IT", "COFORGE": "IT", "MPHASIS": "IT", "OFSS": "IT", # Pharma "SUNPHARMA": "PHARMA", "DRREDDY": "PHARMA", "CIPLA": "PHARMA", "DIVISLAB": "PHARMA", "LUPIN": "PHARMA", "AUROPHARMA": "PHARMA", "BIOCON": "PHARMA", "TORNTPHARM": "PHARMA", "ZYDUSLIFE": "PHARMA", # FMCG "HINDUNILVR": "FMCG", "NESTLEIND": "FMCG", "BRITANNIA": "FMCG", "DABUR": "FMCG", "MARICO": "FMCG", "GODREJCP": "FMCG", "ITC": "FMCG", "TATACONSUM": "FMCG", "COLPAL": "FMCG", # Auto "MARUTI": "AUTO", "TATAMOTORS": "AUTO", "M&M": "AUTO", "BAJAJ-AUTO": "AUTO", "HEROMOTOCO": "AUTO", "EICHERMOT": "AUTO", "TVSMOTOR": "AUTO", "ASHOKLEY": "AUTO", "BOSCHLTD": "AUTO", # Metal "TATASTEEL": "METAL", "JSWSTEEL": "METAL", "HINDALCO": "METAL", "VEDL": "METAL", "COALINDIA": "METAL", "NMDC": "METAL", "JINDALSTEL": "METAL", "SAIL": "METAL", "HINDZINC": "METAL", # Energy "RELIANCE": "ENERGY", "ONGC": "ENERGY", "BPCL": "ENERGY", "IOC": "ENERGY", "NTPC": "ENERGY", "POWERGRID": "ENERGY", "GAIL": "ENERGY", "TATAPOWER": "ENERGY", "ADANIGREEN": "ENERGY", # Realty "DLF": "REALTY", "GODREJPROP": "REALTY", "LODHA": "REALTY", "OBEROIRLTY": "REALTY", "PHOENIXLTD": "REALTY", "PRESTIGE": "REALTY", # Finance (NBFCs) "BAJFINANCE": "FINANCE", "BAJAJFINSV": "FINANCE", "CHOLAFIN": "FINANCE", "MUTHOOTFIN": "FINANCE", "SHRIRAMFIN": "FINANCE", "SBICARD": "FINANCE", "HDFCLIFE": "FINANCE", "SBILIFE": "FINANCE", "ICICIPRULI": "FINANCE", # Infra "LT": "INFRA", "ADANIPORTS": "INFRA", "APOLLOHOSP": "INFRA", "SIEMENS": "INFRA", "ABB": "INFRA", "GMRINFRA": "INFRA", } # ── NSE OFFICIAL SECTOR SOURCE ─────────────────────────────────────────────── # Maps NSE index names (from /api/allIndices) to our internal sector keys. _NSE_INDEX_TO_SECTOR = { "NIFTY BANK": "BANK", "NIFTY IT": "IT", "NIFTY PHARMA": "PHARMA", "NIFTY FMCG": "FMCG", "NIFTY AUTO": "AUTO", "NIFTY METAL": "METAL", "NIFTY REALTY": "REALTY", "NIFTY OIL & GAS": "ENERGY", "NIFTY ENERGY": "ENERGY", "NIFTY FINANCIAL SERVICES": "FINANCE", "NIFTY INFRASTRUCTURE": "INFRA", "NIFTY INFRA": "INFRA", } _NSE_SESSION = _requests.Session() _NSE_SESSION.headers.update({ "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124 Safari/537.36", "Referer": "https://www.nseindia.com/", "Accept": "application/json, text/plain, */*", "Accept-Language": "en-IN,en;q=0.9", }) def _nse_warmup_sector() -> None: try: _NSE_SESSION.get("https://www.nseindia.com", timeout=8) except Exception: pass def _fetch_nse_live_1d() -> dict[str, dict]: """Fetch live 1D % change + current price for each sector from NSE allIndices. Returns a dict keyed by sector name (e.g. "BANK") with keys: change_1d_pct, current Returns {} on any failure — caller falls back to yfinance. """ try: _nse_warmup_sector() r = _NSE_SESSION.get( "https://www.nseindia.com/api/allIndices", timeout=10, ) if r.status_code != 200: return {} rows = r.json().get("data", []) result: dict[str, dict] = {} for row in rows: index_name = row.get("indexSymbol", "").strip().upper() sector = _NSE_INDEX_TO_SECTOR.get(index_name) if not sector: continue pct = row.get("percentChange") current = row.get("last") or row.get("current") if pct is not None and current is not None: result[sector] = { "change_1d_pct": round(float(pct), 2), "current": round(float(current), 2), } return result except Exception as exc: logging.debug("sector_pulse: NSE live fetch failed: %s", exc) return {} # ── CACHE ───────────────────────────────────────────────────────────────────── _CACHE: dict = {} _CACHE_TTL = 300 # 5 minutes def _is_fresh(entry: dict) -> bool: return time.time() - entry.get("_ts", 0) < _CACHE_TTL # ── ROTATION LOGIC ──────────────────────────────────────────────────────────── def _classify_rotation(leading: list[str]) -> str: """ Determine rotation signal from which sectors are leading. We count how many of the top sectors fall in each category bucket. """ if not leading: return "MIXED" n_def = sum(1 for s in leading if s in _DEFENSIVE) n_cyc = sum(1 for s in leading if s in _CYCLICAL) n_grw = sum(1 for s in leading if s in _GROWTH) dominant = max(n_def, n_cyc, n_grw) if dominant == 0: return "MIXED" if n_def == dominant and n_def >= 2: return "DEFENSIVE" if n_cyc == dominant and n_cyc >= 2: return "CYCLICAL" if n_grw == dominant and n_grw >= 2: return "GROWTH" return "MIXED" def _momentum_label(chg_5d: float) -> str: if chg_5d > 2.0: return "LEADING" if chg_5d > 0.5: return "RISING" if chg_5d > -0.5: return "FLAT" if chg_5d > -2.0: return "FALLING" return "LAGGING" # ── DATA FETCH ──────────────────────────────────────────────────────────────── def _fetch_sector_pulse() -> dict: try: import yfinance as yf except ImportError: return _empty_result("yfinance not installed") # Try NSE official API for live 1D data first; blend into yfinance historical. nse_live = _fetch_nse_live_1d() tickers = [s["ticker"] for s in _SECTORS] sector_data = [] failed = [] try: import pandas as pd raw = yf.download(tickers, period="35d", progress=False, auto_adjust=True) close = raw["Close"] if isinstance(close, pd.Series): close = close.to_frame() for s in _SECTORS: ytk = s["ticker"] if ytk not in close.columns: failed.append(s["name"]) continue col = close[ytk].dropna() if len(col) < 2: failed.append(s["name"]) continue yf_latest = float(col.iloc[-1]) def _pct_ago(n: int) -> Optional[float]: if len(col) > n: past = float(col.iloc[-n - 1]) if past > 0: return round((yf_latest / past - 1) * 100, 2) return None chg_5d = _pct_ago(5) or 0.0 chg_1m = _pct_ago(21) or 0.0 # Prefer NSE live data for 1D and current price when available. nse = nse_live.get(s["name"], {}) chg_1d = nse.get("change_1d_pct", _pct_ago(1) or 0.0) current = nse.get("current", yf_latest) sector_data.append({ "name": s["name"], "label": s["label"], "current": round(current, 2), "change_1d_pct": chg_1d, "change_5d_pct": chg_5d, "change_1m_pct": chg_1m, "momentum": _momentum_label(chg_5d), "source": "nse+yf" if nse else "yf", }) except Exception as e: logging.warning("sector_pulse: batch download failed: %s", e) # Try one-by-one fallback for s in _SECTORS: try: import pandas as pd df = yf.download(s["ticker"], period="35d", progress=False, auto_adjust=True) col = df["Close"].dropna() if not df.empty else pd.Series(dtype=float) if len(col) < 2: failed.append(s["name"]) continue yf_latest = float(col.iloc[-1]) def _pct(n: int) -> float: if len(col) > n: past = float(col.iloc[-n - 1]) return round((yf_latest / past - 1) * 100, 2) if past > 0 else 0.0 return 0.0 nse = nse_live.get(s["name"], {}) chg_1d = nse.get("change_1d_pct", _pct(1)) current = nse.get("current", yf_latest) sector_data.append({ "name": s["name"], "label": s["label"], "current": round(current, 2), "change_1d_pct": chg_1d, "change_5d_pct": _pct(5), "change_1m_pct": _pct(21), "momentum": _momentum_label(_pct(5)), "source": "nse+yf" if nse else "yf", }) except Exception: failed.append(s["name"]) if not sector_data: return _empty_result(f"all sector downloads failed ({', '.join(failed)})") # Sort by 5D return (best first) sector_data.sort(key=lambda x: x["change_5d_pct"], reverse=True) leading = [s["name"] for s in sector_data[:3]] lagging = [s["name"] for s in sector_data[-3:]] breadth = sum(1 for s in sector_data if s["change_5d_pct"] > 0) rotation = _classify_rotation(leading) return { "sectors": sector_data, "rotation_signal": rotation, "leading_sectors": leading, "lagging_sectors": lagging, "breadth_score": breadth, # 0–10 sectors with positive 5D return "fetched_at": datetime.now().isoformat(), "_failed": failed, "_ts": time.time(), } def _empty_result(reason: str) -> dict: logging.warning("sector_pulse: returning empty result (%s)", reason) return { "sectors": [], "rotation_signal": "MIXED", "leading_sectors": [], "lagging_sectors": [], "breadth_score": 0, "fetched_at": datetime.now().isoformat(), "_error": reason, "_ts": time.time(), } # ── PUBLIC API ──────────────────────────────────────────────────────────────── def get_sector_pulse(force_refresh: bool = False) -> dict: """ Return NSE sector heatmap dict. Keys: sectors — list of dicts: {name, label, current, change_1d/5d/1m_pct, momentum} rotation_signal — "DEFENSIVE" | "CYCLICAL" | "GROWTH" | "MIXED" leading_sectors — top-3 sector names by 5D return lagging_sectors — bottom-3 sector names by 5D return breadth_score — 0–10: number of sectors with positive 5D return fetched_at — ISO timestamp """ global _CACHE if not force_refresh and _is_fresh(_CACHE): return _CACHE result = _fetch_sector_pulse() _CACHE = result return result def get_sector_for_ticker(ticker: str, pulse: dict | None = None) -> str | None: """ Map an NSE ticker to its sector name (BANK, IT, etc.) Returns None if the ticker's sector is not in the NSE sector index map. Used by predictor_core.py for the sector-relative-strength bonus. """ base = ticker.replace(".NS", "").replace(".BO", "").upper() return TICKER_SECTOR_MAP.get(base) def get_sector_constituents() -> dict[str, list[str]]: """Invert TICKER_SECTOR_MAP → {sector: [TICKER.NS, ...]}. The candidate pool for the sector-driven Top Picks scan (large-cap constituents of each tracked NSE sector index).""" out: dict[str, list[str]] = {} for base, sector in TICKER_SECTOR_MAP.items(): out.setdefault(sector, []).append(f"{base}.NS") return out # ── SECTOR VOLATILITY (avg daily range of the sector index) ─────────────────────────────────── _VOL_CACHE: dict = {} _VOL_TTL = 1800 # 30-min cache def get_sector_volatility(force_refresh: bool = False, window: int = 14) -> list[dict]: """Rank all tracked NSE sectors by realized VOLATILITY (not direction). Volatility = mean intraday range as a % of close, (High-Low)/Close×100, over the last `window` daily bars of each sector index. This is the "most violent sectors" signal used to drive the Top Picks candidate pool. Returns a list of {name, label, volatility_pct, atr_pct, change_5d_pct} sorted by volatility_pct DESC. Cached for 30 min; degrades to an empty list if all downloads fail. """ global _VOL_CACHE now = time.time() if (not force_refresh and _VOL_CACHE.get("data") is not None and now - _VOL_CACHE.get("_ts", 0) < _VOL_TTL): return _VOL_CACHE["data"] rows: list[dict] = [] try: import yfinance as yf import pandas as pd tickers = [s["ticker"] for s in _SECTORS] df = yf.download(tickers, period="35d", progress=False, auto_adjust=True, group_by="ticker") for s in _SECTORS: try: sub = df[s["ticker"]] if s["ticker"] in df.columns.get_level_values(0) else None if sub is None or sub.empty: continue hi = sub["High"].dropna().tail(window) lo = sub["Low"].dropna().tail(window) cl = sub["Close"].dropna().tail(window) n = min(len(hi), len(lo), len(cl)) if n < 3: continue rng_pct = ((hi.iloc[-n:].values - lo.iloc[-n:].values) / cl.iloc[-n:].values) * 100.0 vol_pct = float(pd.Series(rng_pct).mean()) chg_5d = float((cl.iloc[-1] / cl.iloc[-6] - 1) * 100) if len(cl) > 6 else 0.0 rows.append({ "name": s["name"], "label": s["label"], "volatility_pct": round(vol_pct, 2), "atr_pct": round(vol_pct, 2), # alias — index range ≈ ATR% for an index "change_5d_pct": round(chg_5d, 2), }) except Exception: continue except Exception as e: logging.warning("sector_pulse: volatility fetch failed: %s", e) rows.sort(key=lambda r: r["volatility_pct"], reverse=True) _VOL_CACHE = {"data": rows, "_ts": now} return rows def format_pulse_summary(pulse: dict) -> str: """One-line summary of sector pulse for LLM prompts.""" leading = ", ".join(pulse.get("leading_sectors", [])) lagging = ", ".join(pulse.get("lagging_sectors", [])) rotation = pulse.get("rotation_signal", "MIXED") breadth = pulse.get("breadth_score", 0) return ( f"NSE Sector Rotation: {rotation} | Breadth: {breadth}/10 sectors advancing " f"| Leading: {leading or 'N/A'} | Lagging: {lagging or 'N/A'}" ) if __name__ == "__main__": import pprint print("Fetching NSE sector pulse...") pulse = get_sector_pulse(force_refresh=True) print(f"\n{format_pulse_summary(pulse)}\n") for s in pulse["sectors"]: bar = "█" * max(0, int((s["change_5d_pct"] + 5) / 0.5)) print(f" {s['name']:<8} {s['change_5d_pct']:>+6.2f}% (5D) {s['momentum']:<8} {bar}") if pulse.get("_failed"): print(f"\n [skipped: {', '.join(pulse['_failed'])}]")