@prakhar-gitlab / ai-investment-platform / commits / abeb9b9

feat: add bounded NSE daily bar backfill

prakhar82 committed Sep 13, 2026 at 21:19 UTC abeb9b9a48db5878413f2937df4aedc885bec610
4 files changed +757
ai/research-engine/NSE_DAILY_BACKFILL.md new
+123
@@ -0,0 +1,123 @@
1 +# Bounded NSE daily OHLCV backfill
2 +
3 +## Entry point and universe
4 +
5 +`IndiaMarketDataPopulationJobs.backfill_daily_bars(identity_headers=...,
6 +correlation_id=None, offset=0, instrument_ids=None, start=None, end=None,
7 +force=False)` is an explicit async worker operation. It shares the existing
8 +daily-bar lock with single-instrument acquisition. No GET, scanner, submit,
9 +ensure, or scheduler automatically invokes it. Yahoo close-only behavior stays
10 +unchanged. This extends the population boundary, not a second job framework.
11 +
12 +The source is `active_global_equities()` (paginated canonical ACTIVE EQUITY
13 +metadata). India/NSE candidates are deduplicated and sorted by UUID string;
14 +at most `market_data_population_batch_size` (default 50) are selected.
15 +Each selected candidate then resolves current detail metadata and passes the
16 +committed `verified_identity` gate before NSE acquisition. Invalid mappings
17 +consume a bounded batch slot and appear as failures; they are never fetched.
18 +Listing metadata uses `exchange`; detail metadata uses `primaryExchange`, as
19 +specified by the existing canonical API. No holdings/watchlists/sector filter.
20 +
21 +Optional IDs restrict canonical membership; IDs absent from the universe are
22 +not acquired. `next_offset` is null at completion, otherwise the next offset in
23 +the sorted candidate universe or explicit-ID intersection. Selection is stable
24 +for the same snapshot. Universe membership changes can shift offsets; no durable
25 +snapshot cursor is claimed. Only identity metadata is enumerated globally;
26 +history is read one instrument at a time and persisted per window.
27 +
28 +## Date and coverage policy
29 +
30 +Default interval: Asia/Kolkata current DATE minus existing initial lookback
31 +(default 400 days), through current DATE inclusive. Optional bounds may narrow
32 +but not extend that horizon or request future dates. Retrieval timestamps remain
33 +UTC. `plan_windows` produces consecutive non-overlapping oldest-first windows,
34 +using the configured provider limit (default 30 inclusive calendar days). A
35 +401-calendar-date interval uses 14 windows. Thirty days is an operational bound,
36 +not a claimed official NSE maximum.
37 +
38 +Reads reuse `daily_market_bars_for_instruments({id}, end_date=end, provider='NSE')`.
39 +Only the canonical ID, current symbol/currency, REAL NSE rows with non-null OHLC
40 +satisfy coverage. Optional volume/turnover are not required.
41 +
42 +- NO_HISTORY: fetch the target interval.
43 +- PARTIAL_HISTORY: fetch the missing prefix before the earliest usable date;
44 + also fetch a stale tail when needed.
45 +- STALE_HISTORY: retrieval age or latest-date lag exceeds configured historical
46 + freshness (72 hours). Fetch latest date + 1 through target end. If latest date
47 + already equals target end, refresh that final date for corrections/freshness.
48 +- CURRENT_HISTORY: observed prefix is covered and tail sufficiently fresh; skip.
49 + This does not assert that every internal exchange session is present.
50 +- GAP_DETECTED is deliberately not inferred without authoritative sessions.
51 +
52 +INTERNAL_GAP_REPAIR = UNSUPPORTED_IN_THIS_PHASE. Weekly schedules and optional
53 +calendar exceptions do not establish a complete historical NSE holiday calendar.
54 +No weekday/holiday is synthesized or labelled missing market data.
55 +
56 +Successful request ranges are remembered for the existing freshness interval
57 +within the worker, keyed by canonical ID/symbol/currency. This suppresses repeated
58 +boundary probes when a completed range has no bar on its exact first date.
59 +Memory is used only alongside existing usable persisted rows. Missing rows still
60 +cause NO_HISTORY acquisition. Empty/failed requests are never marked complete.
61 +State is process-local, like existing jobs; after restart boundary probes can
62 +recur. Listing-date-aware prefix suppression and durable empty-range evidence
63 +are not implemented.
64 +
65 +Force defaults false. Explicit force reacquires the requested bounded interval
66 +for idempotent correction, but does not bypass cooldown or throttling.
67 +
68 +## Execution and results
69 +
70 +One lazily created NseHistoricalDailyProvider/session per job, closed in finally
71 +including cancellation. Its existing cookie/header, spacing and bounded retry
72 +behavior is reused. The population lock and inter-job spacing serialize this
73 +operation with single-instrument acquisition. No global HTTP session, proxy
74 +logic or concurrent NSE requests. Coordination is process-local, not distributed.
75 +
76 +Each successful window uses `persist_daily_result` and existing
77 +`ResearchRepository.upsert_daily_market_bars_async`. No SQL/schema change and no
78 +close-only dual-write. Earlier good rows survive later failures.
79 +
80 +Failures distinguish identity lookup, mapping, provider, throttling, HTTP, parser,
81 +empty response, persistence and planning. Most failures stop the current
82 +instrument's remaining windows, then continue with the next instrument. Empty
83 +windows remain failures but allow later windows (for example, after listing).
84 +Partial parser rejections persist accepted rows but count as failed windows and
85 +never mark request coverage complete. A 429 stops further NSE calls for the job
86 +and sets the existing 12-hour cooldown for subsequent backfill calls. Other
87 +failures use per-instrument cooldown. Force cannot override either cooldown.
88 +
89 +Requested-window counts are actual attempts, excluding unattempted windows
90 +after failures. Processed instruments include failures and skips, and equal
91 +succeeded + failed + skipped_current + skipped_cooldown. Per-instrument results
92 +include requested DATE windows, coverage state, status/failure class/reason,
93 +HTTP status, row counts and observed persisted bounds. Persisted row counts are
94 +upserts, not necessarily newly inserted keys. No cookies, authentication headers
95 +or raw exception text appear in summaries.
96 +
97 +## Runtime smoke, 2026-09-13
98 +
99 +A read-only canonical PostgreSQL snapshot contained 2,568 ACTIVE equities. The
100 +existing trusted gate accepted all six examples: NILKAMAL, RAYMOND, GODREJAGRO,
101 +ADANIPORTS, TDPOWERSYS, GEEKAYWIRE, plus POLYCAB. Names are validation examples,
102 +not implementation selection logic.
103 +
104 +A runtime adapter exposed this canonical snapshot through existing universe and
105 +metadata method shapes. The generic worker was restricted to NILKAMAL
106 +`4b085a61-0864-4ef1-ae23-0ba7e3ec6afb` and POLYCAB
107 +`f8cb0fc7-082c-4d95-a77d-b1a9ca21d5a4`, September 1–4, 2026.
108 +
109 +First run: two NO_HISTORY instruments, two successful windows, eight rows
110 +received/accepted/persisted. Three NSE HTTP calls total: one bootstrap and two
111 +history requests, all HTTP 200. Retries were disabled for this smoke. Current
112 +ResearchRepository/SQLite persistence read back eight unique provider/day rows.
113 +Immediate repeat: two CURRENT_HISTORY skips, zero windows, zero NSE calls,
114 +identical rows. Runtime artifacts are under ignored `.tmp/`.
115 +
116 +This validates live NSE and current local orchestration/persistence, not
117 +authenticated canonical HTTP enumeration or deployed PostgreSQL application
118 +end-to-end behavior. Mocked provider/HTTP and real SQLite repository tests cover
119 +multi-window failure isolation, continuation, corrections, and cooldowns.
120 +
121 +Out of scope: deployed PostgreSQL validation, ATR/ADX and volume technical
122 +wiring, internal-gap repair without an authoritative calendar, sector benchmark
123 +mapping/history, and broad-market benchmark history.
ai/research-engine/app/market_data_population.py
+21
@@ -57,6 +57,27 @@ class IndiaMarketDataPopulationJobs:
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_daily_bars(
83 self, global_instrument_id: UUID, *, start: date, end: date,
ai/research-engine/app/nse_daily_backfill.py new
+256
@@ -0,0 +1,256 @@
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
ai/research-engine/tests/test_nse_daily_backfill.py new
+357
@@ -0,0 +1,357 @@
1 +from datetime import date, datetime, timedelta, timezone
2 +from decimal import Decimal
3 +from types import SimpleNamespace
4 +from unittest.mock import AsyncMock
5 +from uuid import UUID
6 +
7 +import httpx
8 +import pytest
9 +
10 +import app.nse_daily_backfill as backfill
11 +from app.market_data_population import IndiaMarketDataPopulationJobs
12 +from app.models import DailyMarketBar
13 +from app.nse_historical_daily import NseHistoricalDailyProvider, NseHistoricalResult, BOOTSTRAP
14 +from app.persistence import SqliteResearchPersistence
15 +from app.repository import ResearchRepository
16 +from app.settings import Settings
17 +
18 +NOW = datetime(2026, 9, 13, tzinfo=timezone.utc)
19 +START, END = date(2026, 9, 1), date(2026, 9, 4)
20 +
21 +
22 +def metadata(n=1, **changes):
23 + return dict(globalInstrumentId=str(UUID(int=n)), status='ACTIVE', assetType='EQUITY', country='IN',
24 + primaryExchange='NSE', currency='INR', providerMappings=[dict(provider='NSE', status='VERIFIED',
25 + providerSymbol=f'S{n}', currency='INR', resolutionSource='OFFICIAL_NSE')]) | changes
26 +
27 +
28 +def bar(n=1, day=START, **changes):
29 + return DailyMarketBar(global_instrument_id=UUID(int=n), trading_date=day, open=Decimal('10'), high=Decimal('12'),
30 + low=Decimal('9'), close=Decimal('11'), volume=0, currency='INR', provider='NSE', provider_symbol=f'S{n}',
31 + source_mode='REAL', source_url='https://www.nseindia.com/history', retrieved_at=NOW).model_copy(update=changes)
32 +
33 +
34 +class Provider:
35 + def __init__(self):
36 + self.calls = []
37 + self.closed = False
38 + self.failures = {}
39 + self.price = Decimal('11')
40 +
41 + async def fetch(self, key, *, start, end, **kwargs):
42 + self.calls.append((key, start, end))
43 + reason = self.failures.get((key.int, start), self.failures.get(key.int))
44 + if isinstance(reason, Exception):
45 + raise reason
46 + values = [bar(key.int, day=start, close=self.price)]
47 + if start != end:
48 + values.append(bar(key.int, day=end, close=self.price))
49 + return NseHistoricalResult(key, start, end, provider_symbol=f'S{key.int}', http_status=200,
50 + status='UNAVAILABLE' if reason else 'SUCCESS', failure_reason=reason,
51 + rows_parsed=0 if reason else len(values), bars=[] if reason else values)
52 +
53 + async def aclose(self): self.closed = True
54 +
55 +
56 +def setup(monkeypatch, values=None, **settings):
57 + values = [metadata()] if values is None else values
58 + universe = [v | {'exchange': v.get('primaryExchange')} for v in values]
59 + by_id = {UUID(v['globalInstrumentId']): v for v in values}
60 + orchestrator = SimpleNamespace(active_global_equities=AsyncMock(return_value=universe),
61 + global_instrument_metadata=AsyncMock(side_effect=lambda key, **kw: by_id[key]))
62 + store = SqliteResearchPersistence()
63 + config = Settings(**settings)
64 + repo = ResearchRepository(settings=config, persistence=store)
65 + jobs = IndiaMarketDataPopulationJobs(repo, None, orchestrator, config, clock=lambda: NOW, sleep=AsyncMock())
66 + providers = []
67 + def factory(*args):
68 + p = Provider(); providers.append(p); return p
69 + monkeypatch.setattr(backfill, 'NseHistoricalDailyProvider', factory)
70 + return jobs, store, providers
71 +
72 +
73 +async def run(jobs, **kwargs):
74 + return await jobs.backfill_daily_bars(identity_headers={}, start=START, end=END, **kwargs)
75 +
76 +
77 +@pytest.mark.parametrize('start,end,size,count', [
78 + (date(2026,1,1),date(2026,1,30),30,1), (date(2026,1,1),date(2026,1,31),30,2),
79 + (date(2025,12,15),date(2026,3,5),30,3), (date(2024,2,1),date(2024,3,1),30,1),
80 + (START,START,30,1), (date(2026,1,31),date(2026,2,1),1,2)])
81 +def test_windows(start,end,size,count):
82 + windows=backfill.plan_windows(start,end,size)
83 + assert len(windows)==count and windows[0][0]==start and windows[-1][1]==end
84 + assert windows==backfill.plan_windows(start,end,size)
85 + assert all(0 <= (b-a).days < size for a,b in windows)
86 + assert all(windows[i][1]+timedelta(days=1)==windows[i+1][0] for i in range(len(windows)-1))
87 +
88 +
89 +@pytest.mark.parametrize('start,end,size',[(END,START,30),(START,END,0),(START,END,-1),(NOW,END,30)])
90 +def test_invalid_windows(start,end,size):
91 + with pytest.raises(ValueError): backfill.plan_windows(start,end,size)
92 +
93 +
94 +def test_coverage_states_and_no_calendar_inference():
95 + assert backfill.coverage_plan([],START,END,NOW,72)[0]=='NO_HISTORY'
96 + assert backfill.coverage_plan([bar(day=END)],START,END,NOW,72)==('PARTIAL_HISTORY',[(START,END-timedelta(days=1))])
97 + assert backfill.coverage_plan([bar(),bar(day=END)],START,END,NOW,72)==('CURRENT_HISTORY',[])
98 + old=NOW-timedelta(days=5)
99 + assert backfill.coverage_plan([bar(retrieved_at=old)],START,END,NOW,72)==('STALE_HISTORY',[(START+timedelta(days=1),END)])
100 + # Sparse internal dates are not enough to claim a session gap.
101 + assert backfill.coverage_plan([bar(),bar(day=END)],START,END,NOW,72)[0]!='GAP_DETECTED'
102 + assert backfill.coverage_plan([bar(),bar(day=END)],START,END,NOW,72,force=True)[1]==[(START,END)]
103 +
104 +
105 +@pytest.mark.asyncio
106 +async def test_current_no_session_and_other_provider_does_not_count(monkeypatch):
107 + jobs,store,providers=setup(monkeypatch)
108 + store.upsert_daily_market_bars([bar(provider='OTHER'),bar(day=END,provider='OTHER')])
109 + first=await run(jobs)
110 + assert first['instruments'][0]['coverage_state_before']=='NO_HISTORY' and first['rows_persisted']==2
111 + assert len(store.load_daily_market_bars({UUID(int=1)}))==4
112 + second=await run(jobs)
113 + assert second['skipped_current']==1 and second['requested_windows']==0 and len(providers)==1
114 + assert providers[0].closed
115 +
116 +
117 +@pytest.mark.asyncio
118 +async def test_prefix_tail_and_null_rows(monkeypatch):
119 + jobs,store,providers=setup(monkeypatch)
120 + store.upsert_daily_market_bar(bar(day=date(2026,9,3),retrieved_at=NOW-timedelta(days=5)))
121 + r=await run(jobs)
122 + assert [(a,b) for _,a,b in providers[0].calls]==[(START,date(2026,9,2)),(END,END)]
123 + assert r['instruments'][0]['coverage_state_before']=='PARTIAL_HISTORY'
124 + assert backfill.usable_bars([None,bar(open=None)],UUID(int=1),'S1','INR',END)==[]
125 +
126 +
127 +@pytest.mark.asyncio
128 +async def test_batch_order_offset_and_membership(monkeypatch):
129 + jobs,store,providers=setup(monkeypatch,[metadata(3),metadata(1),metadata(2)],market_data_population_batch_size=1)
130 + first=await run(jobs)
131 + assert first['requested_instruments']==1 and first['next_offset']==1
132 + second=await run(jobs,offset=1)
133 + assert second['instruments'][0]['globalInstrumentId']==str(UUID(int=2)) and second['next_offset']==2
134 + third=await run(jobs,instrument_ids={UUID(int=3),UUID(int=999)})
135 + assert third['requested_instruments']==1 and third['instruments'][0]['globalInstrumentId']==str(UUID(int=3))
136 + assert third['next_offset'] is None
137 +
138 +
139 +@pytest.mark.asyncio
140 +@pytest.mark.parametrize('values',[[],[metadata(assetType='ETF')],[metadata(status='INACTIVE')],[metadata(country='US')],[metadata(primaryExchange='BSE')]])
141 +async def test_empty_eligible_universe(monkeypatch,values):
142 + jobs,_,providers=setup(monkeypatch,values)
143 + r=await run(jobs)
144 + assert r['requested_instruments']==0 and not providers and r['status']=='COMPLETED'
145 +
146 +
147 +@pytest.mark.asyncio
148 +@pytest.mark.parametrize('mapping',[[],[dict(provider='NSE',status='INVALID',providerSymbol='X')],
149 + [dict(provider='NSE',status='VERIFIED',providerSymbol=' ')], metadata()['providerMappings']*2,
150 + [metadata()['providerMappings'][0]|dict(active=False)], [metadata()['providerMappings'][0]|dict(currency='USD')]])
151 +async def test_identity_failure_isolated(monkeypatch,mapping):
152 + jobs,_,providers=setup(monkeypatch,[metadata(providerMappings=mapping),metadata(2)])
153 + r=await run(jobs)
154 + assert r['failed_instruments']==1 and r['succeeded_instruments']==1
155 + assert r['instruments'][0]['failure_class']=='NO_TRUSTED_MAPPING'
156 + assert [key.int for key,_,_ in providers[0].calls]==[2]
157 +
158 +
159 +@pytest.mark.asyncio
160 +@pytest.mark.parametrize('reason,category',[('HISTORICAL_HTTP_403','HTTP_ERROR'),('NON_CSV_RESPONSE','PARSER_FAILURE'),
161 + ('EMPTY_RESPONSE','EMPTY_RESPONSE'),('HISTORICAL_TIMEOUT','PROVIDER_UNAVAILABLE'),(RuntimeError('secret-cookie'),'PROVIDER_UNAVAILABLE')])
162 +async def test_provider_failure_isolation_and_cooldown(monkeypatch,reason,category):
163 + jobs,store,_=setup(monkeypatch,[metadata(),metadata(2)])
164 + provider=Provider(); provider.failures[1]=reason
165 + monkeypatch.setattr(backfill,'NseHistoricalDailyProvider',lambda *args:provider)
166 + r=await run(jobs)
167 + assert r['failed_instruments']==1 and r['succeeded_instruments']==1 and r['failed_windows']==1
168 + assert r['instruments'][0]['failure_class']==category and provider.closed
169 + assert 'secret-cookie' not in str(r)
170 + repeat=await run(jobs)
171 + assert repeat['skipped_cooldown']==1 and repeat['skipped_current']==1
172 +
173 +
174 +@pytest.mark.asyncio
175 +async def test_throttle_stops_job_and_next_job(monkeypatch):
176 + jobs,_,_=setup(monkeypatch,[metadata(),metadata(2)])
177 + provider=Provider(); provider.failures[1]='HISTORICAL_HTTP_429'
178 + monkeypatch.setattr(backfill,'NseHistoricalDailyProvider',lambda *args:provider)
179 + r=await run(jobs)
180 + assert r['failed_instruments']==1 and r['skipped_cooldown']==1 and len(provider.calls)==1
181 + r=await run(jobs,force=True)
182 + assert r['skipped_cooldown']==2 and len(provider.calls)==1
183 +
184 +
185 +@pytest.mark.asyncio
186 +async def test_per_window_persistence_later_failure_preserves_rows(monkeypatch):
187 + jobs,store,_=setup(monkeypatch,[metadata(),metadata(2)],nse_historical_request_window_days=2)
188 + provider=Provider(); provider.failures[(1,date(2026,9,3))]='HISTORICAL_HTTP_500'
189 + monkeypatch.setattr(backfill,'NseHistoricalDailyProvider',lambda *args:provider)
190 + r=await run(jobs)
191 + assert r['requested_windows']==4 and r['successful_windows']==3 and r['failed_windows']==1
192 + assert r['rows_received']==r['rows_accepted']==r['rows_persisted']==6
193 + assert len(store.load_daily_market_bars({UUID(int=1)},provider='NSE'))==2
194 + assert r['instruments'][0]['earliest_persisted_date']==START
195 + assert r['first_requested_date']==START and r['last_requested_date']==END
196 +
197 +
198 +@pytest.mark.asyncio
199 +async def test_persistence_failure_next_instrument(monkeypatch):
200 + jobs,store,_=setup(monkeypatch,[metadata(),metadata(2)])
201 + original=jobs.repository.upsert_daily_market_bars_async
202 + async def writer(bars):
203 + if bars[0].global_instrument_id.int==1: raise RuntimeError('secret')
204 + return await original(bars)
205 + jobs.repository.upsert_daily_market_bars_async=writer
206 + r=await run(jobs)
207 + assert r['instruments'][0]['failure_class']=='PERSISTENCE_FAILURE' and r['succeeded_instruments']==1
208 + assert 'secret' not in str(r)
209 +
210 +
211 +@pytest.mark.asyncio
212 +async def test_force_correction_idempotence(monkeypatch):
213 + jobs,store,_=setup(monkeypatch)
214 + provider=Provider(); monkeypatch.setattr(backfill,'NseHistoricalDailyProvider',lambda *args:provider)
215 + await run(jobs)
216 + provider.price=Decimal('10.5')
217 + r=await run(jobs,force=True)
218 + assert r['rows_persisted']==2
219 + rows=store.load_daily_market_bars({UUID(int=1)},provider='NSE')
220 + assert len(rows)==2 and all(b.close==Decimal('10.5') for b in rows)
221 +
222 +
223 +@pytest.mark.asyncio
224 +async def test_default_lookback_exchange_date_and_invalid_range(monkeypatch):
225 + jobs,_,providers=setup(monkeypatch)
226 + jobs._clock=lambda:datetime(2026,9,12,20,tzinfo=timezone.utc)
227 + r=await jobs.backfill_daily_bars(identity_headers={})
228 + assert r['target_end']==date(2026,9,13) and r['target_start']==date(2026,9,13)-timedelta(days=400)
229 + assert r['requested_windows']==14 and providers[0].closed
230 + bad=await jobs.backfill_daily_bars(identity_headers={},start=END,end=START)
231 + assert bad['failure_reason']=='WINDOW_PLANNING_FAILURE'
232 +
233 +
234 +@pytest.mark.asyncio
235 +async def test_real_provider_one_session_verified_query_and_serialization(monkeypatch):
236 + jobs,_,_=setup(monkeypatch,[metadata(),metadata(2)],nse_historical_request_window_days=2)
237 + calls=[]
238 + def handler(req):
239 + calls.append(req)
240 + if str(req.url)==BOOTSTRAP: return httpx.Response(200,headers={'set-cookie':'session=private; Path=/'})
241 + assert req.headers['cookie']=='session=private'
242 + symbol=req.url.params['symbol']; day=req.url.params['from']
243 + return httpx.Response(200,text=f'Date,Symbol,Series,Open,High,Low,Close\n{day},{symbol},EQ,10,12,9,11\n')
244 + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
245 + sleep=AsyncMock()
246 + provider=NseHistoricalDailyProvider(jobs.orchestrator,jobs.settings,client=client,sleep=sleep)
247 + provider.aclose=AsyncMock()
248 + monkeypatch.setattr(backfill,'NseHistoricalDailyProvider',lambda *args:provider)
249 + r=await run(jobs)
250 + assert len(calls)==5 and sum(str(c.url)==BOOTSTRAP for c in calls)==1
251 + assert [c.url.params['symbol'] for c in calls[1:]]==['S1','S1','S2','S2']
252 + assert sleep.await_count>=4 and r['successful_windows']==4
253 + provider.aclose.assert_awaited_once()
254 + assert 'private' not in str(r)
255 +
256 +@pytest.mark.asyncio
257 +async def test_successful_range_evidence_avoids_boundary_refetch(monkeypatch):
258 + jobs,store,_=setup(monkeypatch)
259 + p=Provider()
260 + async def fetch(key, *, start, end, **kw):
261 + p.calls.append((key,start,end))
262 + return NseHistoricalResult(key,start,end,status='SUCCESS',bars=[bar(day=date(2026,9,2))],rows_parsed=1)
263 + p.fetch=fetch
264 + monkeypatch.setattr(backfill,'NseHistoricalDailyProvider',lambda *args:p)
265 + await run(jobs)
266 + r=await run(jobs)
267 + assert r['skipped_current']==1 and len(p.calls)==1
268 + jobs._clock=lambda: NOW+timedelta(hours=73)
269 + r=await run(jobs)
270 + assert r['requested_windows']>0 and len(p.calls)>1
271 +
272 +
273 +@pytest.mark.asyncio
274 +async def test_cooldown_expiry_and_identity_unavailable(monkeypatch):
275 + jobs,_,_=setup(monkeypatch)
276 + original=jobs.orchestrator.global_instrument_metadata.side_effect
277 + jobs.orchestrator.global_instrument_metadata.side_effect=ValueError('Cookie: SECRET')
278 + r=await run(jobs)
279 + assert r['instruments'][0]['failure_class']=='IDENTITY_UNAVAILABLE'
280 + assert 'SECRET' not in str(r)
281 + jobs.orchestrator.global_instrument_metadata.side_effect=original
282 + assert (await run(jobs))['skipped_cooldown']==1
283 + jobs._clock=lambda: NOW+timedelta(hours=13)
284 + assert (await run(jobs))['succeeded_instruments']==1
285 +
286 +
287 +@pytest.mark.asyncio
288 +async def test_session_closes_on_cancel(monkeypatch):
289 + import asyncio
290 + jobs,_,_=setup(monkeypatch)
291 + p=Provider()
292 + p.fetch=AsyncMock(side_effect=asyncio.CancelledError())
293 + monkeypatch.setattr(backfill,'NseHistoricalDailyProvider',lambda *args:p)
294 + with pytest.raises(asyncio.CancelledError): await run(jobs)
295 + assert p.closed
296 +
297 +
298 +@pytest.mark.asyncio
299 +async def test_population_lock_serializes_backfill_jobs(monkeypatch):
300 + import asyncio
301 + jobs,_,providers=setup(monkeypatch)
302 + a,b=await asyncio.gather(run(jobs),run(jobs))
303 + assert a['succeeded_instruments']==1 and b['skipped_current']==1 and len(providers)==1
304 +
305 +
306 +@pytest.mark.asyncio
307 +async def test_window_planning_failure_and_universe_failure(monkeypatch):
308 + jobs,_,providers=setup(monkeypatch)
309 + r=await jobs.backfill_daily_bars(identity_headers={},offset=-1)
310 + assert r['failure_reason']=='WINDOW_PLANNING_FAILURE' and not providers
311 + jobs.orchestrator.active_global_equities.side_effect=RuntimeError('secret')
312 + r=await run(jobs)
313 + assert r['failure_reason']=='UNIVERSE_UNAVAILABLE' and 'secret' not in str(r)
314 +
315 +
316 +@pytest.mark.asyncio
317 +async def test_empty_window_does_not_block_later_history(monkeypatch):
318 + jobs,store,_=setup(monkeypatch,nse_historical_request_window_days=2)
319 + p=Provider(); p.failures[(1,START)]='NO_VALID_HISTORY'
320 + monkeypatch.setattr(backfill,'NseHistoricalDailyProvider',lambda *args:p)
321 + r=await run(jobs)
322 + assert r['requested_windows']==2 and r['failed_windows']==1 and r['successful_windows']==1
323 + assert r['failed_instruments']==1 and r['rows_persisted']==2
324 + assert len(store.load_daily_market_bars({UUID(int=1)},provider='NSE'))==2
325 +
326 +
327 +@pytest.mark.asyncio
328 +async def test_partial_parser_rows_persist_but_not_marked_complete(monkeypatch):
329 + jobs,store,_=setup(monkeypatch)
330 + p=Provider()
331 + async def fetch(key,*,start,end,**kw):
332 + return NseHistoricalResult(key,start,end,status='SUCCESS',rows_parsed=2,rows_rejected=1,bars=[bar()])
333 + p.fetch=fetch; monkeypatch.setattr(backfill,'NseHistoricalDailyProvider',lambda *args:p)
334 + r=await run(jobs)
335 + assert r['failed_windows']==1 and r['rows_persisted']==1 and r['rows_received']==2 and r['rows_accepted']==1
336 + assert r['instruments'][0]['failure_class']=='PARSER_FAILURE'
337 + assert jobs._daily_bar_completed_ranges[(UUID(int=1),'S1','INR')]==[]
338 +
339 +@pytest.mark.asyncio
340 +async def test_missing_persisted_rows_override_request_memory(monkeypatch):
341 + jobs,store,providers=setup(monkeypatch)
342 + await run(jobs)
343 + jobs.repository.daily_market_bars_for_instruments=AsyncMock(return_value={UUID(int=1):None})
344 + again=await run(jobs)
345 + assert again['instruments'][0]['coverage_state_before']=='NO_HISTORY'
346 + assert again['requested_windows']==1 and len(providers)==2
347 +
348 +
349 +@pytest.mark.asyncio
350 +async def test_rejected_rows_are_parser_failure_not_empty(monkeypatch):
351 + jobs,_,_=setup(monkeypatch,nse_historical_request_window_days=2)
352 + p=Provider()
353 + p.fetch=AsyncMock(return_value=NseHistoricalResult(UUID(int=1),START,END,
354 + failure_reason='NO_VALID_HISTORY', rows_parsed=1, rows_rejected=1))
355 + monkeypatch.setattr(backfill,'NseHistoricalDailyProvider',lambda *args:p)
356 + result=await run(jobs)
357 + assert result['requested_windows']==1 and result['instruments'][0]['failure_class']=='PARSER_FAILURE'