| 1 | """Official NSE daily evidence; invoked only by explicit acquisition workers.""" |
| 2 | from __future__ import annotations |
| 3 | |
| 4 | import asyncio |
| 5 | import csv |
| 6 | import io |
| 7 | import re |
| 8 | from dataclasses import dataclass, field |
| 9 | from datetime import date, datetime, timezone |
| 10 | from decimal import Decimal |
| 11 | from uuid import UUID |
| 12 | |
| 13 | import httpx |
| 14 | |
| 15 | from app.models import DailyMarketBar |
| 16 | from app.portfolio_orchestration import _trusted_provider_mapping |
| 17 | from app.research_fetching import _retry_after_seconds |
| 18 | |
| 19 | ENDPOINT = "https://www.nseindia.com/api/historicalOR/generateSecurityWiseHistoricalData" |
| 20 | BOOTSTRAP = "https://www.nseindia.com/report-detail/eq_security" |
| 21 | HEADERS = { |
| 22 | "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/131.0.0.0 Safari/537.36", |
| 23 | "Accept": "text/csv,text/html;q=0.9,*/*;q=0.8", |
| 24 | "Accept-Language": "en-US,en;q=0.9", "Referer": BOOTSTRAP, |
| 25 | } |
| 26 | REQUIRED = {"date", "symbol", "series", "open", "high", "low", "close"} |
| 27 | ALIASES = {f"{name} price": name for name in ("open", "high", "low", "close")} |
| 28 | # Accept ungrouped, Western grouping, or Indian grouping; never strip arbitrary commas. |
| 29 | NUMBER = re.compile(r"(?:[0-9]+|[0-9]{1,3}(?:,[0-9]{3})+|[0-9]{1,2}(?:,[0-9]{2})*,[0-9]{3})(?:\.[0-9]+)?\Z") |
| 30 | MONTHS = {name: i for i, name in enumerate("Jan Feb Mar Apr May Jun Jul Aug Sep Oct Nov Dec".split(), 1)} |
| 31 | |
| 32 | |
| 33 | @dataclass |
| 34 | class NseHistoricalResult: |
| 35 | global_instrument_id: UUID |
| 36 | request_from: date |
| 37 | request_to: date |
| 38 | retrieved_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) |
| 39 | provider_symbol: str | None = None |
| 40 | http_status: int | None = None |
| 41 | status: str = "UNAVAILABLE" |
| 42 | failure_reason: str | None = None |
| 43 | source_url: str = ENDPOINT |
| 44 | source_mode: str = "REAL" |
| 45 | headers: list[str] = field(default_factory=list) |
| 46 | rows_parsed: int = 0 |
| 47 | rows_rejected: int = 0 |
| 48 | rejection_reasons: dict[str, int] = field(default_factory=dict) |
| 49 | bars: list[DailyMarketBar] = field(default_factory=list) |
| 50 | persisted_rows: int = 0 |
| 51 | |
| 52 | @property |
| 53 | def rows_accepted(self): |
| 54 | return len(self.bars) |
| 55 | |
| 56 | @property |
| 57 | def first_trading_date(self): |
| 58 | return self.bars[0].trading_date if self.bars else None |
| 59 | |
| 60 | @property |
| 61 | def last_trading_date(self): |
| 62 | return self.bars[-1].trading_date if self.bars else None |
| 63 | |
| 64 | |
| 65 | def verified_identity(metadata: dict, key: UUID) -> tuple[str, str]: |
| 66 | if str(metadata.get("globalInstrumentId")) != str(key): |
| 67 | raise ValueError("IDENTITY_MISMATCH") |
| 68 | if (metadata.get("status") != "ACTIVE" or metadata.get("assetType") != "EQUITY" |
| 69 | or metadata.get("country") not in {"IN", "IND", "INDIA"} |
| 70 | or metadata.get("primaryExchange") not in {"NSE", "XNSE"}): |
| 71 | raise ValueError("UNSUPPORTED_INSTRUMENT") |
| 72 | mappings = [m for m in metadata.get("providerMappings", []) if isinstance(m, dict) |
| 73 | and str(m.get("provider", "")).upper() == "NSE"] |
| 74 | if len(mappings) != 1: |
| 75 | raise ValueError("NO_UNAMBIGUOUS_NSE_MAPPING") |
| 76 | mapping = mappings[0] |
| 77 | if not _trusted_provider_mapping(mapping) or mapping.get("active") is False: |
| 78 | raise ValueError("NO_TRUSTED_NSE_MAPPING") |
| 79 | symbol = str(mapping.get("providerSymbol") or "").strip().upper() |
| 80 | if not symbol or mapping.get("exchange") not in {None, "NSE", "XNSE"}: |
| 81 | raise ValueError("INVALID_NSE_MAPPING") |
| 82 | currencies = {str(c).strip().upper() for c in (metadata.get("currency"), mapping.get("currency")) if c and str(c).strip()} |
| 83 | if len(currencies) != 1: |
| 84 | raise ValueError("MISSING_OR_AMBIGUOUS_CURRENCY") |
| 85 | return symbol, currencies.pop() |
| 86 | |
| 87 | |
| 88 | def number(value: str | None, *, integer=False, required=False): |
| 89 | value = (value or "").strip() |
| 90 | if value in {"", "-"} and not required: |
| 91 | return None |
| 92 | if not NUMBER.fullmatch(value) or (integer and "." in value): |
| 93 | raise ValueError("MALFORMED_NUMBER") |
| 94 | return int(value.replace(",", "")) if integer else Decimal(value.replace(",", "")) |
| 95 | |
| 96 | |
| 97 | def trading_date(value: str) -> date: |
| 98 | day, month, year = value.strip().split("-") |
| 99 | return date(int(year), MONTHS[month.title()] if month.isalpha() else int(month), int(day)) |
| 100 | |
| 101 | |
| 102 | def parse_csv(content: bytes, result: NseHistoricalResult, currency: str) -> None: |
| 103 | """Header aliases are explicit; symbol/series normalization is strip + uppercase. |
| 104 | |
| 105 | Turnover ₹ is rupees (live fixture 2026-09-13), with no scaling. Other |
| 106 | turnover units, including lacs, remain unavailable until independently verified. |
| 107 | Duplicate dates are all rejected, avoiding arbitrary correction selection. |
| 108 | """ |
| 109 | text = content.decode("utf-8-sig") |
| 110 | if not text.strip(): |
| 111 | raise ValueError("EMPTY_RESPONSE") |
| 112 | if text.lstrip().startswith(("<", "{", "[")): |
| 113 | raise ValueError("NON_CSV_RESPONSE") |
| 114 | reader = csv.reader(io.StringIO(text, newline=""), strict=True) |
| 115 | result.headers = next(reader) |
| 116 | headers = [" ".join(h.strip().casefold().split()) for h in result.headers] |
| 117 | headers = [ALIASES.get(h, h) for h in headers] |
| 118 | if len(set(headers)) != len(headers) or not REQUIRED.issubset(headers): |
| 119 | raise ValueError("MISSING_OR_DUPLICATE_HEADERS") |
| 120 | candidates: dict[date, list[DailyMarketBar]] = {} |
| 121 | def reject(reason): |
| 122 | result.rows_rejected += 1 |
| 123 | result.rejection_reasons[reason] = result.rejection_reasons.get(reason, 0) + 1 |
| 124 | for values in reader: |
| 125 | if not any(v.strip() for v in values): |
| 126 | continue |
| 127 | result.rows_parsed += 1 |
| 128 | if len(values) != len(headers): |
| 129 | reject("MALFORMED_ROW") |
| 130 | continue |
| 131 | row = dict(zip(headers, (v.strip() for v in values))) |
| 132 | if row["symbol"].upper() != result.provider_symbol: |
| 133 | reject("SYMBOL_MISMATCH") |
| 134 | continue |
| 135 | if row["series"].upper() != "EQ": |
| 136 | reject("UNSUPPORTED_SERIES") |
| 137 | continue |
| 138 | try: |
| 139 | day = trading_date(row["date"]) |
| 140 | if not result.request_from <= day <= result.request_to: |
| 141 | reject("OUTSIDE_REQUEST_RANGE") |
| 142 | continue |
| 143 | bar = DailyMarketBar(global_instrument_id=result.global_instrument_id, trading_date=day, |
| 144 | **{k: number(row[k], required=True) for k in ("open", "high", "low", "close")}, |
| 145 | previous_close=number(row.get("prev close")), |
| 146 | volume=number(row.get("total traded quantity"), integer=True), |
| 147 | turnover=number(row.get("turnover ₹")) if currency == "INR" else None, |
| 148 | currency=currency, provider="NSE", provider_symbol=result.provider_symbol, |
| 149 | source_mode="REAL", source_url=result.source_url, retrieved_at=result.retrieved_at) |
| 150 | candidates.setdefault(day, []).append(bar) |
| 151 | except (ValueError, KeyError, OverflowError): |
| 152 | reject("INVALID_DATE_OR_VALUE") |
| 153 | for day, bars in sorted(candidates.items()): |
| 154 | if len(bars) != 1: |
| 155 | for _ in bars: |
| 156 | reject("DUPLICATE_DATE") |
| 157 | else: |
| 158 | result.bars.append(bars[0]) |
| 159 | |
| 160 | |
| 161 | class NseHistoricalDailyProvider: |
| 162 | provider_name = "NSE" |
| 163 | |
| 164 | def __init__(self, orchestrator, settings, *, client=None, sleep=asyncio.sleep): |
| 165 | self.orchestrator, self.settings = orchestrator, settings |
| 166 | self.client = client or httpx.AsyncClient(timeout=httpx.Timeout(20, connect=4)) |
| 167 | self._owns_client = client is None |
| 168 | self._sleep = sleep |
| 169 | self._lock = asyncio.Lock() |
| 170 | self._bootstrapped = False |
| 171 | self._last_request = None |
| 172 | |
| 173 | async def aclose(self): |
| 174 | if self._owns_client: |
| 175 | await self.client.aclose() |
| 176 | |
| 177 | async def _get(self, url, result, **kwargs): |
| 178 | for attempt in range(self.settings.nse_historical_max_retries + 1): |
| 179 | now = asyncio.get_running_loop().time() |
| 180 | if self._last_request is not None: |
| 181 | await self._sleep(max(0, self.settings.market_data_population_request_interval_seconds - (now - self._last_request))) |
| 182 | self._last_request = asyncio.get_running_loop().time() |
| 183 | try: |
| 184 | response = await self.client.get(url, headers=HEADERS, **kwargs) |
| 185 | result.http_status = response.status_code |
| 186 | if response.status_code < 400: |
| 187 | response.raise_for_status() |
| 188 | return response |
| 189 | if response.status_code != 429 and response.status_code < 500: |
| 190 | response.raise_for_status() |
| 191 | if attempt == self.settings.nse_historical_max_retries: |
| 192 | response.raise_for_status() |
| 193 | retry_after = _retry_after_seconds(response.headers.get("retry-after")) |
| 194 | # Do not retry earlier than a long server-requested cooldown. |
| 195 | if retry_after is not None and retry_after > 30: |
| 196 | response.raise_for_status() |
| 197 | await self._sleep(max(min(2 ** attempt, 8), retry_after or 0)) |
| 198 | except (httpx.TimeoutException, httpx.NetworkError): |
| 199 | if attempt == self.settings.nse_historical_max_retries: |
| 200 | raise |
| 201 | await self._sleep(min(2 ** attempt, 8)) |
| 202 | |
| 203 | async def fetch(self, global_instrument_id: UUID, *, start: date, end: date, |
| 204 | identity_headers=None, correlation_id=None) -> NseHistoricalResult: |
| 205 | result = NseHistoricalResult(global_instrument_id, start, end) |
| 206 | stage = "IDENTITY" |
| 207 | try: |
| 208 | if type(start) is not date or type(end) is not date or start > end: |
| 209 | raise ValueError("INVALID_DATE_RANGE") |
| 210 | if (end - start).days + 1 > self.settings.nse_historical_request_window_days: |
| 211 | raise ValueError("REQUEST_WINDOW_EXCEEDED") |
| 212 | metadata = await self.orchestrator.global_instrument_metadata(global_instrument_id, |
| 213 | identity_headers=identity_headers, correlation_id=correlation_id) |
| 214 | result.provider_symbol, currency = verified_identity(metadata, global_instrument_id) |
| 215 | async with self._lock: |
| 216 | if not self._bootstrapped: |
| 217 | stage = "BOOTSTRAP" |
| 218 | await self._get(BOOTSTRAP, result) |
| 219 | self._bootstrapped = True |
| 220 | stage = "HISTORICAL" |
| 221 | response = await self._get(ENDPOINT, result, params={ |
| 222 | "from": start.strftime("%d-%m-%Y"), "to": end.strftime("%d-%m-%Y"), |
| 223 | "symbol": result.provider_symbol, "type": "priceVolumeDeliverable", "series": "EQ", "csv": "true"}) |
| 224 | result.retrieved_at = datetime.now(timezone.utc) |
| 225 | result.source_url = str(response.url) |
| 226 | if any(t in response.headers.get("content-type", "").lower() for t in ("html", "json")): |
| 227 | raise ValueError("NON_CSV_RESPONSE") |
| 228 | parse_csv(response.content, result, currency) |
| 229 | result.status = "SUCCESS" if result.bars else "UNAVAILABLE" |
| 230 | result.failure_reason = None if result.bars else "NO_VALID_HISTORY" |
| 231 | except httpx.HTTPStatusError as exc: |
| 232 | result.failure_reason = f"{stage}_HTTP_{exc.response.status_code}" |
| 233 | if exc.response.status_code == 403: |
| 234 | self._bootstrapped = False |
| 235 | except httpx.TimeoutException: |
| 236 | result.failure_reason = f"{stage}_TIMEOUT" |
| 237 | except httpx.HTTPError: |
| 238 | result.failure_reason = f"{stage}_CONNECTION_FAILURE" |
| 239 | except (ValueError, csv.Error, UnicodeError) as exc: |
| 240 | result.failure_reason = str(exc) if type(exc) is ValueError else "MALFORMED_CSV" |
| 241 | except Exception: |
| 242 | result.failure_reason = f"{stage}_UNAVAILABLE" |
| 243 | if result.failure_reason: |
| 244 | result.bars.clear() |
| 245 | return result |
| 246 | |
| 247 | |
| 248 | async def persist_daily_result(repository, result: NseHistoricalResult) -> NseHistoricalResult: |
| 249 | if result.status == "SUCCESS": |
| 250 | try: |
| 251 | result.persisted_rows = await repository.upsert_daily_market_bars_async(result.bars) |
| 252 | except Exception: |
| 253 | result.status = "UNAVAILABLE" |
| 254 | result.failure_reason = "DAILY_BAR_PERSISTENCE_UNAVAILABLE" |
| 255 | return result |