""" ========================================= CV6 AI Trading OS — F&O Symbol Resolver ========================================= Angel One NFO-ல trade பண்ண symbol token வேணும். இந்த module: • Angel One scrip master download பண்றது • NIFTY/BANKNIFTY CE/PE strikes → token resolve பண்றது • Expiry dates கண்டுபிடிக்குது • Lot sizes return பண்றது ========================================= """ from __future__ import annotations import json import os import re import time import threading from datetime import datetime, date, timedelta from typing import Optional, Dict, List, Tuple from dataclasses import dataclass import requests from loguru import logger # ── Angel One Scrip Master URL ──────────────────────────────────────────────── _SCRIP_MASTER_URL = ( "https://margincalculator.angelbroking.com/OpenAPI_File/files/OpenAPIScripMaster.json" ) _SCRIP_CACHE_FILE = os.path.join( os.path.dirname(__file__), "..", "..", "scrip_master_nfo.json" ) _SCRIP_TTL = 86400 # refresh once per day # ── Lot sizes (NSE standard as of 2025) ────────────────────────────────────── LOT_SIZES: Dict[str, int] = { "NIFTY": 75, "BANKNIFTY": 30, "FINNIFTY": 65, "MIDCPNIFTY": 75, "SENSEX": 10, # BSE } # ── Index → exchange map ────────────────────────────────────────────────────── INDEX_EXCHANGE: Dict[str, str] = { "NIFTY": "NFO", "BANKNIFTY": "NFO", "FINNIFTY": "NFO", "MIDCPNIFTY": "NFO", "SENSEX": "BFO", } @dataclass class OptionContract: symbol: str # e.g. NIFTY07AUG25C22500 token: str # Angel One token strike: float option_type: str # CE | PE expiry: str # DD-Mon-YYYY expiry_date: date lot_size: int exchange: str # NFO | BFO underlying: str # NIFTY | BANKNIFTY # ── Thread-safe scrip master cache ──────────────────────────────────────────── _lock = threading.Lock() _scrip_data: List[dict] = [] _scrip_loaded = 0.0 # timestamp def _load_scrip_master(force: bool = False) -> List[dict]: """Download and cache Angel One scrip master (NFO only).""" global _scrip_data, _scrip_loaded with _lock: now = time.time() if not force and _scrip_data and (now - _scrip_loaded) < _SCRIP_TTL: return _scrip_data # Try disk cache first cache_path = os.path.abspath(_SCRIP_CACHE_FILE) if ( not force and os.path.exists(cache_path) and (now - os.path.getmtime(cache_path)) < _SCRIP_TTL ): try: with open(cache_path, "r") as f: _scrip_data = json.load(f) _scrip_loaded = now logger.info(f"[FnO Symbols] Loaded {len(_scrip_data)} NFO contracts from disk cache") return _scrip_data except Exception as e: logger.warning(f"[FnO Symbols] Disk cache read failed: {e}") # Download fresh try: logger.info("[FnO Symbols] Downloading Angel One scrip master...") r = requests.get(_SCRIP_MASTER_URL, timeout=30) r.raise_for_status() all_data = r.json() # Filter NFO + BFO only to keep memory small nfo_data = [ s for s in all_data if s.get("exch_seg") in ("NFO", "BFO") and s.get("instrumenttype") in ("OPTIDX", "OPTSTK") ] _scrip_data = nfo_data _scrip_loaded = now # Save to disk try: with open(cache_path, "w") as f: json.dump(nfo_data, f) except Exception: pass logger.info(f"[FnO Symbols] Downloaded {len(nfo_data)} options contracts") return _scrip_data except Exception as e: logger.error(f"[FnO Symbols] Scrip master download failed: {e}") return _scrip_data or [] # ── Expiry helpers ───────────────────────────────────────────────────────────── def _parse_expiry(expiry_str: str) -> Optional[date]: """Parse Angel One expiry string to date. Formats: '28AUG2025', '07AUG2025'""" try: return datetime.strptime(expiry_str.strip(), "%d%b%Y").date() except Exception: try: return datetime.strptime(expiry_str.strip(), "%d-%b-%Y").date() except Exception: return None def get_upcoming_expiries(underlying: str = "NIFTY", count: int = 4) -> List[date]: """Return next N expiry dates for the underlying.""" scrip = _load_scrip_master() ul = underlying.upper() today = date.today() expiries = set() for s in scrip: name = s.get("name", "").upper() if name != ul: continue exp = _parse_expiry(s.get("expiry", "")) if exp and exp >= today: expiries.add(exp) return sorted(expiries)[:count] def get_nearest_expiry(underlying: str = "NIFTY") -> Optional[date]: """Return the nearest (current week) expiry.""" expiries = get_upcoming_expiries(underlying, count=1) return expiries[0] if expiries else None # ── Strike resolver ──────────────────────────────────────────────────────────── def resolve_option( underlying: str, strike: float, option_type: str, # CE | PE expiry_date: Optional[date] = None, ) -> Optional[OptionContract]: """ Resolve a Nifty/BankNifty option to an Angel One OptionContract. Example: resolve_option("NIFTY", 22500, "CE") → OptionContract(symbol="NIFTY07AUG25C22500", token="58627", ...) """ scrip = _load_scrip_master() ul = underlying.upper() ot = option_type.upper() # CE or PE if expiry_date is None: expiry_date = get_nearest_expiry(ul) if expiry_date is None: logger.error(f"[FnO Symbols] No expiry found for {ul}") return None strike_int = int(round(strike, 0)) lot_size = LOT_SIZES.get(ul, 50) exchange = INDEX_EXCHANGE.get(ul, "NFO") # Find matching contract for s in scrip: if s.get("name", "").upper() != ul: continue if s.get("exch_seg") != exchange: continue exp = _parse_expiry(s.get("expiry", "")) if exp != expiry_date: continue # Strike match (within ₹0.5 to handle float precision) try: s_strike = float(s.get("strike", 0)) / 100 # Angel One stores strike × 100 except Exception: continue if abs(s_strike - strike_int) > 0.5: continue # Option type sym = s.get("symbol", "") if ot == "CE" and ("CE" not in sym.upper() and "C" not in sym[-10:].upper()): continue if ot == "PE" and ("PE" not in sym.upper() and "P" not in sym[-10:].upper()): continue return OptionContract( symbol = sym, token = str(s.get("token", "")), strike = s_strike, option_type = ot, expiry = expiry_date.strftime("%d-%b-%Y").upper(), expiry_date = expiry_date, lot_size = int(s.get("lotsize", lot_size)), exchange = exchange, underlying = ul, ) logger.warning( f"[FnO Symbols] Not found: {ul} {strike_int} {ot} exp={expiry_date}" ) return None def get_atm_strike(underlying: str, spot_price: float) -> float: """Round spot price to nearest strike (50 for Nifty, 100 for BankNifty).""" step = 100 if "BANK" in underlying.upper() else 50 return round(round(spot_price / step) * step, 0) def get_otm_strike( underlying: str, spot_price: float, option_type: str, steps: int = 2, ) -> float: """Get OTM strike N steps away from ATM.""" step = 100 if "BANK" in underlying.upper() else 50 atm = get_atm_strike(underlying, spot_price) if option_type.upper() == "CE": return atm + (step * steps) else: return atm - (step * steps) def available_strikes( underlying: str, expiry_date: Optional[date] = None, num_strikes: int = 10, ) -> List[float]: """Return list of available strikes for an underlying around ATM.""" scrip = _load_scrip_master() ul = underlying.upper() if expiry_date is None: expiry_date = get_nearest_expiry(ul) strikes = set() for s in scrip: if s.get("name", "").upper() != ul: continue exp = _parse_expiry(s.get("expiry", "")) if exp != expiry_date: continue try: st = float(s.get("strike", 0)) / 100 strikes.add(st) except Exception: pass return sorted(strikes) # ── Singleton ───────────────────────────────────────────────────────────────── fno_symbols = type("FnoSymbols", (), { "resolve": staticmethod(resolve_option), "atm_strike": staticmethod(get_atm_strike), "otm_strike": staticmethod(get_otm_strike), "nearest_expiry": staticmethod(get_nearest_expiry), "upcoming_expiries": staticmethod(get_upcoming_expiries), "strikes": staticmethod(available_strikes), "lot_size": staticmethod(lambda ul: LOT_SIZES.get(ul.upper(), 50)), "load": staticmethod(_load_scrip_master), })()