feat: add NSE historical daily bar ingestion
prakhar82 committed
Sep 13, 2026 at 20:22 UTC
77ee4a76b5e0d22295d90f557f1226bc70efe0ca
6 files changed
+588
-2
ai/research-engine/NSE_HISTORICAL_DAILY.md
new
+76
@@ -0,0 +1,76 @@
1
+# NSE historical daily acquisition
2
+
3
+Integration: `IndiaMarketDataPopulationJobs.populate_daily_bars(global_instrument_id,
4
+start=date, end=date, identity_headers=...)` is an explicit, single-instrument
5
+worker entry point. It resolves current portfolio-service metadata, applies the
6
+existing trusted provider-mapping gate plus active equity/ambiguity/currency
7
+checks, and calls `ResearchRepository.upsert_daily_market_bars_async`.
8
+It is deliberately not called by submit, ensure, scanners, or GET handlers.
9
+The existing `HistoricalPriceProvider.closes` protocol and Yahoo priority remain
10
+close-only. No dual-write occurs.
11
+
12
+## Observed contract
13
+
14
+`tests/fixtures/nse_historical_daily.csv` is the unmodified successful official
15
+NSE response captured on 2026-09-13, after a normal cookie-bearing bootstrap at
16
+https://www.nseindia.com/report-detail/eq_security (HTTP 200).
17
+
18
+The canonical store was checked at runtime: globalInstrumentId
19
+`f8cb0fc7-082c-4d95-a77d-b1a9ca21d5a4`, POLYCAB, ACTIVE EQUITY, NSE, IN, INR;
20
+NSE mapping VERIFIED, resolution source OFFICIAL_NSE_NIFTY500.
21
+
22
+One historical request, one window, no retries:
23
+https://www.nseindia.com/api/historicalOR/generateSecurityWiseHistoricalData?from=01-09-2026&to=04-09-2026&symbol=POLYCAB&type=priceVolumeDeliverable&series=EQ&csv=true
24
+
25
+HTTP 200, text/csv, 910 bytes, four rows; both requested boundaries were present.
26
+Observed headers (surrounding whitespace omitted): Symbol, Series, Date,
27
+Prev Close, Open Price, High Price, Low Price, Last Price, Close Price,
28
+Average Price, Total Traded Quantity, Turnover ₹, No. of Trades,
29
+Deliverable Qty, % Dly Qt to Traded Qty.
30
+
31
+Required: Date, Symbol, Series, Open/High/Low/Close (the explicit ` Price`
32
+aliases are supported). Optional: Prev Close, Total Traded Quantity, Turnover ₹.
33
+Headers are trimmed, case-folded, and whitespace-collapsed; duplicate normalized
34
+headers fail the response. Symbols and series are trimmed and uppercased only.
35
+Dates accept DD-MM-YYYY and observed DD-Mon-YYYY with an explicit English month
36
+map, without locale or UTC conversion. Numeric grouping accepts Western and
37
+Indian comma grouping; prices never pass through float. Missing optional cells
38
+(blank or `-`) remain null. Duplicate valid dates reject all contenders.
39
+
40
+TURNOVER_UNIT: VERIFIED for the observed `Turnover ₹` header (INR).
41
+TURNOVER_CONVERSION: remove validated grouping commas, parse Decimal directly;
42
+multiplier 1, and only with trusted INR currency metadata. For example,
43
+`13,02,54,81,530.00` becomes Decimal(`13025481530.00`). Unobserved turnover
44
+headers, including `Turnover (in Lacs)`, remain unavailable/null; no lakh
45
+conversion is implemented or claimed verified.
46
+
47
+## Operational policy
48
+
49
+`nse_historical_request_window_days` defaults to 30 inclusive calendar days.
50
+This is an operational bound, not an NSE guaranteed maximum. Larger requests
51
+fail before metadata/network calls; this slice does not split or backfill them.
52
+`nse_historical_max_retries` defaults to 2 (validated range 0–3).
53
+The existing `market_data_population_request_interval_seconds` governs all
54
+bootstrap/history attempts; ordinary httpx cookie storage and supported content
55
+decoders are used. One provider serializes its requests; the population entry
56
+point serializes calls and applies spacing between sessions. No distributed
57
+rate limiter is claimed; use the existing single worker deployment convention.
58
+
59
+403/404 and other non-429 4xx fail without retry. 429 and 5xx, timeouts and network
60
+errors have bounded exponential backoff (1s, 2s, capped at 8s). Retry-After is
61
+honored; when it exceeds 30 seconds this call fails without retrying early.
62
+Cookies and request headers are never logged by this module. Results expose
63
+HTTP status, identity, bounds, provenance, retrieval time, counts, date coverage,
64
+row rejection reasons, and explicit acquisition/persistence failure reasons.
65
+Failures never delete previously persisted evidence.
66
+
67
+## Runtime persistence validation
68
+
69
+The same captured live response (no second history request) was parsed and four
70
+bars were persisted/read back through the current ResearchRepository and
71
+SqliteResearchPersistence at `.tmp/nse-runtime.sqlite`. DATE, OHLC range, integer
72
+volume, provider, symbol, source URL and exact model roundtrip were checked.
73
+The deployed research-engine does not yet contain DailyMarketBar, so deployed
74
+PostgreSQL end-to-end validation was not possible without a separate deployment.
75
+No deployment or migration was performed. The local runtime database is ignored
76
+and is not part of the change.
ai/research-engine/app/market_data_population.py
+23
-1
@@ -8,7 +8,7 @@ from __future__ import annotations
8
9
import asyncio
10
import logging
11
-from datetime import datetime, timedelta, timezone
11
+from datetime import date, datetime, timedelta, timezone
12
from typing import Any, Callable
13
from uuid import UUID, uuid4
14
@@ -56,6 +56,28 @@ class IndiaMarketDataPopulationJobs:
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
+
61
+ async def populate_daily_bars(
62
+ self, global_instrument_id: UUID, *, start: date, end: date,
63
+ identity_headers: dict[str, str | None], correlation_id: str | None = None,
64
+ ):
65
+ """Explicit one-instrument acquisition; never invoked by submit/ensure.
66
+
67
+ Reuse canonical metadata and the daily persistence boundary without
68
+ changing historical-close provider priority or dual-writing observations.
69
+ """
70
+ from app.nse_historical_daily import NseHistoricalDailyProvider, persist_daily_result
71
+
72
+ async with self._daily_bar_lock:
73
+ provider = NseHistoricalDailyProvider(self.orchestrator, self.settings)
74
+ try:
75
+ result = await provider.fetch(global_instrument_id, start=start, end=end,
76
+ identity_headers=identity_headers, correlation_id=correlation_id)
77
+ return await persist_daily_result(self.repository, result)
78
+ finally:
79
+ await provider.aclose()
80
+ await self._sleep(self.settings.market_data_population_request_interval_seconds)
81
82
async def submit(
83
self,
ai/research-engine/app/nse_historical_daily.py
new
+255
@@ -0,0 +1,255 @@
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
ai/research-engine/app/settings.py
+4
-1
@@ -5,7 +5,7 @@ from contextvars import ContextVar, Token
5
from datetime import datetime, timezone
6
from uuid import UUID
7
8
-from pydantic import field_validator
8
+from pydantic import Field, field_validator
9
from pydantic_settings import BaseSettings, SettingsConfigDict
10
11
@@ -106,6 +106,9 @@ class Settings(BaseSettings):
106
portfolio_service_base_url: str = "http://portfolio-service"
107
market_data_nifty_refresh_timeout_seconds: float = 30.0
108
market_data_population_batch_size: int = 50
109
+ # Operational single-request bound, not a claimed NSE maximum.
110
+ nse_historical_request_window_days: int = Field(default=30, ge=1)
111
+ nse_historical_max_retries: int = Field(default=2, ge=0, le=3)
112
market_data_population_request_interval_seconds: float = 0.20
113
market_data_population_initial_lookback_days: int = 400
114
market_data_nifty_freshness_hours: int = 12
ai/research-engine/tests/fixtures/nse_historical_daily.csv
new
+5
@@ -0,0 +1,5 @@
1
+"Symbol ","Series ","Date ","Prev Close ","Open Price ","High Price ","Low Price ","Last Price ","Close Price ","Average Price ","Total Traded Quantity ","Turnover ₹ ","No. of Trades ","Deliverable Qty ","% Dly Qt to Traded Qty "
2
+"POLYCAB","EQ","04-Sep-2026","8,807.50","8,500.00","8,500.00","8,250.00","8,300.00","8,300.00","8,351.05","15,59,741","13,02,54,81,530.00","1,62,753","7,21,733","46.27"
3
+"POLYCAB","EQ","03-Sep-2026","8,804.00","8,879.50","8,930.00","8,787.00","8,807.50","8,807.50","8,853.10","2,02,650","1,79,40,81,382.50","27,588","97,817","48.27"
4
+"POLYCAB","EQ","02-Sep-2026","8,895.50","8,876.50","8,933.00","8,732.00","8,804.00","8,804.00","8,814.01","4,34,543","3,83,00,67,920.00","54,283","2,06,370","47.49"
5
+"POLYCAB","EQ","01-Sep-2026","9,445.00","9,200.00","9,209.00","8,871.00","8,895.50","8,895.50","9,004.96","6,53,035","5,88,05,56,999.50","72,556","3,57,376","54.73"
\ No newline at end of file
ai/research-engine/tests/test_nse_historical_daily.py
new
+225
@@ -0,0 +1,225 @@
1
+import csv
2
+import io
3
+import logging
4
+from datetime import date
5
+from decimal import Decimal
6
+from pathlib import Path
7
+from types import SimpleNamespace
8
+from unittest.mock import AsyncMock
9
+from uuid import UUID
10
+
11
+import httpx
12
+import pytest
13
+from app.nse_historical_daily import BOOTSTRAP, NseHistoricalDailyProvider, NseHistoricalResult, parse_csv, persist_daily_result, verified_identity
14
+from app.persistence import SqliteResearchPersistence
15
+from app.settings import Settings
16
+
17
+KEY=UUID(int=1)
18
+START, END=date(2026,9,1),date(2026,9,4)
19
+FIXTURE=Path(__file__).parent/'fixtures/nse_historical_daily.csv'
20
+META=dict(globalInstrumentId=str(KEY),status='ACTIVE',assetType='EQUITY',country='IN',primaryExchange='NSE',currency='INR',providerMappings=[dict(provider='NSE',providerSymbol='POLYCAB',status='VERIFIED',resolutionSource='OFFICIAL_NSE_NIFTY500',currency='INR')])
21
+
22
+def result(): return NseHistoricalResult(KEY,START,END,provider_symbol='POLYCAB')
23
+def modified(**changes):
24
+ rows=list(csv.reader(io.StringIO(FIXTURE.read_text(encoding='utf-8-sig'))))
25
+ headers=[h.strip() for h in rows[0]]
26
+ for key,value in changes.items(): rows[1][headers.index(key)]=value
27
+ out=io.StringIO(newline=''); csv.writer(out).writerows(rows)
28
+ return out.getvalue().encode()
29
+
30
+def test_real_contract():
31
+ r=result(); parse_csv(FIXTURE.read_bytes(),r,'INR')
32
+ assert r.rows_parsed==r.rows_accepted==4 and r.rows_rejected==0
33
+ assert r.first_trading_date==START and r.last_trading_date==END
34
+ assert r.bars[-1].turnover==Decimal('13025481530.00')
35
+ assert r.bars[-1].volume==1559741 and type(r.bars[-1].volume) is int
36
+ assert r.bars[0].close==Decimal('8895.50')
37
+ assert r.bars[0].retrieved_at.utcoffset().total_seconds()==0
38
+
39
+def test_optional_aliases_order_zero_precision():
40
+ r=result(); parse_csv(b'Close,Date,Symbol,Series,Open,High,Low,Turnover (in Lacs)\n2,01-09-2026, polycab , eq ,1,2,1,100\n\n',r,'INR')
41
+ assert r.bars[0].turnover is r.bars[0].volume is r.bars[0].previous_close is None
42
+ r=result(); parse_csv(modified(**{'Total Traded Quantity':'0','Turnover ₹':'0','Prev Close':'-','Open Price':'8500.123456789012'}),r,'INR')
43
+ assert r.bars[-1].turnover==r.bars[-1].volume==0 and r.bars[-1].previous_close is None
44
+ assert r.bars[-1].open==Decimal('8500.123456789012')
45
+
46
+@pytest.mark.parametrize('field,value',[('Symbol','OTHER'),('Series','BE'),('Date','31-Sep-2026'),('Date','31-Aug-2026'),('Open Price','NaN'),('High Price','1'),('Low Price','0'),('Close Price','-1'),('Prev Close','0'),('Total Traded Quantity','9223372036854775808'),('Total Traded Quantity','1.5'),('Total Traded Quantity','-1'),('Turnover ₹','-1'),('Open Price','1,2'),('Close Price','inf')])
47
+def test_row_validation(field,value):
48
+ r=result(); parse_csv(modified(**{field:value}),r,'INR')
49
+ assert r.rows_parsed==4 and r.rows_accepted==3 and r.rows_rejected==1
50
+
51
+def test_bigint_duplicates():
52
+ r=result(); parse_csv(modified(**{'Total Traded Quantity':'9223372036854775807'}),r,'INR')
53
+ assert r.bars[-1].volume==9223372036854775807
54
+ r=result(); parse_csv(modified(Date='03-Sep-2026'),r,'INR')
55
+ assert r.rows_accepted==2 and r.rejection_reasons=={'DUPLICATE_DATE':2}
56
+
57
+@pytest.mark.parametrize('body',[b'',b'<html>x</html>',b'{}',b'Date,Close\n1,2',b'Date,Symbol,Series,Open,High,Low,Close\n"unclosed'])
58
+def test_bad_contract(body):
59
+ with pytest.raises((ValueError,csv.Error)): parse_csv(body,result(),'INR')
60
+
61
+@pytest.mark.parametrize('change',[dict(providerMappings=[]),dict(status='INACTIVE'),dict(assetType='ETF'),dict(globalInstrumentId=str(UUID(int=2))),dict(providerMappings=META['providerMappings']*2),dict(currency=None,providerMappings=[dict(provider='NSE',providerSymbol='X',status='VERIFIED')]),*[dict(providerMappings=[META['providerMappings'][0]|u]) for u in [dict(status='UNVERIFIED'),dict(active=False),dict(providerSymbol=' '),dict(resolutionSource='BROKER_IMPORT_IDENTITY'),dict(currency='USD')]]])
62
+def test_identity(change):
63
+ with pytest.raises(ValueError): verified_identity(META|change,KEY)
64
+
65
+async def fetch_with(handler,metadata=META,**settings):
66
+ async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
67
+ sleep=AsyncMock()
68
+ provider=NseHistoricalDailyProvider(SimpleNamespace(global_instrument_metadata=AsyncMock(return_value=metadata)),Settings(**settings),client=client,sleep=sleep)
69
+ return await provider.fetch(KEY,start=START,end=END),sleep
70
+
71
+@pytest.mark.asyncio
72
+async def test_session_query_no_cookie_logs(caplog):
73
+ calls=[]
74
+ def handler(req):
75
+ calls.append(req)
76
+ assert req.headers['user-agent'].startswith('Mozilla/') and req.headers['referer']==BOOTSTRAP
77
+ assert 'accept-language' in req.headers
78
+ if len(calls)==1: return httpx.Response(200,headers={'set-cookie':'session=secret-cookie; Path=/; Secure'})
79
+ assert req.headers['cookie']=='session=secret-cookie'
80
+ assert dict(req.url.params)==dict(symbol='POLYCAB',series='EQ',csv='true',type='priceVolumeDeliverable',**{'from':'01-09-2026','to':'04-09-2026'})
81
+ return httpx.Response(200,content=FIXTURE.read_bytes(),headers={'content-type':'text/csv'})
82
+ with caplog.at_level(logging.INFO): r,sleep=await fetch_with(handler)
83
+ assert r.status=='SUCCESS' and len(calls)==2 and sleep.await_count>=1
84
+ assert 'secret-cookie' not in caplog.text
85
+
86
+@pytest.mark.asyncio
87
+@pytest.mark.parametrize('status,count',[(403,1),(404,1),(429,3),(500,3)])
88
+async def test_http_failures(status,count):
89
+ calls=[]
90
+ def handler(req):
91
+ if str(req.url)==BOOTSTRAP: return httpx.Response(200)
92
+ calls.append(req); return httpx.Response(status)
93
+ r,_=await fetch_with(handler)
94
+ assert r.failure_reason==f'HISTORICAL_HTTP_{status}' and len(calls)==count and not r.bars
95
+
96
+@pytest.mark.asyncio
97
+@pytest.mark.parametrize('body,ctype',[(b'','text/csv'),(b'<html>x</html>','text/html'),(b'wrong,headers','text/csv'),(b'Date,Symbol,Series,Open,High,Low,Close\n"oops','text/csv')])
98
+async def test_provider_bad_response(body,ctype):
99
+ r,_=await fetch_with(lambda req: httpx.Response(200) if str(req.url)==BOOTSTRAP else httpx.Response(200,content=body,headers={'content-type':ctype}))
100
+ assert r.status=='UNAVAILABLE' and r.failure_reason and not r.bars
101
+
102
+@pytest.mark.asyncio
103
+async def test_bootstrap_and_missing_identity():
104
+ r,_=await fetch_with(lambda req:httpx.Response(403))
105
+ assert r.failure_reason=='BOOTSTRAP_HTTP_403'
106
+ handler=AsyncMock()
107
+ r,_=await fetch_with(handler,metadata=META|dict(providerMappings=[]))
108
+ assert r.failure_reason=='NO_UNAMBIGUOUS_NSE_MAPPING'; handler.assert_not_called()
109
+
110
+@pytest.mark.asyncio
111
+@pytest.mark.parametrize('error',[httpx.ReadTimeout,httpx.ConnectError])
112
+async def test_transport(error):
113
+ calls=[]
114
+ def handler(req): calls.append(req); raise error('unavailable')
115
+ r,_=await fetch_with(handler)
116
+ assert len(calls)==3 and r.failure_reason in {'BOOTSTRAP_TIMEOUT','BOOTSTRAP_CONNECTION_FAILURE'}
117
+
118
+@pytest.mark.asyncio
119
+async def test_persistence():
120
+ store=SqliteResearchPersistence()
121
+ repo=SimpleNamespace(upsert_daily_market_bars_async=AsyncMock(side_effect=store.upsert_daily_market_bars))
122
+ r=result(); parse_csv(FIXTURE.read_bytes(),r,'INR'); r.status='SUCCESS'
123
+ other=r.bars[0].model_copy(update={'provider':'OTHER'}); store.upsert_daily_market_bar(other)
124
+ await persist_daily_result(repo,r); await persist_daily_result(repo,r)
125
+ r.bars[0]=r.bars[0].model_copy(update={'close':Decimal('8900')}); await persist_daily_result(repo,r)
126
+ loaded=store.load_daily_market_bars({KEY})
127
+ assert len(loaded)==5 and other in loaded and r.persisted_rows==4
128
+ assert store.load_daily_market_bars({KEY},provider='NSE')[0].close==Decimal('8900')
129
+ assert store.load_market_price_observations({KEY})==[]
130
+ await persist_daily_result(repo,result())
131
+ assert store.load_daily_market_bars({KEY})==loaded
132
+
133
+@pytest.mark.asyncio
134
+async def test_window_rejected_before_identity_or_network():
135
+ resolver=SimpleNamespace(global_instrument_metadata=AsyncMock())
136
+ async with httpx.AsyncClient(transport=httpx.MockTransport(lambda req: pytest.fail('network'))) as client:
137
+ p=NseHistoricalDailyProvider(resolver,Settings(nse_historical_request_window_days=2),client=client)
138
+ r=await p.fetch(KEY,start=START,end=END)
139
+ assert r.failure_reason=='REQUEST_WINDOW_EXCEEDED'
140
+ r=await p.fetch(KEY,start=END,end=START)
141
+ assert r.failure_reason=='INVALID_DATE_RANGE'
142
+ resolver.global_instrument_metadata.assert_not_called()
143
+
144
+@pytest.mark.asyncio
145
+async def test_throttle_recovery_and_long_retry_after():
146
+ calls=[]
147
+ def handler(req):
148
+ if str(req.url)==BOOTSTRAP: return httpx.Response(200)
149
+ calls.append(req)
150
+ return httpx.Response(429,headers={'retry-after':'5'}) if len(calls)==1 else httpx.Response(200,content=FIXTURE.read_bytes())
151
+ r,sleep=await fetch_with(handler)
152
+ assert r.status=='SUCCESS' and len(calls)==2
153
+ assert any(c.args==(5,) for c in sleep.await_args_list)
154
+ calls.clear()
155
+ def throttled(req):
156
+ if str(req.url)==BOOTSTRAP: return httpx.Response(200)
157
+ calls.append(req); return httpx.Response(429,headers={'retry-after':'120'})
158
+ r,_=await fetch_with(throttled)
159
+ assert r.failure_reason=='HISTORICAL_HTTP_429' and len(calls)==1
160
+
161
+@pytest.mark.asyncio
162
+async def test_session_reused_serially():
163
+ active=0
164
+ calls=[]
165
+ async def handler(req):
166
+ nonlocal active
167
+ import asyncio
168
+ active+=1; assert active==1
169
+ await asyncio.sleep(0)
170
+ active-=1; calls.append(req)
171
+ return httpx.Response(200,content=FIXTURE.read_bytes())
172
+ import asyncio
173
+ async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
174
+ p=NseHistoricalDailyProvider(SimpleNamespace(global_instrument_metadata=AsyncMock(return_value=META)),Settings(),client=client,sleep=AsyncMock())
175
+ results=await asyncio.gather(p.fetch(KEY,start=START,end=END),p.fetch(KEY,start=START,end=END))
176
+ assert all(r.status=='SUCCESS' for r in results)
177
+ assert sum(str(r.url)==BOOTSTRAP for r in calls)==1
178
+
179
+@pytest.mark.asyncio
180
+async def test_explicit_population_boundary(monkeypatch):
181
+ from app.market_data_population import IndiaMarketDataPopulationJobs
182
+ from app.repository import ResearchRepository
183
+ import app.nse_historical_daily as module
184
+ store=SqliteResearchPersistence()
185
+ repo=ResearchRepository(settings=Settings(),persistence=store)
186
+ resolver=SimpleNamespace(global_instrument_metadata=AsyncMock(return_value=META))
187
+ async with httpx.AsyncClient(transport=httpx.MockTransport(lambda req:httpx.Response(200,content=FIXTURE.read_bytes()))) as client:
188
+ provider=NseHistoricalDailyProvider(resolver,Settings(),client=client,sleep=AsyncMock())
189
+ monkeypatch.setattr(module,'NseHistoricalDailyProvider',lambda *args:provider)
190
+ jobs=IndiaMarketDataPopulationJobs(repo,None,resolver,Settings(),sleep=AsyncMock())
191
+ r=await jobs.populate_daily_bars(KEY,start=START,end=END,identity_headers={})
192
+ assert r.persisted_rows==4
193
+ assert len((await repo.daily_market_bars_for_instruments({KEY},provider='NSE'))[KEY])==4
194
+ assert store.load_market_price_observations({KEY})==[]
195
+ assert jobs.historical_provider.provider_name=='YAHOO_FINANCE'
196
+
197
+@pytest.mark.asyncio
198
+async def test_persistence_failure_explicit():
199
+ r=result(); parse_csv(FIXTURE.read_bytes(),r,'INR'); r.status='SUCCESS'
200
+ await persist_daily_result(SimpleNamespace(upsert_daily_market_bars_async=AsyncMock(side_effect=RuntimeError())),r)
201
+ assert r.failure_reason=='DAILY_BAR_PERSISTENCE_UNAVAILABLE' and r.persisted_rows==0
202
+
203
+def test_malformed_row_shape_and_duplicate_header():
204
+ r=result(); parse_csv(b'Date,Symbol,Series,Open,High,Low,Close\n01-09-2026,POLYCAB,EQ,1,2,1,2,extra\n',r,'INR')
205
+ assert r.rows_rejected==1 and r.rejection_reasons=={'MALFORMED_ROW':1}
206
+ with pytest.raises(ValueError): parse_csv(b'Date,Symbol,Series,Open,Open Price,High,Low,Close\n',result(),'INR')
207
+
208
+def test_observed_rupee_turnover_matches_quantity_times_average():
209
+ # NSE average is rounded to two decimals; turnover is explicitly rupees.
210
+ for row in csv.DictReader(io.StringIO(FIXTURE.read_text(encoding='utf-8-sig'))):
211
+ row={k.strip():v for k,v in row.items()}
212
+ qty=Decimal(row['Total Traded Quantity'].replace(',',''))
213
+ average=Decimal(row['Average Price'].replace(',',''))
214
+ turnover=Decimal(row['Turnover ₹'].replace(',',''))
215
+ assert abs(turnover-qty*average)<=qty*Decimal('0.005')
216
+
217
+@pytest.mark.asyncio
218
+async def test_generic_verified_symbol_not_primary_symbol():
219
+ metadata=META|dict(primarySymbol='DO_NOT_USE',providerMappings=[META['providerMappings'][0]|dict(providerSymbol='ANOTHER')])
220
+ def handler(req):
221
+ if str(req.url)==BOOTSTRAP: return httpx.Response(200)
222
+ assert req.url.params['symbol']=='ANOTHER'
223
+ return httpx.Response(200,content=FIXTURE.read_bytes().replace(b'POLYCAB',b'ANOTHER'))
224
+ r,_=await fetch_with(handler,metadata=metadata)
225
+ assert r.rows_accepted==4 and all(b.provider_symbol=='ANOTHER' and b.global_instrument_id==KEY for b in r.bars)