"""Typed, validated settings for indiaopt. Configuration is loaded (in priority order) from: 1. Explicit keyword arguments to :class:`Settings`. 2. Environment variables (prefixed with ``INDIAOPT_``). 3. A ``.env`` file in the current working directory. 4. Hard-coded defaults. Example .env file:: INDIAOPT_SYMBOLS=NIFTY,BANKNIFTY,RELIANCE INDIAOPT_MAX_RETRIES=5 INDIAOPT_FETCH_TIMEOUT=20.0 INDIAOPT_PROXY_URLS=http://proxy1:8080,http://proxy2:8080 INDIAOPT_LOG_LEVEL=DEBUG Example usage:: from indiaopt import Settings, get_settings # Use defaults / env vars settings = get_settings() # Or override at runtime settings = Settings(symbols=["NIFTY"], max_retries=3) """ from __future__ import annotations import functools from typing import Annotated from pydantic import Field, field_validator, model_validator from pydantic_settings import BaseSettings, SettingsConfigDict from indiaopt.constants import ( DEFAULT_ASYNC_TIMEOUT_S, DEFAULT_BACKOFF_BASE, DEFAULT_BACKOFF_CAP_S, DEFAULT_CACHE_TTL_S, DEFAULT_CIRCUIT_BACKOFF_S, DEFAULT_CIRCUIT_FAILURE_THRESHOLD, DEFAULT_EXPIRY_CACHE_TTL_S, DEFAULT_FETCH_TIMEOUT_S, DEFAULT_MAX_RETRIES, DEFAULT_THREAD_POOL_WORKERS, DEFAULT_WARMUP_TIMEOUT_S, ) class Settings(BaseSettings): """All runtime configuration for indiaopt. Fields can be set via environment variables prefixed with ``INDIAOPT_`` or via ``.env`` file. Attributes: symbols: Default list of symbols to fetch. poll_interval_seconds: Polling interval in seconds (for polling use-cases). max_retries: Maximum number of retry attempts (1-10). backoff_base: Exponential backoff base in seconds. backoff_cap_seconds: Maximum backoff sleep duration in seconds. backoff_jitter: Random jitter added to each sleep (seconds). fetch_timeout: Per-request HTTP timeout in seconds. warmup_timeout: Session warm-up HTTP timeout in seconds. async_timeout: Total coroutine-level timeout in seconds. proxy_urls: Optional HTTP proxy URLs (rotated randomly). circuit_failure_threshold: Consecutive failures before circuit opens. circuit_backoff_seconds: How long the circuit stays open (seconds). cache_ttl: TTL for option chain cache entries (seconds). expiry_cache_ttl: TTL for expiry date cache entries (seconds). thread_pool_workers: Thread pool size for sync-in-async execution. log_level: Logging level (``"DEBUG"``, ``"INFO"``, …). session_retention_hours: Hours before purging old session data. redis_url: Optional Redis URL for distributed caching. """ model_config = SettingsConfigDict( env_prefix="INDIAOPT_", env_file=".env", env_file_encoding="utf-8", case_sensitive=False, extra="ignore", populate_by_name=True, ) # ── Symbol / Polling ───────────────────────────────────────────────────── symbols: list[str] = Field( default=["NIFTY"], description="Default symbols to watch.", ) poll_interval_seconds: Annotated[int, Field(ge=10, le=86400)] = Field( default=60, description="Polling interval in seconds.", ) # ── Retry / Backoff ────────────────────────────────────────────────────── max_retries: Annotated[int, Field(ge=1, le=10)] = Field( default=DEFAULT_MAX_RETRIES, description="Maximum number of retry attempts (1–10).", ) backoff_base: Annotated[float, Field(gt=0, le=10)] = Field( default=DEFAULT_BACKOFF_BASE, description="Exponential backoff base multiplier.", ) backoff_cap_seconds: Annotated[float, Field(gt=0, le=120)] = Field( default=DEFAULT_BACKOFF_CAP_S, description="Maximum backoff sleep duration in seconds.", ) backoff_jitter: Annotated[float, Field(ge=0, le=5)] = Field( default=1.0, description="Random jitter (seconds) added to each backoff sleep.", ) # ── Timeouts ───────────────────────────────────────────────────────────── fetch_timeout: Annotated[float, Field(gt=0, le=120)] = Field( default=DEFAULT_FETCH_TIMEOUT_S, description="Per-request HTTP timeout in seconds.", ) warmup_timeout: Annotated[float, Field(gt=0, le=30)] = Field( default=DEFAULT_WARMUP_TIMEOUT_S, description="Session warm-up timeout in seconds.", ) async_timeout: Annotated[float, Field(gt=0, le=300)] = Field( default=DEFAULT_ASYNC_TIMEOUT_S, description="Total async coroutine timeout in seconds.", ) # ── Proxy ───────────────────────────────────────────────────────────────── proxy_urls: list[str] = Field( default=[], description="List of proxy URLs rotated randomly.", ) # ── Circuit Breaker ─────────────────────────────────────────────────────── circuit_failure_threshold: Annotated[int, Field(ge=1, le=50)] = Field( default=DEFAULT_CIRCUIT_FAILURE_THRESHOLD, description="Consecutive failures before circuit opens.", ) circuit_backoff_seconds: Annotated[float, Field(gt=0, le=3600)] = Field( default=DEFAULT_CIRCUIT_BACKOFF_S, description="How long the circuit stays open in seconds.", ) # ── Cache ───────────────────────────────────────────────────────────────── cache_ttl: Annotated[int, Field(ge=1, le=86400)] = Field( default=DEFAULT_CACHE_TTL_S, description="TTL for option chain cache entries in seconds.", ) expiry_cache_ttl: Annotated[int, Field(ge=60, le=86400)] = Field( default=DEFAULT_EXPIRY_CACHE_TTL_S, description="TTL for expiry date cache entries in seconds.", ) # ── Infrastructure ──────────────────────────────────────────────────────── thread_pool_workers: Annotated[int, Field(ge=1, le=64)] = Field( default=DEFAULT_THREAD_POOL_WORKERS, description="Thread pool size for sync-in-async execution.", ) log_level: str = Field( default="INFO", description="Logging level: DEBUG, INFO, WARNING, ERROR, CRITICAL.", ) session_retention_hours: Annotated[int, Field(ge=1, le=48)] = Field( default=12, description="Session data older than this is eligible for purge.", ) redis_url: str | None = Field( default=None, description="Redis URL for distributed caching (optional).", ) # ── Validators ──────────────────────────────────────────────────────────── @field_validator("symbols", mode="before") @classmethod def _parse_symbols(cls, v: object) -> list[str]: """Accept comma-separated string or list.""" if isinstance(v, str): return [s.strip().upper() for s in v.split(",") if s.strip()] if isinstance(v, (list, tuple)): return [str(s).strip().upper() for s in v if str(s).strip()] return ["NIFTY"] @field_validator("proxy_urls", mode="before") @classmethod def _parse_proxy_urls(cls, v: object) -> list[str]: """Accept comma-separated string or list.""" if isinstance(v, str): return [p.strip() for p in v.split(",") if p.strip()] if isinstance(v, (list, tuple)): return [str(p).strip() for p in v if str(p).strip()] return [] @field_validator("log_level", mode="before") @classmethod def _validate_log_level(cls, v: object) -> str: valid = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"} s = str(v).upper() if s not in valid: raise ValueError(f"log_level must be one of {valid}, got {v!r}") return s @model_validator(mode="after") def _validate_timeout_order(self) -> Settings: if self.async_timeout <= self.fetch_timeout: raise ValueError( f"async_timeout ({self.async_timeout}s) must be greater than " f"fetch_timeout ({self.fetch_timeout}s)." ) return self @functools.lru_cache(maxsize=1) def get_settings() -> Settings: """Return the cached global :class:`Settings` instance. Reads configuration from environment variables and ``.env`` file on first call. Subsequent calls return the cached instance. To reset (e.g. in tests), call ``get_settings.cache_clear()`` first:: from indiaopt.config.settings import get_settings get_settings.cache_clear() settings = get_settings() """ return Settings() __all__ = ["Settings", "get_settings"]