| 1 | """Explicit, bounded INDIA historical-price population jobs. |
| 2 | |
| 3 | This module is deliberately separate from Sector Performance reads. It may |
| 4 | invoke portfolio-service's mapping reconciliation and Yahoo only after an |
| 5 | authenticated operational POST starts a job. |
| 6 | """ |
| 7 | from __future__ import annotations |
| 8 | |
| 9 | import asyncio |
| 10 | import logging |
| 11 | from datetime import date, datetime, timedelta, timezone |
| 12 | from typing import Any, Callable |
| 13 | from uuid import UUID, uuid4 |
| 14 | |
| 15 | import yfinance as yf |
| 16 | |
| 17 | from app.historical_market_data import ( |
| 18 | HistoricalPricePersistenceError, |
| 19 | HistoricalPricePopulationService, |
| 20 | HistoricalPriceProvider, |
| 21 | HistoricalPriceProviderError, |
| 22 | YahooHistoricalPriceProvider, |
| 23 | has_year_historical_coverage, |
| 24 | ) |
| 25 | from app.market_universe import IndiaMarketUniverseProvider, MarketUniverseInstrument, MarketUniverseUnavailable |
| 26 | from app.portfolio_orchestration import PortfolioServiceUnavailableError |
| 27 | from app.settings import Settings |
| 28 | |
| 29 | |
| 30 | _ACTIVE_STATUSES = {"QUEUED", "RUNNING"} |
| 31 | logger = logging.getLogger(__name__) |
| 32 | |
| 33 | |
| 34 | class IndiaMarketDataPopulationJobs: |
| 35 | """One bounded in-process INDIA population worker with pollable status.""" |
| 36 | |
| 37 | def __init__( |
| 38 | self, |
| 39 | repository, |
| 40 | universe_provider: IndiaMarketUniverseProvider, |
| 41 | orchestrator, |
| 42 | settings: Settings, |
| 43 | *, |
| 44 | historical_provider: HistoricalPriceProvider | None = None, |
| 45 | clock: Callable[[], datetime] | None = None, |
| 46 | sleep: Callable[[float], Any] = asyncio.sleep, |
| 47 | ) -> None: |
| 48 | self.repository = repository |
| 49 | self.universe_provider = universe_provider |
| 50 | self.orchestrator = orchestrator |
| 51 | self.settings = settings |
| 52 | self.historical_provider = historical_provider or YahooHistoricalPriceProvider(yf.Ticker) |
| 53 | self.population = HistoricalPricePopulationService(repository, self.historical_provider) |
| 54 | self._clock = clock or (lambda: datetime.now(timezone.utc)) |
| 55 | self._sleep = sleep |
| 56 | self._jobs: dict[str, dict[str, Any]] = {} |
| 57 | self._tasks: dict[str, asyncio.Task] = {} |
| 58 | self._active_job_id: str | None = None |
| 59 | self._daily_bar_lock = asyncio.Lock() |
| 60 | self._daily_bar_failures: dict[UUID, datetime] = {} |
| 61 | self._daily_bar_throttled_at: datetime | None = None |
| 62 | self._daily_bar_completed_ranges: dict[tuple, list] = {} |
| 63 | |
| 64 | async def backfill_daily_bars( |
| 65 | self, *, identity_headers: dict[str, str | None], |
| 66 | correlation_id: str | None = None, offset: int = 0, |
| 67 | instrument_ids: set[UUID] | None = None, start: date | None = None, |
| 68 | end: date | None = None, force: bool = False, |
| 69 | ) -> dict[str, Any]: |
| 70 | """Explicit bounded worker operation; never called by submit/ensure/GET. |
| 71 | |
| 72 | Optional IDs restrict canonical universe membership; they do not supply |
| 73 | identity. State/cooldowns are process-local, like existing population jobs. |
| 74 | """ |
| 75 | from app.nse_daily_backfill import run_backfill |
| 76 | |
| 77 | async with self._daily_bar_lock: |
| 78 | return await run_backfill(self, identity_headers=identity_headers, |
| 79 | correlation_id=correlation_id, offset=offset, instrument_ids=instrument_ids, |
| 80 | start=start, end=end, force=force) |
| 81 | |
| 82 | async def populate_benchmark_history(self, instrument_ids: set[UUID], *, start: date, end: date, |
| 83 | identity_headers: dict[str, str | None], correlation_id=None): |
| 84 | """Explicit bounded index acquisition; registration is a separate master operation.""" |
| 85 | from app.nse_index_history import NseIndexHistoryProvider, ENDPOINT |
| 86 | from app.nse_historical_daily import NseHistoricalResult, persist_daily_result |
| 87 | if len(instrument_ids) > self.settings.market_data_population_batch_size: |
| 88 | raise ValueError('BENCHMARK_BATCH_LIMIT') |
| 89 | results = [] |
| 90 | async with self._daily_bar_lock: |
| 91 | provider = NseIndexHistoryProvider(self.orchestrator, self.settings) |
| 92 | try: |
| 93 | for key in sorted(instrument_ids, key=str): |
| 94 | now = self._clock() |
| 95 | failures = [t for t in (self._daily_bar_failures.get(key), self._daily_bar_throttled_at) if t] |
| 96 | failed_at = max(failures) if failures else None |
| 97 | if failed_at and now - failed_at < timedelta(hours=self.settings.market_data_population_retry_cooldown_hours): |
| 98 | results.append(NseHistoricalResult(key, start, end, source_url=ENDPOINT, failure_reason='RETRY_COOLDOWN')) |
| 99 | continue |
| 100 | result = await provider.fetch(key, start=start, end=end, |
| 101 | identity_headers=identity_headers, correlation_id=correlation_id) |
| 102 | result = await persist_daily_result(self.repository, result) |
| 103 | results.append(result) |
| 104 | if result.failure_reason: |
| 105 | self._daily_bar_failures[key] = self._clock() |
| 106 | if '429' in result.failure_reason: |
| 107 | self._daily_bar_throttled_at = self._clock() |
| 108 | finally: |
| 109 | await provider.aclose() |
| 110 | await self._sleep(self.settings.market_data_population_request_interval_seconds) |
| 111 | return results |
| 112 | |
| 113 | async def populate_daily_bars( |
| 114 | self, global_instrument_id: UUID, *, start: date, end: date, |
| 115 | identity_headers: dict[str, str | None], correlation_id: str | None = None, |
| 116 | ): |
| 117 | """Explicit one-instrument acquisition; never invoked by submit/ensure. |
| 118 | |
| 119 | Reuse canonical metadata and the daily persistence boundary without |
| 120 | changing historical-close provider priority or dual-writing observations. |
| 121 | """ |
| 122 | from app.nse_historical_daily import NseHistoricalDailyProvider, persist_daily_result |
| 123 | |
| 124 | async with self._daily_bar_lock: |
| 125 | provider = NseHistoricalDailyProvider(self.orchestrator, self.settings) |
| 126 | try: |
| 127 | result = await provider.fetch(global_instrument_id, start=start, end=end, |
| 128 | identity_headers=identity_headers, correlation_id=correlation_id) |
| 129 | return await persist_daily_result(self.repository, result) |
| 130 | finally: |
| 131 | await provider.aclose() |
| 132 | await self._sleep(self.settings.market_data_population_request_interval_seconds) |
| 133 | |
| 134 | async def submit( |
| 135 | self, |
| 136 | *, |
| 137 | identity_headers: dict[str, str | None], |
| 138 | correlation_id: str | None = None, |
| 139 | ) -> dict[str, Any]: |
| 140 | if self._active_job_id is not None: |
| 141 | active = self._jobs.get(self._active_job_id) |
| 142 | if active is not None and active["status"] in _ACTIVE_STATUSES: |
| 143 | return _public_job(active) |
| 144 | |
| 145 | job_id = str(uuid4()) |
| 146 | job: dict[str, Any] = { |
| 147 | "jobId": job_id, |
| 148 | "status": "QUEUED", |
| 149 | "region": "INDIA", |
| 150 | "universeSize": 0, |
| 151 | "attempted": 0, |
| 152 | "populated": 0, |
| 153 | "skipped": 0, |
| 154 | "failed": 0, |
| 155 | "reasons": {}, |
| 156 | "_failureStages": {}, |
| 157 | "_failureClasses": {}, |
| 158 | "startedAt": None, |
| 159 | "completedAt": None, |
| 160 | } |
| 161 | self._jobs[job_id] = job |
| 162 | self._active_job_id = job_id |
| 163 | task = asyncio.create_task( |
| 164 | self._run(job, dict(identity_headers), correlation_id), |
| 165 | name=f"india-market-data-{job_id}", |
| 166 | ) |
| 167 | self._tasks[job_id] = task |
| 168 | task.add_done_callback(lambda _task, key=job_id: self._tasks.pop(key, None)) |
| 169 | return _public_job(job) |
| 170 | |
| 171 | def get(self, job_id: UUID | str) -> dict[str, Any] | None: |
| 172 | job = self._jobs.get(str(job_id)) |
| 173 | return _public_job(job) if job is not None else None |
| 174 | |
| 175 | def active(self) -> dict[str, Any] | None: |
| 176 | if self._active_job_id is None: |
| 177 | return None |
| 178 | job = self._jobs.get(self._active_job_id) |
| 179 | return _public_job(job) if job is not None and job["status"] in _ACTIVE_STATUSES else None |
| 180 | |
| 181 | def latest(self) -> dict[str, Any] | None: |
| 182 | if not self._jobs: |
| 183 | return None |
| 184 | job_id = next(reversed(self._jobs)) |
| 185 | return _public_job(self._jobs[job_id]) |
| 186 | |
| 187 | async def wait(self, job_id: UUID | str) -> dict[str, Any] | None: |
| 188 | task = self._tasks.get(str(job_id)) |
| 189 | if task is not None: |
| 190 | await task |
| 191 | return self.get(job_id) |
| 192 | |
| 193 | async def _run( |
| 194 | self, |
| 195 | job: dict[str, Any], |
| 196 | identity_headers: dict[str, str | None], |
| 197 | correlation_id: str | None, |
| 198 | ) -> None: |
| 199 | job.update(status="RUNNING", startedAt=self._clock()) |
| 200 | try: |
| 201 | universe = await self.universe_provider.listings( |
| 202 | correlation_id=correlation_id, |
| 203 | identity_headers=identity_headers, |
| 204 | ) |
| 205 | except MarketUniverseUnavailable: |
| 206 | _record_reason(job, "UNIVERSE_UNAVAILABLE") |
| 207 | job.update(status="FAILED", failed=1, completedAt=self._clock()) |
| 208 | _log_job_completed(job) |
| 209 | return |
| 210 | |
| 211 | job["universeSize"] = len(universe) |
| 212 | logger.info( |
| 213 | "market_data_population event=POPULATION_JOB_STARTED jobId=%s universeSize=%s", |
| 214 | job["jobId"], |
| 215 | job["universeSize"], |
| 216 | ) |
| 217 | batch_size = self.settings.market_data_population_batch_size |
| 218 | for offset in range(0, len(universe), batch_size): |
| 219 | batch = universe[offset:offset + batch_size] |
| 220 | for instrument in batch: |
| 221 | try: |
| 222 | await self._populate_one(job, instrument, identity_headers, correlation_id) |
| 223 | except Exception: |
| 224 | # A malformed individual record must not strand or abort a |
| 225 | # market-wide job. Known failure modes are classified in |
| 226 | # _populate_one; this is a final per-item safety boundary. |
| 227 | job["failed"] += 1 |
| 228 | _record_reason(job, "INSTRUMENT_PROCESSING_FAILED") |
| 229 | finally: |
| 230 | delay = self.settings.market_data_population_request_interval_seconds |
| 231 | if instrument.canonical_sector and delay: |
| 232 | await self._sleep(delay) |
| 233 | |
| 234 | job.update( |
| 235 | status="COMPLETED" if job["failed"] == 0 else "COMPLETED_WITH_ERRORS", |
| 236 | completedAt=self._clock(), |
| 237 | ) |
| 238 | _log_job_completed(job) |
| 239 | |
| 240 | async def _populate_one( |
| 241 | self, |
| 242 | job: dict[str, Any], |
| 243 | instrument: MarketUniverseInstrument, |
| 244 | identity_headers: dict[str, str | None], |
| 245 | correlation_id: str | None, |
| 246 | ) -> None: |
| 247 | if not instrument.canonical_sector: |
| 248 | job["skipped"] += 1 |
| 249 | _record_reason(job, "NO_CANONICAL_SECTOR") |
| 250 | return |
| 251 | |
| 252 | job["attempted"] += 1 |
| 253 | try: |
| 254 | metadata = await self.orchestrator.global_instrument_metadata( |
| 255 | instrument.global_instrument_id, |
| 256 | correlation_id=correlation_id, |
| 257 | identity_headers=identity_headers, |
| 258 | ) |
| 259 | except PortfolioServiceUnavailableError as exc: |
| 260 | job["failed"] += 1 |
| 261 | _record_reason(job, "PROVIDER_UNAVAILABLE") |
| 262 | _record_failure_stage(job, "PORTFOLIO_METADATA_UNAVAILABLE", exc) |
| 263 | return |
| 264 | |
| 265 | mapping = _verified_yahoo_nse_mapping(metadata) |
| 266 | if mapping is None: |
| 267 | try: |
| 268 | metadata = await self.orchestrator.reconcile_global_instrument_metadata( |
| 269 | instrument.global_instrument_id, |
| 270 | correlation_id=correlation_id, |
| 271 | identity_headers=identity_headers, |
| 272 | ) |
| 273 | except PortfolioServiceUnavailableError as exc: |
| 274 | job["failed"] += 1 |
| 275 | _record_reason(job, "PROVIDER_UNAVAILABLE") |
| 276 | _record_failure_stage(job, "PORTFOLIO_RECONCILE_UNAVAILABLE", exc) |
| 277 | return |
| 278 | mapping = _verified_yahoo_nse_mapping(metadata) |
| 279 | |
| 280 | if mapping is None: |
| 281 | job["skipped"] += 1 |
| 282 | _record_reason(job, "NO_VERIFIED_HISTORICAL_MAPPING") |
| 283 | return |
| 284 | |
| 285 | now = self._clock() |
| 286 | try: |
| 287 | existing = (await self.repository.market_price_observations_for_instruments( |
| 288 | {instrument.global_instrument_id} |
| 289 | )).get(instrument.global_instrument_id, []) |
| 290 | except Exception: |
| 291 | job["failed"] += 1 |
| 292 | _record_reason(job, "PERSISTENCE_UNAVAILABLE") |
| 293 | return |
| 294 | initial_start = now - timedelta(days=self.settings.market_data_population_initial_lookback_days) |
| 295 | observed_at = [value.observed_at for value in existing] |
| 296 | year_coverage = bool(observed_at) and has_year_historical_coverage( |
| 297 | min(observed_at), |
| 298 | max(observed_at), |
| 299 | len(observed_at), |
| 300 | ) |
| 301 | if year_coverage: |
| 302 | start = max(observed_at) + timedelta(days=1) |
| 303 | else: |
| 304 | start = initial_start |
| 305 | end = now + timedelta(days=1) |
| 306 | |
| 307 | # Yahoo receives date-only bounds with an exclusive end. A YEAR-capable |
| 308 | # series with no remaining date interval is a successful idempotent no-op. |
| 309 | if year_coverage and start.date() >= end.date(): |
| 310 | job["populated"] += 1 |
| 311 | return |
| 312 | |
| 313 | payload = instrument.as_payload() |
| 314 | payload["structuredProviderTicker"] = mapping |
| 315 | try: |
| 316 | written = await self.population.populate([payload], start=start, end=end) |
| 317 | except HistoricalPriceProviderError as exc: |
| 318 | job["failed"] += 1 |
| 319 | _record_reason(job, "PROVIDER_UNAVAILABLE") |
| 320 | _record_failure_stage(job, "HISTORICAL_PROVIDER_UNAVAILABLE", exc) |
| 321 | return |
| 322 | except HistoricalPricePersistenceError: |
| 323 | job["failed"] += 1 |
| 324 | _record_reason(job, "PERSISTENCE_UNAVAILABLE") |
| 325 | return |
| 326 | |
| 327 | if written > 0: |
| 328 | job["populated"] += 1 |
| 329 | else: |
| 330 | job["skipped"] += 1 |
| 331 | _record_reason(job, "NO_VALID_HISTORY") |
| 332 | |
| 333 | |
| 334 | def _verified_yahoo_nse_mapping(metadata: dict[str, Any]) -> str | None: |
| 335 | """Return only portfolio-service-owned, verified Yahoo NSE identity.""" |
| 336 | for mapping in metadata.get("providerMappings", []) if isinstance(metadata, dict) else []: |
| 337 | if not isinstance(mapping, dict): |
| 338 | continue |
| 339 | symbol = str(mapping.get("providerSymbol") or "").strip() |
| 340 | if ( |
| 341 | str(mapping.get("provider") or "").strip().upper() == "YAHOO_FINANCE" |
| 342 | and str(mapping.get("status") or "").strip().upper() == "VERIFIED" |
| 343 | and symbol.upper().endswith(".NS") |
| 344 | ): |
| 345 | return symbol |
| 346 | return None |
| 347 | |
| 348 | |
| 349 | def _record_reason(job: dict[str, Any], reason: str) -> None: |
| 350 | reasons = job["reasons"] |
| 351 | reasons[reason] = reasons.get(reason, 0) + 1 |
| 352 | |
| 353 | |
| 354 | def _record_failure_stage(job: dict[str, Any], stage: str, exc: Exception) -> None: |
| 355 | stages = job["_failureStages"] |
| 356 | stages[stage] = stages.get(stage, 0) + 1 |
| 357 | cause = exc.__cause__ |
| 358 | exception_class = type(cause).__name__ if cause is not None else type(exc).__name__ |
| 359 | classes = job["_failureClasses"] |
| 360 | key = f"{stage}:{exception_class}" |
| 361 | classes[key] = classes.get(key, 0) + 1 |
| 362 | |
| 363 | |
| 364 | def _log_job_completed(job: dict[str, Any]) -> None: |
| 365 | logger.info( |
| 366 | "market_data_population event=POPULATION_JOB_COMPLETED jobId=%s status=%s " |
| 367 | "attempted=%s populated=%s skipped=%s failed=%s reasons=%s " |
| 368 | "failureStages=%s failureClasses=%s", |
| 369 | job["jobId"], |
| 370 | job["status"], |
| 371 | job["attempted"], |
| 372 | job["populated"], |
| 373 | job["skipped"], |
| 374 | job["failed"], |
| 375 | job["reasons"], |
| 376 | job["_failureStages"], |
| 377 | job["_failureClasses"], |
| 378 | ) |
| 379 | |
| 380 | |
| 381 | def _public_job(job: dict[str, Any]) -> dict[str, Any]: |
| 382 | value = dict(job) |
| 383 | value["reasons"] = dict(job["reasons"]) |
| 384 | value.pop("_failureStages", None) |
| 385 | value.pop("_failureClasses", None) |
| 386 | return value |