""" ========================================= CV6 AI Trading OS — Angel One Live Feed ========================================= Real-time WebSocket tick data via Angel One SmartWebSocketV2. Replaces yfinance + REST polling for live LTP during market hours. Architecture: 1. SmartWebSocketV2 WebSocket connection (push, real-time) 2. Angel One REST ltpData() fallback (pull, per-request) 3. yfinance fallback (delayed, last resort) Usage: from app.market.live_feed import live_feed # Connect (call after Angel One broker.connect() succeeds): live_feed.connect(broker_instance) # Subscribe to symbols: live_feed.subscribe(["RELIANCE", "INFY", "NIFTY"]) # Get LTP: price = live_feed.get_ltp("RELIANCE") # returns float or None ========================================= """ from __future__ import annotations import threading import time from dataclasses import dataclass, field from typing import Any, Dict, List, Optional import requests from loguru import logger from app.config.settings import settings # ── Constants ────────────────────────────────────────────────────────────────── _SCRIP_MASTER_URL = ( "https://margincalculator.angelbroking.com/OpenAPI_File/files/OpenAPIScripMaster.json" ) _SCRIP_CACHE_TTL = 86_400 # reload scrip master once per day # WebSocket subscription modes _MODE_LTP = 1 # only LTP _MODE_QUOTE = 2 # LTP + OHLCV (what we use) # Exchange type codes for SmartWebSocketV2 _EXCH_NSE = 1 _EXCH_NFO = 2 _EXCH_BSE = 3 # NSE index tokens (hardcoded — don't change) _INDEX_TOKENS: Dict[str, str] = { "NIFTY": "26000", "NIFTY50": "26000", "NIFTY 50": "26000", "BANKNIFTY": "26009", "FINNIFTY": "26037", "MIDCPNIFTY": "26074", "SENSEX": "1", } # Tick cache max age — beyond this, fall back to REST/yfinance _TICK_MAX_AGE_SECS = 60 # ── Tick dataclass ───────────────────────────────────────────────────────────── @dataclass class Tick: symbol: str token: str ltp: float = 0.0 open: float = 0.0 high: float = 0.0 low: float = 0.0 close: float = 0.0 # previous close volume: int = 0 change_pct: float = 0.0 timestamp: float = field(default_factory=time.time) @property def age_secs(self) -> float: return time.time() - self.timestamp @property def is_fresh(self) -> bool: return self.age_secs < _TICK_MAX_AGE_SECS # ── Main manager ─────────────────────────────────────────────────────────────── class LiveFeedManager: """ Singleton WebSocket manager for real-time NSE market data. Thread-safe. The WebSocket runs in a background daemon thread. All public methods (get_ltp, subscribe, status) are safe to call from the FastAPI async event loop or any thread. """ def __init__(self): self._lock = threading.Lock() self._tick_cache: Dict[str, Tick] = {} # symbol → latest Tick self._token_map: Dict[str, str] = {} # symbol → scrip token self._rev_map: Dict[str, str] = {} # token → symbol self._scrip_data: List[dict] = [] self._scrip_ts: float = 0.0 self._sws = None self._connected = False self._ws_thread: Optional[threading.Thread] = None self._subscribed: set = set() # tokens currently subscribed # Angel One auth — populated by connect() self._auth_token = "" self._feed_token = "" self._client_code = "" self._api_key = settings.ANGEL_API_KEY # TIER-2 (2B): application-level reconnect supervisor state self._broker = None # source broker for token refresh on reconnect self._should_run = False # feed is supposed to be up self._reconnect_lock = threading.Lock() self._supervisor_thread: Optional[threading.Thread] = None self._stop_supervisor = False self._last_tick_ts = 0.0 # ── Scrip master ────────────────────────────────────────────────────────── def _load_scrip_master(self) -> List[dict]: """Download and cache the Angel One scrip master JSON.""" now = time.time() if self._scrip_data and (now - self._scrip_ts) < _SCRIP_CACHE_TTL: return self._scrip_data try: r = requests.get(_SCRIP_MASTER_URL, timeout=20) r.raise_for_status() data = r.json() with self._lock: self._scrip_data = data self._scrip_ts = now logger.info(f"[LiveFeed] Scrip master loaded: {len(data):,} entries") return data except Exception as e: logger.error(f"[LiveFeed] Scrip master load failed: {e}") return self._scrip_data def get_token(self, symbol: str) -> Optional[str]: """ Return NSE scrip token for a symbol. Checks index map → token cache → scrip master search. """ sym = symbol.upper().strip() # Index tokens (instant) if sym in _INDEX_TOKENS: tok = _INDEX_TOKENS[sym] with self._lock: self._token_map[sym] = tok self._rev_map[tok] = sym return tok # Cache hit with self._lock: if sym in self._token_map: return self._token_map[sym] # Scrip master search data = self._load_scrip_master() nse_sym = f"{sym}-EQ" for entry in data: if ( entry.get("exch_seg") == "NSE" and entry.get("symbol") == nse_sym ): tok = entry["token"] with self._lock: self._token_map[sym] = tok self._rev_map[tok] = sym return tok logger.debug(f"[LiveFeed] Token not found for: {sym}") return None # ── WebSocket setup ─────────────────────────────────────────────────────── def connect(self, broker_instance: Any) -> bool: """ Start the WebSocket using credentials from an authenticated AngelOneBroker instance. Safe to call multiple times. """ if self._connected: logger.debug("[LiveFeed] Already connected") return True try: from SmartApi.smartWebSocketV2 import SmartWebSocketV2 # noqa except ImportError: logger.warning( "[LiveFeed] SmartApi package missing — " "run: pip install smartapi-python" ) return False if not broker_instance or not broker_instance.is_connected(): logger.warning("[LiveFeed] Broker not connected — cannot start WebSocket") return False jwt_token = getattr(broker_instance, "jwt_token", "") feed_token = getattr(broker_instance, "feed_token", "") client_id = getattr(broker_instance, "client_id", "") if not feed_token: logger.warning("[LiveFeed] No feed_token from broker — WebSocket unavailable") return False self._auth_token = jwt_token self._feed_token = feed_token self._client_code = client_id self._broker = broker_instance # kept for token refresh on reconnect self._should_run = True self._spawn_ws() self._start_supervisor() return True def _spawn_ws(self) -> None: """Start (or restart) the SmartWebSocketV2 client thread with the current tokens. Extracted so both connect() and the reconnect supervisor share one implementation.""" def _run(): try: from SmartApi.smartWebSocketV2 import SmartWebSocketV2 sws = SmartWebSocketV2( self._auth_token, self._api_key, self._client_code, self._feed_token, max_retry_attempt=5, ) sws.on_open = self._on_open sws.on_data = self._on_data sws.on_error = self._on_error sws.on_close = self._on_close self._sws = sws logger.info("[LiveFeed] WebSocket connecting…") sws.connect() except Exception as exc: self._connected = False logger.error(f"[LiveFeed] WebSocket thread error: {exc}") self._ws_thread = threading.Thread( target=_run, name="cv6-live-feed", daemon=True ) self._ws_thread.start() # ── TIER-2 (2B): reconnect supervisor ─────────────────────────────────── _SUPERVISOR_TICK = 5 # seconds between health checks _RECONNECT_MAX_BACKOFF = 300 def _start_supervisor(self) -> None: if self._supervisor_thread and self._supervisor_thread.is_alive(): return self._stop_supervisor = False self._supervisor_thread = threading.Thread( target=self._supervise, name="cv6-feed-supervisor", daemon=True ) self._supervisor_thread.start() logger.info("[LiveFeed] Reconnect supervisor started") def _supervise(self) -> None: """Watchdog: while the feed is supposed to be up, reconnect it with exponential backoff whenever it drops — the SmartApi SDK only retries 5 times internally and then gives up, after which nothing else would restart the feed.""" backoff = self._SUPERVISOR_TICK while not self._stop_supervisor: time.sleep(self._SUPERVISOR_TICK) if not self._should_run or self._connected: backoff = self._SUPERVISOR_TICK continue logger.warning(f"[LiveFeed] Supervisor: feed is DOWN — attempting reconnect") if self._reconnect(): logger.info("[LiveFeed] Supervisor: reconnect succeeded") backoff = self._SUPERVISOR_TICK else: time.sleep(backoff) backoff = min(backoff * 2, self._RECONNECT_MAX_BACKOFF) def _reconnect(self) -> bool: with self._reconnect_lock: if self._connected: return True broker = self._broker if broker is None: return False try: # Refresh the broker session so we get a fresh feed/jwt token # (a dropped feed is often a symptom of an expired session). if hasattr(broker, "validate_session"): broker.validate_session() if not broker.is_connected(): return False self._auth_token = getattr(broker, "jwt_token", self._auth_token) self._feed_token = getattr(broker, "feed_token", self._feed_token) self._client_code = getattr(broker, "client_id", self._client_code) try: if self._sws is not None: self._sws.close_connection() except Exception: pass self._sws = None self._spawn_ws() time.sleep(2) # allow _on_open (which re-subscribes) to fire return self._connected except Exception as e: logger.error(f"[LiveFeed] reconnect failed: {e}") return False # ── WebSocket callbacks ─────────────────────────────────────────────────── def _on_open(self, wsapp): self._connected = True logger.info("[LiveFeed] ✓ WebSocket open") # Re-subscribe on reconnect if self._subscribed: self._do_subscribe(list(self._subscribed)) def _on_data(self, wsapp, message): """Parse incoming tick message and store in cache.""" try: # SmartWebSocketV2 delivers already-parsed dict msg = message if isinstance(message, dict) else {} token = str(msg.get("token", "")) if not token: return with self._lock: sym = self._rev_map.get(token) if not sym: return # Prices are in paise — divide by 100 for rupees def _p(key: str) -> float: val = msg.get(key, 0) return round(float(val) / 100, 2) if val else 0.0 ltp = _p("last_traded_price") open_ = _p("open_price_of_the_day") high_ = _p("high_price_of_the_day") low_ = _p("low_price_of_the_day") close_= _p("closed_price") vol = int(msg.get("volume_trade_for_the_day", 0)) chg_pct = ( round((ltp - close_) / close_ * 100, 2) if close_ else 0.0 ) tick = Tick( symbol = sym, token = token, ltp = ltp, open = open_, high = high_, low = low_, close = close_, volume = vol, change_pct = chg_pct, timestamp = time.time(), ) with self._lock: self._tick_cache[sym] = tick self._last_tick_ts = tick.timestamp except Exception as exc: logger.debug(f"[LiveFeed] Tick parse error: {exc}") def _on_error(self, wsapp, error): logger.error(f"[LiveFeed] WebSocket error: {error}") self._connected = False def _on_close(self, wsapp): # TIER-2 (2B): the app-level supervisor (not just the SDK's 5 retries) # will re-establish the feed while _should_run is True. logger.warning("[LiveFeed] WebSocket closed — supervisor will reconnect") self._connected = False # ── Subscription ───────────────────────────────────────────────────────── def subscribe(self, symbols: List[str]) -> None: """ Subscribe to live ticks for a list of NSE symbols. Symbols not yet token-mapped are looked up automatically. Safe to call before WebSocket is open (tokens queued). """ new_tokens: List[str] = [] for sym in symbols: tok = self.get_token(sym) if tok and tok not in self._subscribed: self._subscribed.add(tok) new_tokens.append(tok) if new_tokens: if self._connected and self._sws: self._do_subscribe(new_tokens) else: logger.debug( f"[LiveFeed] Queued {len(new_tokens)} tokens " f"(WebSocket not yet open)" ) def _do_subscribe(self, tokens: List[str]) -> None: """Send subscribe request to the open WebSocket.""" try: # Split by exchange type index_toks = [t for t in tokens if t in _INDEX_TOKENS.values()] equity_toks = [t for t in tokens if t not in _INDEX_TOKENS.values()] token_list: List[dict] = [] if index_toks: token_list.append({ "exchangeType": _EXCH_NSE, "tokens": index_toks, }) if equity_toks: token_list.append({ "exchangeType": _EXCH_NSE, "tokens": equity_toks, }) if token_list: self._sws.subscribe("cv6_feed", _MODE_QUOTE, token_list) logger.info( f"[LiveFeed] Subscribed: {len(tokens)} tokens " f"({len(index_toks)} indices, {len(equity_toks)} equities)" ) except Exception as exc: logger.error(f"[LiveFeed] subscribe() error: {exc}") def unsubscribe(self, symbols: List[str]) -> None: """Remove symbols from live feed.""" tokens_to_drop: List[str] = [] for sym in symbols: tok = self._token_map.get(sym.upper().strip()) if tok and tok in self._subscribed: self._subscribed.discard(tok) tokens_to_drop.append(tok) if tokens_to_drop and self._connected and self._sws: try: token_list = [{"exchangeType": _EXCH_NSE, "tokens": tokens_to_drop}] self._sws.unsubscribe("cv6_feed", _MODE_QUOTE, token_list) except Exception as exc: logger.debug(f"[LiveFeed] unsubscribe error: {exc}") # ── Public data API ─────────────────────────────────────────────────────── def get_ltp(self, symbol: str) -> Optional[float]: """ Get last traded price for a symbol. Priority: 1. WebSocket tick cache (real-time, freshest) 2. Angel One REST ltpData (reliable, slightly slower) 3. yfinance (delayed ~15 min, last resort) Returns None if all three fail. """ sym = symbol.upper().strip() # ── 1. WebSocket cache ─────────────────────────────────────────────── with self._lock: tick = self._tick_cache.get(sym) if tick and tick.is_fresh and tick.ltp > 0: return tick.ltp # Symbol not subscribed yet → subscribe now for future calls if self.get_token(sym) and sym not in [ self._rev_map.get(t) for t in self._subscribed ]: self.subscribe([sym]) # ── 2. Angel One REST ltpData (5-second timeout to avoid thread exhaustion) try: from app.api.broker_api import manager as _mgr if _mgr.connected and _mgr.broker_name == "angelone": tok = self.get_token(sym) if tok: import concurrent.futures _ex = concurrent.futures.ThreadPoolExecutor(max_workers=1) fut = _ex.submit(_mgr.broker.quotes, { "exchange": "NSE", "tradingsymbol": f"{sym}-EQ", "symboltoken": tok, }) resp = None try: resp = fut.result(timeout=5) except concurrent.futures.TimeoutError: logger.debug(f"[LiveFeed] REST fallback timeout for {sym}") finally: # wait=False: don't block the calling thread while DNS resolves _ex.shutdown(wait=False) if resp and resp.get("status") and resp.get("data"): ltp = float(resp["data"].get("ltp", 0)) if ltp > 0: logger.debug( f"[LiveFeed] REST fallback {sym}: ₹{ltp:.2f}" ) return ltp except Exception as _rest_err: logger.debug(f"[LiveFeed] REST fallback error for {sym}: {_rest_err}") # ── 3. yfinance fallback (delayed) ─────────────────────────────────── try: import yfinance as yf info = yf.Ticker(f"{sym}.NS").fast_info ltp = float(getattr(info, "last_price", 0) or 0) if ltp > 0: logger.debug( f"[LiveFeed] yfinance fallback {sym}: ₹{ltp:.2f} (delayed)" ) return ltp except Exception: pass return None def get_tick(self, symbol: str) -> Optional[Tick]: """Return the full Tick object (OHLCV + change_pct).""" with self._lock: return self._tick_cache.get(symbol.upper().strip()) def get_price_data(self, symbol: str) -> dict: """ Return a price dict compatible with market_api.get_live_price() format. Used as drop-in by the market API. """ sym = symbol.upper().strip() with self._lock: tick = self._tick_cache.get(sym) if tick and tick.is_fresh and tick.ltp > 0: return { "symbol": sym, "ltp": tick.ltp, "open": tick.open, "high": tick.high, "low": tick.low, "change": round(tick.ltp - tick.close, 2), "change_pct": tick.change_pct, "volume": tick.volume, "source": "websocket", } # Fall through to LTP-only response ltp = self.get_ltp(sym) if ltp: return { "symbol": sym, "ltp": ltp, "open": ltp, "high": ltp, "low": ltp, "change": 0.0, "change_pct": 0.0, "volume": 0, "source": "rest_fallback", } return {} # ── Status & control ────────────────────────────────────────────────────── def is_connected(self) -> bool: return self._connected def status(self) -> dict: with self._lock: cached = list(self._tick_cache.keys()) return { "connected": self._connected, "subscribed": len(self._subscribed), "cached_ticks": len(cached), "symbols": cached, "feed_token": bool(self._feed_token), } def disconnect(self) -> None: """Gracefully close the WebSocket and stop the reconnect supervisor (so an intentional disconnect isn't fought by the watchdog).""" self._should_run = False self._stop_supervisor = True try: if self._sws: self._sws.close_connection() except Exception: pass self._connected = False logger.info("[LiveFeed] Disconnected") # ── Singleton ───────────────────────────────────────────────────────────────── live_feed = LiveFeedManager()