"""BSEClient — async client for BSE option chain data. Usage:: import asyncio from indiaopt import BSEClient async def main(): async with BSEClient() as client: # BSE uses numeric scrip codes, e.g. "999920" for SENSEX derivatives result = await client.fetch_option_chain("999920", is_index=True) print(f"Spot: {result.spot_price}, ATM: {result.atm_strike}") asyncio.run(main()) .. note:: BSE's option chain API uses scrip codes rather than symbol names. The mapping between names (e.g. "SENSEX") and scrip codes is not part of this library — you'll need to maintain your own mapping or use BSE's instrument list API. """ from __future__ import annotations import asyncio import logging from concurrent.futures import ThreadPoolExecutor from typing import Any from indiaopt.circuit import CircuitBreaker, get_default_breaker from indiaopt.config.settings import Settings, get_settings from indiaopt.constants import ( BSE_BASE_URL, BSE_OPTION_CHAIN_PATH, RATE_LIMIT_STATUS_CODES, ) from indiaopt.exceptions import ( FetchError, FetchTimeoutError, NetworkError, ParseError, RateLimitError, ) from indiaopt.http.retry import RetryPolicy from indiaopt.http.session import create_bse_session from indiaopt.models.option_chain import OptionChainResult from indiaopt.parsers import bse as bse_parser logger = logging.getLogger("indiaopt.exchanges.bse") EXCHANGE = "BSE" class BSEClient: """Async client for BSE option chain data. Mirrors the :class:`~indiaopt.exchanges.nse.NSEClient` interface exactly so the two are interchangeable when accessed through the :class:`~indiaopt.exchanges.base.ExchangeClient` Protocol. Args: settings: Optional :class:`~indiaopt.config.settings.Settings` override. breaker: Optional :class:`~indiaopt.circuit.CircuitBreaker` override. Context manager (recommended):: async with BSEClient() as client: result = await client.fetch_option_chain("999920", is_index=True) """ def __init__( self, settings: Settings | None = None, breaker: CircuitBreaker | None = None, ) -> None: self._settings: Settings = settings or get_settings() self._breaker: CircuitBreaker = breaker or get_default_breaker( failure_threshold=self._settings.circuit_failure_threshold, backoff_seconds=self._settings.circuit_backoff_seconds, ) self._retry_policy = RetryPolicy.from_settings(self._settings) self._pool = ThreadPoolExecutor( max_workers=self._settings.thread_pool_workers, thread_name_prefix="indiaopt-bse", ) self._session: Any = None async def __aenter__(self) -> BSEClient: return self async def __aexit__(self, *_: object) -> None: await self.close() async def close(self) -> None: """Release all resources (session, thread pool).""" if self._session is not None: try: self._session.close() except Exception: pass self._session = None self._pool.shutdown(wait=False) async def fetch_raw_option_chain( self, symbol: str, *, is_index: bool = True, expiry: str | None = None, ) -> dict[str, Any]: """Fetch the unparsed raw JSON dictionary for *symbol* from BSE.""" symbol = symbol.strip() self._breaker.raise_if_open(EXCHANGE, symbol) loop = asyncio.get_running_loop() try: return await asyncio.wait_for( loop.run_in_executor( self._pool, self._fetch_sync, symbol, is_index, expiry, ), timeout=self._settings.async_timeout, ) except asyncio.TimeoutError: self._breaker.record_failure(EXCHANGE, symbol) raise FetchTimeoutError( symbol=symbol, timeout_s=self._settings.async_timeout, exchange=EXCHANGE, ) async def fetch_option_chain( self, symbol: str, *, is_index: bool = True, expiry: str | None = None, ) -> OptionChainResult: """Fetch the full option chain for *symbol* from BSE. Args: symbol: BSE scrip code (e.g. ``"999920"`` for SENSEX derivatives). is_index: ``True`` for index instruments, ``False`` for equities. expiry: Specific expiry date string. Uses today's date when not provided. Returns: :class:`~indiaopt.models.option_chain.OptionChainResult` """ symbol = symbol.strip() self._breaker.raise_if_open(EXCHANGE, symbol) loop = asyncio.get_running_loop() try: raw = await asyncio.wait_for( loop.run_in_executor( self._pool, self._fetch_sync, symbol, is_index, expiry, ), timeout=self._settings.async_timeout, ) except asyncio.TimeoutError: self._breaker.record_failure(EXCHANGE, symbol) raise FetchTimeoutError( symbol=symbol, timeout_s=self._settings.async_timeout, exchange=EXCHANGE, ) return bse_parser.parse_option_chain(raw, symbol) async def get_expiry_dates(self, symbol: str) -> list[str]: """Return available expiry dates for *symbol*. BSE does not have a standard public expiry endpoint — returns an empty list with a warning. """ logger.warning( "BSE does not expose a public expiry-dates API. " "Pass the expiry date explicitly to fetch_option_chain()." ) return [] def reset_circuit(self, symbol: str) -> None: """Manually reset the circuit breaker for *symbol*.""" self._breaker.reset(EXCHANGE, symbol) def circuit_stats(self) -> dict[str, dict[str, object]]: """Return a snapshot of all circuit breaker states for monitoring.""" return self._breaker.stats() # ── Synchronous internals ───────────────────────────────────────────────── def _get_or_create_session(self) -> Any: if self._session is None: self._session = create_bse_session( proxy_urls=self._settings.proxy_urls, warmup_timeout=self._settings.warmup_timeout, ) return self._session def _fetch_sync( self, symbol: str, is_index: bool, expiry: str | None, ) -> dict[str, Any]: from datetime import datetime session = self._get_or_create_session() type_str = "Indices" if is_index else "Equities" expiry = expiry or datetime.now().strftime("%d-%b-%Y") url = ( f"{BSE_BASE_URL}{BSE_OPTION_CHAIN_PATH}" f"?Scripcode={symbol}&ExpiryDate={expiry}&Type={type_str}" ) last_exc: Exception | None = None for attempt in range(self._retry_policy.max_retries): try: resp = session.get(url, timeout=self._settings.fetch_timeout) if resp.status_code == 200: try: data: dict[str, Any] = resp.json() except Exception as json_exc: logger.warning( "JSON decode failed for BSE/%s (attempt %d): %s. Refreshing session.", symbol, attempt + 1, json_exc, ) self._session = None session = self._get_or_create_session() raise ParseError( f"Invalid JSON from BSE: {json_exc}", symbol=symbol, exchange=EXCHANGE, attempts=attempt + 1, ) from json_exc self._breaker.record_success(EXCHANGE, symbol) return data if resp.status_code in RATE_LIMIT_STATUS_CODES: last_exc = RateLimitError( f"HTTP {resp.status_code} from BSE for {symbol}", symbol=symbol, exchange=EXCHANGE, status_code=resp.status_code, attempts=attempt + 1, ) logger.warning( "Rate-limited by BSE for %s (attempt %d/%d, HTTP %d).", symbol, attempt + 1, self._retry_policy.max_retries, resp.status_code, ) if attempt < self._retry_policy.max_retries - 1: self._retry_policy.sleep_for(attempt) continue raise FetchError( f"Unexpected HTTP {resp.status_code} from BSE for {symbol}", symbol=symbol, exchange=EXCHANGE, status_code=resp.status_code, attempts=attempt + 1, ) except (FetchError, ParseError): raise except OSError as exc: last_exc = NetworkError( f"Network error for BSE/{symbol}: {exc}", symbol=symbol, exchange=EXCHANGE, attempts=attempt + 1, ) self._session = None session = self._get_or_create_session() if attempt < self._retry_policy.max_retries - 1: self._retry_policy.sleep_for(attempt) except Exception as exc: last_exc = exc if attempt < self._retry_policy.max_retries - 1: self._retry_policy.sleep_for(attempt) self._breaker.record_failure(EXCHANGE, symbol) raise FetchError( f"All {self._retry_policy.max_retries} attempts failed for BSE/{symbol}", symbol=symbol, exchange=EXCHANGE, attempts=self._retry_policy.max_retries, ) from last_exc __all__ = ["BSEClient"]