"""BSE option chain response parser. Transforms raw JSON from BSE's ``DerivOptionChain_IV`` endpoint into a typed :class:`~indiaopt.models.option_chain.OptionChainResult`. BSE response schema (approximate — field names differ from NSE):: { "Table": [ { "StrikePrice": 74000, "CE_OpenInterest": 1234, "CE_ChgInOI": 56, "CE_IV": 12.5, "CE_LTP": 45.0, "PE_OpenInterest": 789, "PE_ChgInOI": -10, "PE_IV": 11.0, "PE_LTP": 60.5, ... } ], "Table1": [ {"UnderlyingValue": 74250.3, "ExpiryDate": "2024-06-27T00:00:00"} ] } .. note:: BSE's API schema is not officially documented. This parser is a best-effort implementation based on observed responses. If you have access to official BSE API documentation, adjust field names in :func:`_map_row` accordingly. """ from __future__ import annotations import logging from datetime import datetime, timezone from typing import Any from indiaopt.exceptions import ParseError from indiaopt.models.option_chain import OptionChainResult, OptionChainRow from indiaopt.utils.coerce import safe_float, safe_int, safe_int_or_zero logger = logging.getLogger("indiaopt.parsers.bse") EXCHANGE = "BSE" # ── BSE field name mappings ─────────────────────────────────────────────────── # BSE uses different field names than NSE. Map them here so the rest of the # parser logic stays clean. _STRIKE_FIELDS = ("StrikePrice", "strikePrice", "Strike", "strike") _CALL_OI_FIELDS = ("CE_OpenInterest", "CE_OI", "CallOI") _CALL_COI_FIELDS = ("CE_ChgInOI", "CE_ChangeInOI", "CallChgOI") _CALL_IV_FIELDS = ("CE_IV", "CallIV", "CE_ImpliedVolatility") _CALL_LTP_FIELDS = ("CE_LTP", "CallLTP", "CE_LastTradedPrice") _PUT_OI_FIELDS = ("PE_OpenInterest", "PE_OI", "PutOI") _PUT_COI_FIELDS = ("PE_ChgInOI", "PE_ChangeInOI", "PutChgOI") _PUT_IV_FIELDS = ("PE_IV", "PutIV", "PE_ImpliedVolatility") _PUT_LTP_FIELDS = ("PE_LTP", "PutLTP", "PE_LastTradedPrice") def _first(d: dict[str, Any], *keys: str) -> Any: """Return the first non-None value found in *d* for any of *keys*.""" for k in keys: if k in d and d[k] is not None: return d[k] return None def parse_option_chain(raw: dict[str, Any], symbol: str) -> OptionChainResult: """Parse a BSE option chain response into :class:`OptionChainResult`. Args: raw: Raw JSON dict from BSE API. symbol: The trading symbol (BSE scrip code or name). Returns: Fully typed :class:`OptionChainResult`. Raises: :class:`~indiaopt.exceptions.ParseError`: If the response structure is unrecognisable. """ try: return _parse(raw, symbol) except ParseError: raise except Exception as exc: raise ParseError( f"Unexpected error parsing BSE response for {symbol}: {exc}", symbol=symbol, exchange=EXCHANGE, raw_preview=str(raw)[:200], ) from exc def _parse(raw: dict[str, Any], symbol: str) -> OptionChainResult: # BSE returns data in "Table" key; fall back to "data" for future-proofing table: list[Any] = ( raw.get("Table") or raw.get("data") or raw.get("records", {}).get("data") or [] ) if not table: logger.warning( "BSE response for %s has no data table. " "Exchange may be closed or scrip code may be incorrect.", symbol, ) # ── Meta: spot price and expiry ─────────────────────────────────────────── meta_table: list[dict[str, Any]] = raw.get("Table1") or raw.get("meta") or [] spot_price: float | None = None expiry: str | None = None for meta_row in meta_table: if spot_price is None: spot_price = safe_float( _first(meta_row, "UnderlyingValue", "SpotPrice", "LTP", "underlyingValue") ) if expiry is None: raw_expiry = _first(meta_row, "ExpiryDate", "Expiry", "expiryDate") if raw_expiry: # Normalise ISO datetime to date string "DD-Mon-YYYY" expiry = _normalise_expiry(str(raw_expiry)) if spot_price is not None and expiry is not None: break # ── Strike rows ─────────────────────────────────────────────────────────── rows: list[OptionChainRow] = [] skipped = 0 for item in table: if not isinstance(item, dict): skipped += 1 continue strike = safe_float(_first(item, *_STRIKE_FIELDS)) if strike is None: skipped += 1 continue rows.append( OptionChainRow( strike=strike, call_oi=safe_int_or_zero(_first(item, *_CALL_OI_FIELDS)), call_coi=safe_int_or_zero(_first(item, *_CALL_COI_FIELDS)), put_oi=safe_int_or_zero(_first(item, *_PUT_OI_FIELDS)), put_coi=safe_int_or_zero(_first(item, *_PUT_COI_FIELDS)), call_iv=safe_float(_first(item, *_CALL_IV_FIELDS)), put_iv=safe_float(_first(item, *_PUT_IV_FIELDS)), call_ltp=safe_float(_first(item, *_CALL_LTP_FIELDS)), put_ltp=safe_float(_first(item, *_PUT_LTP_FIELDS)), call_vol=safe_int(_first(item, "CE_Volume", "CallVolume")), put_vol=safe_int(_first(item, "PE_Volume", "PutVolume")), ) ) if skipped: logger.debug("Skipped %d malformed rows for BSE/%s.", skipped, symbol) rows.sort(key=lambda r: r.strike) atm_strike: float | None = None if spot_price is not None and rows: atm_strike = min(rows, key=lambda r: abs(r.strike - spot_price)).strike return OptionChainResult( symbol=symbol, exchange=EXCHANGE, expiry=expiry, spot_price=spot_price, atm_strike=atm_strike, data=rows, fetched_at=datetime.now(timezone.utc), expiry_dates=[expiry] if expiry else [], ) def _normalise_expiry(raw: str) -> str: """Attempt to convert BSE expiry strings to ``DD-Mon-YYYY`` format. Handles: - ``"2024-06-27T00:00:00"`` → ``"27-Jun-2024"`` - ``"27-06-2024"`` → ``"27-Jun-2024"`` - Already-formatted strings are returned as-is. """ _MONTH_ABBR = { "01": "Jan", "02": "Feb", "03": "Mar", "04": "Apr", "05": "May", "06": "Jun", "07": "Jul", "08": "Aug", "09": "Sep", "10": "Oct", "11": "Nov", "12": "Dec", } try: # Try ISO format first dt = datetime.fromisoformat(raw.split("T")[0]) month_abbr = _MONTH_ABBR[f"{dt.month:02d}"] return f"{dt.day:02d}-{month_abbr}-{dt.year}" except Exception: return raw # return as-is if we can't parse it __all__ = ["parse_option_chain"]