| 1 | """Bounded NSE daily-bar planning and execution within population jobs. |
| 2 | |
| 3 | No exchange sessions are synthesized. Coverage means observed boundary coverage, |
| 4 | not proof that every internal trading session is present. |
| 5 | """ |
| 6 | from __future__ import annotations |
| 7 | |
| 8 | from datetime import date, datetime, timedelta |
| 9 | from uuid import UUID |
| 10 | from zoneinfo import ZoneInfo |
| 11 | |
| 12 | from app.nse_historical_daily import NseHistoricalDailyProvider, persist_daily_result, verified_identity |
| 13 | |
| 14 | DAY = timedelta(days=1) |
| 15 | |
| 16 | |
| 17 | def plan_windows(start: date, end: date, size: int) -> list[tuple[date, date]]: |
| 18 | if type(start) is not date or type(end) is not date or start > end or type(size) is not int or size < 1: |
| 19 | raise ValueError('WINDOW_PLANNING_FAILURE') |
| 20 | windows = [] |
| 21 | cursor = start |
| 22 | while cursor <= end: |
| 23 | last = cursor + timedelta(days=min(size - 1, (end - cursor).days)) |
| 24 | windows.append((cursor, last)) |
| 25 | if last == end: |
| 26 | break |
| 27 | cursor = last + DAY |
| 28 | return windows |
| 29 | |
| 30 | |
| 31 | def usable_bars(rows, key, symbol, currency, end): |
| 32 | return sorted((b for b in (rows or []) if b is not None |
| 33 | and b.global_instrument_id == key and b.provider == 'NSE' |
| 34 | and b.source_mode == 'REAL' and b.provider_symbol == symbol and b.currency == currency |
| 35 | and b.trading_date <= end and all(getattr(b, name) is not None for name in ('open', 'high', 'low', 'close'))), |
| 36 | key=lambda b: b.trading_date) |
| 37 | |
| 38 | |
| 39 | def coverage_plan(rows, start, end, now, freshness_hours, *, force=False): |
| 40 | """Conservative prefix/tail policy; short tail lag is freshness, not a holiday claim.""" |
| 41 | if not rows: |
| 42 | return 'NO_HISTORY', [(start, end)] |
| 43 | first, last = rows[0], rows[-1] |
| 44 | freshness = timedelta(hours=freshness_hours) |
| 45 | prefix = first.trading_date > start |
| 46 | stale = now - last.retrieved_at >= freshness or timedelta(days=(end - last.trading_date).days) >= freshness |
| 47 | state = 'PARTIAL_HISTORY' if prefix else 'STALE_HISTORY' if stale else 'CURRENT_HISTORY' |
| 48 | if force: |
| 49 | return state, [(start, end)] |
| 50 | intervals = [] |
| 51 | if prefix: |
| 52 | intervals.append((start, min(end, first.trading_date - DAY))) |
| 53 | if stale: |
| 54 | # Revisit the last bar only when no later date exists, to allow correction |
| 55 | # and refresh retrieval provenance; otherwise acquire an incremental tail. |
| 56 | tail = min(end, last.trading_date + DAY) if last.trading_date < end else end |
| 57 | intervals.append((max(start, tail), end)) |
| 58 | return state, intervals |
| 59 | |
| 60 | |
| 61 | def uncovered(intervals, completed): |
| 62 | """Subtract successful request ranges, not imagined exchange sessions.""" |
| 63 | remaining = list(intervals) |
| 64 | for low, high in sorted(completed): |
| 65 | next_ranges = [] |
| 66 | for start, end in remaining: |
| 67 | if high < start or low > end: |
| 68 | next_ranges.append((start, end)) |
| 69 | else: |
| 70 | if start < low: |
| 71 | next_ranges.append((start, low - DAY)) |
| 72 | if high < end: |
| 73 | next_ranges.append((high + DAY, end)) |
| 74 | remaining = next_ranges |
| 75 | return remaining |
| 76 | |
| 77 | |
| 78 | def failure_class(reason): |
| 79 | if reason == 'DAILY_BAR_PERSISTENCE_UNAVAILABLE': |
| 80 | return 'PERSISTENCE_FAILURE' |
| 81 | if '429' in reason: |
| 82 | return 'THROTTLED' |
| 83 | if 'HTTP_' in reason: |
| 84 | return 'HTTP_ERROR' |
| 85 | if reason in {'EMPTY_RESPONSE', 'NO_VALID_HISTORY'}: |
| 86 | return 'EMPTY_RESPONSE' |
| 87 | if any(s in reason for s in ('CSV', 'HEADERS', 'REJECTED_ROWS')): |
| 88 | return 'PARSER_FAILURE' |
| 89 | if any(s in reason for s in ('MAPPING', 'CURRENCY', 'UNSUPPORTED_INSTRUMENT', 'IDENTITY_MISMATCH')): |
| 90 | return 'NO_TRUSTED_MAPPING' |
| 91 | if reason.startswith('IDENTITY'): |
| 92 | return 'IDENTITY_UNAVAILABLE' |
| 93 | if 'WINDOW' in reason or 'DATE_RANGE' in reason: |
| 94 | return 'WINDOW_PLANNING_FAILURE' |
| 95 | return 'PROVIDER_UNAVAILABLE' |
| 96 | |
| 97 | |
| 98 | def _item(key): |
| 99 | return dict(globalInstrumentId=str(key), nse_symbol=None, coverage_state_before='UNKNOWN', |
| 100 | requested_windows=0, successful_windows=0, failed_windows=0, rows_received=0, |
| 101 | rows_accepted=0, rows_persisted=0, status='FAILED', failure_reason=None, |
| 102 | failure_class=None, earliest_persisted_date=None, latest_persisted_date=None, windows=[]) |
| 103 | |
| 104 | |
| 105 | async def run_backfill(jobs, *, identity_headers, correlation_id, offset, instrument_ids, start, end, force): |
| 106 | now = jobs._clock() |
| 107 | today = now.astimezone(ZoneInfo('Asia/Kolkata')).date() |
| 108 | settings = jobs.settings |
| 109 | end = today if end is None else end |
| 110 | start = today - timedelta(days=settings.market_data_population_initial_lookback_days) if start is None else start |
| 111 | summary = dict(job_started_at=now, job_finished_at=None, status='RUNNING', failure_reason=None, |
| 112 | requested_instruments=0, processed_instruments=0, succeeded_instruments=0, failed_instruments=0, |
| 113 | skipped_current=0, skipped_cooldown=0, requested_windows=0, successful_windows=0, |
| 114 | failed_windows=0, rows_received=0, rows_accepted=0, rows_persisted=0, |
| 115 | first_requested_date=None, last_requested_date=None, target_start=start, target_end=end, |
| 116 | offset=offset, next_offset=None, candidate_instruments=0, instruments=[], |
| 117 | internal_gap_repair='UNSUPPORTED_IN_THIS_PHASE') |
| 118 | provider = None |
| 119 | job_failure_stage = 'WINDOW_PLANNING_FAILURE' |
| 120 | try: |
| 121 | if (type(offset) is not int or offset < 0 or type(force) is not bool |
| 122 | or type(start) is not date or type(end) is not date or start > end or end > today |
| 123 | or start < today - timedelta(days=settings.market_data_population_initial_lookback_days)): |
| 124 | raise ValueError('WINDOW_PLANNING_FAILURE') |
| 125 | # Canonical metadata only; no portfolio/watchlist/scanner membership. |
| 126 | job_failure_stage = 'UNIVERSE_UNAVAILABLE' |
| 127 | universe = await jobs.orchestrator.active_global_equities( |
| 128 | identity_headers=identity_headers, correlation_id=correlation_id) |
| 129 | candidates = set() |
| 130 | for value in universe: |
| 131 | if not isinstance(value, dict): |
| 132 | continue |
| 133 | if (value.get('status') != 'ACTIVE' or value.get('assetType') != 'EQUITY' |
| 134 | or value.get('country') not in {'IN', 'IND', 'INDIA'} |
| 135 | or value.get('exchange') not in {'NSE', 'XNSE'}): |
| 136 | continue |
| 137 | try: |
| 138 | key = UUID(str(value.get('globalInstrumentId'))) |
| 139 | except ValueError: |
| 140 | continue |
| 141 | if instrument_ids is None or key in instrument_ids: |
| 142 | candidates.add(key) |
| 143 | ordered = sorted(candidates, key=str) |
| 144 | selected = ordered[offset:offset + settings.market_data_population_batch_size] |
| 145 | summary['candidate_instruments'] = len(ordered) |
| 146 | summary['requested_instruments'] = len(selected) |
| 147 | summary['next_offset'] = offset + len(selected) if offset + len(selected) < len(ordered) else None |
| 148 | throttled = (jobs._daily_bar_throttled_at is not None and now - jobs._daily_bar_throttled_at |
| 149 | < timedelta(hours=settings.market_data_population_retry_cooldown_hours)) |
| 150 | for key in selected: |
| 151 | item = _item(key) |
| 152 | summary['instruments'].append(item) |
| 153 | summary['processed_instruments'] += 1 |
| 154 | stage = 'IDENTITY_UNAVAILABLE' |
| 155 | try: |
| 156 | metadata = await jobs.orchestrator.global_instrument_metadata(key, |
| 157 | identity_headers=identity_headers, correlation_id=correlation_id) |
| 158 | symbol, currency = verified_identity(metadata, key) |
| 159 | item['nse_symbol'] = symbol |
| 160 | stage = 'PERSISTENCE_FAILURE' |
| 161 | existing = await jobs.repository.daily_market_bars_for_instruments({key}, end_date=end, provider='NSE') |
| 162 | rows = usable_bars(existing.get(key), key, symbol, currency, end) |
| 163 | if rows: |
| 164 | item['earliest_persisted_date'], item['latest_persisted_date'] = rows[0].trading_date, rows[-1].trading_date |
| 165 | stage = 'WINDOW_PLANNING_FAILURE' |
| 166 | state, intervals = coverage_plan(rows, start, end, now, |
| 167 | settings.market_data_historical_freshness_hours, force=force) |
| 168 | item['coverage_state_before'] = state |
| 169 | cache_key = (key, symbol, currency) |
| 170 | fresh_ranges = [(a, b, stamp) for a, b, stamp in jobs._daily_bar_completed_ranges.get(cache_key, []) |
| 171 | if now - stamp < timedelta(hours=settings.market_data_historical_freshness_hours)] |
| 172 | jobs._daily_bar_completed_ranges[cache_key] = fresh_ranges |
| 173 | if not force and rows: |
| 174 | intervals = uncovered(intervals, [(a, b) for a, b, _ in fresh_ranges]) |
| 175 | failed_at = jobs._daily_bar_failures.get(key) |
| 176 | if throttled or (failed_at is not None and now - failed_at < timedelta(hours=settings.market_data_population_retry_cooldown_hours)): |
| 177 | item.update(status='SKIPPED_COOLDOWN', failure_reason='JOB_THROTTLED' if throttled else 'RETRY_COOLDOWN') |
| 178 | continue |
| 179 | if not intervals: |
| 180 | item.update(status='SKIPPED_CURRENT') |
| 181 | continue |
| 182 | windows = [window for a, b in intervals for window in plan_windows(a, b, settings.nse_historical_request_window_days)] |
| 183 | stage = 'PROVIDER_UNAVAILABLE' |
| 184 | if provider is None: |
| 185 | provider = NseHistoricalDailyProvider(jobs.orchestrator, settings) |
| 186 | for a, b in windows: |
| 187 | item['requested_windows'] += 1 |
| 188 | detail = dict(start=a, end=b, status='FAILED', failure_reason=None, http_status=None) |
| 189 | item['windows'].append(detail) |
| 190 | summary['first_requested_date'] = min(summary['first_requested_date'] or a, a) |
| 191 | summary['last_requested_date'] = max(summary['last_requested_date'] or b, b) |
| 192 | try: |
| 193 | result = await provider.fetch(key, start=a, end=b, |
| 194 | identity_headers=identity_headers, correlation_id=correlation_id) |
| 195 | item['rows_received'] += result.rows_parsed |
| 196 | item['rows_accepted'] += result.rows_accepted |
| 197 | detail['http_status'] = result.http_status |
| 198 | result = await persist_daily_result(jobs.repository, result) |
| 199 | item['rows_persisted'] += result.persisted_rows |
| 200 | if result.persisted_rows: |
| 201 | item['earliest_persisted_date'] = min(item['earliest_persisted_date'] or result.first_trading_date, result.first_trading_date) |
| 202 | item['latest_persisted_date'] = max(item['latest_persisted_date'] or result.last_trading_date, result.last_trading_date) |
| 203 | reason = result.failure_reason |
| 204 | if result.rows_rejected and reason in {None, 'NO_VALID_HISTORY'}: |
| 205 | reason = 'REJECTED_ROWS' |
| 206 | if reason: |
| 207 | detail['failure_reason'] = reason |
| 208 | item['failed_windows'] += 1 |
| 209 | item.update(failure_reason=reason, failure_class=failure_class(reason)) |
| 210 | jobs._daily_bar_failures[key] = jobs._clock() |
| 211 | if '429' in reason: |
| 212 | throttled = True |
| 213 | jobs._daily_bar_throttled_at = jobs._clock() |
| 214 | # Empty windows may precede listing or contain closures; |
| 215 | # they are failures, not fabricated zero-bar successes. |
| 216 | if failure_class(reason) != 'EMPTY_RESPONSE': |
| 217 | break |
| 218 | else: |
| 219 | detail['status'] = 'SUCCESS' |
| 220 | item['successful_windows'] += 1 |
| 221 | jobs._daily_bar_completed_ranges[cache_key].append((a, b, jobs._clock())) |
| 222 | except Exception: |
| 223 | item['failed_windows'] += 1 |
| 224 | detail['failure_reason'] = 'PROVIDER_UNAVAILABLE' |
| 225 | item.update(failure_reason='PROVIDER_UNAVAILABLE', failure_class='PROVIDER_UNAVAILABLE') |
| 226 | jobs._daily_bar_failures[key] = jobs._clock() |
| 227 | break |
| 228 | if not item['failed_windows']: |
| 229 | item['status'] = 'SUCCESS' |
| 230 | jobs._daily_bar_failures.pop(key, None) |
| 231 | except Exception as exc: |
| 232 | safe_reasons = {'IDENTITY_MISMATCH', 'UNSUPPORTED_INSTRUMENT', 'NO_UNAMBIGUOUS_NSE_MAPPING', |
| 233 | 'NO_TRUSTED_NSE_MAPPING', 'INVALID_NSE_MAPPING', 'MISSING_OR_AMBIGUOUS_CURRENCY', 'WINDOW_PLANNING_FAILURE'} |
| 234 | reason = str(exc) if type(exc) is ValueError and str(exc) in safe_reasons else stage |
| 235 | item.update(failure_reason=reason, failure_class=failure_class(reason) if stage == 'IDENTITY_UNAVAILABLE' else stage) |
| 236 | jobs._daily_bar_failures[key] = jobs._clock() |
| 237 | finally: |
| 238 | for metric in ('requested_windows', 'successful_windows', 'failed_windows', 'rows_received', 'rows_accepted', 'rows_persisted'): |
| 239 | summary[metric] += item[metric] |
| 240 | counter = {'SUCCESS':'succeeded_instruments', 'FAILED':'failed_instruments', |
| 241 | 'SKIPPED_CURRENT':'skipped_current', 'SKIPPED_COOLDOWN':'skipped_cooldown'}[item['status']] |
| 242 | summary[counter] += 1 |
| 243 | summary['status'] = 'COMPLETED_WITH_ERRORS' if summary['failed_instruments'] else 'COMPLETED' |
| 244 | except Exception: |
| 245 | summary['status'] = 'FAILED' |
| 246 | summary['failure_reason'] = job_failure_stage |
| 247 | finally: |
| 248 | if provider is not None: |
| 249 | try: |
| 250 | await provider.aclose() |
| 251 | except Exception: |
| 252 | summary.update(status='FAILED', failure_reason='SESSION_CLOSE_FAILURE') |
| 253 | finally: |
| 254 | await jobs._sleep(settings.market_data_population_request_interval_seconds) |
| 255 | summary['job_finished_at'] = jobs._clock() |
| 256 | return summary |