main
py 1,130 lines 62.4 KB
Raw
1 from datetime import datetime, timedelta, timezone
2 from decimal import Decimal
3 import logging
4 from uuid import UUID, uuid4
5
6 import pytest
7 import httpx
8
9 from app.models import (
10 CompanyResearchProfile, DocumentStatus, DocumentType, ReliabilityLevel, ResearchDocument,
11 ShareholdingCategory, ShareholdingSnapshot, ShareholdingSnapshotValue, SourceClassification,
12 SourceMode, SourceType,
13 )
14 from app.persistence import SqliteResearchPersistence
15 from app.repository import ResearchRepository, _fair_official_filing_order
16 from app.research_fetching import FetchError, FetchResult, HttpResearchFetcher
17 from app.source_discovery import DiscoveryResult
18 from app.source_registry import RegisteredResearchSource
19 from app.settings import Settings
20 from app.shareholding import parse_nse_shareholding_xbrl, parse_official_shareholding
21 from app.source_discovery import OfficialFilingDiscovery, OfficialNseShareholdingDiscovery, SearchProviderError, _is_shareholding_announcement
22 from app.portfolio_orchestration import _global_master_instrument
23
24
25 def _snapshot(instrument_id, source="nse-filing-1"):
26 return ShareholdingSnapshot(instrument_id=instrument_id, period_end=datetime(2026, 6, 30, tzinfo=timezone.utc),
27 source_provider="NSE", source_type="EXCHANGE_ANNOUNCEMENT", source_identity_key=source,
28 source_url="https://www.nseindia.com/files/shareholding.pdf", confidence=Decimal("0.90"),
29 reliability_level=ReliabilityLevel.LEVEL_A, source_mode=SourceMode.REAL,
30 values=[ShareholdingSnapshotValue(category=ShareholdingCategory.PROMOTER, percentage=Decimal("42.5"), raw_source_label="Promoter")])
31
32
33 class _EmptyOfficialFilingDiscovery:
34 async def discover(self, *_args):
35 return []
36
37
38 def _nse_xbrl_fixture() -> str:
39 contexts = {
40 "promoter": "ShareholdingOfPromoterAndPromoterGroupMember",
41 "dii": "InstitutionsDomesticMember",
42 "mf": "MutualFundsOrUTIMember",
43 "insurance": "InsuranceCompaniesMember",
44 "government": "GovernmentsMember",
45 "retail": "ResidentIndividualShareholdersHoldingNominalShareCapitalUpToRsTwoLakhMember",
46 "others": "OtherNonInstitutionsMember",
47 "fpi1": "InstitutionsForeignPortfolioInvestorCategoryOneMember",
48 "fpi2": "InstitutionsForeignPortfolioInvestorCategoryTwoMember",
49 "public": "PublicShareholdingMember",
50 }
51 context_xml = "".join(
52 f'<xbrli:context id="{key}"><xbrli:scenario><xbrldi:explicitMember>{member}</xbrldi:explicitMember></xbrli:scenario></xbrli:context>'
53 for key, member in contexts.items()
54 )
55 ratios = {
56 "promoter": "0.494", "dii": "0.0068", "mf": "0.0014", "insurance": "0.003",
57 "government": "0", "retail": "0.2476", "others": "0.0168", "fpi1": "0.0789",
58 "fpi2": "0.0027", "public": "0.506",
59 }
60 fact_xml = "".join(
61 f'<n:ShareholdingAsAPercentageOfTotalNumberOfShares contextRef="{key}">{value}</n:ShareholdingAsAPercentageOfTotalNumberOfShares>'
62 for key, value in ratios.items()
63 )
64 fact_xml += '<n:EncumberedShareUnderPledgedAsPercentageOfTotalNumberOfShares contextRef="promoter">0.4474</n:EncumberedShareUnderPledgedAsPercentageOfTotalNumberOfShares>'
65 return f'<xbrli:xbrl xmlns:xbrli="http://www.xbrl.org/2003/instance" xmlns:xbrldi="http://xbrl.org/2006/xbrldi" xmlns:n="urn:test">{context_xml}{fact_xml}</xbrli:xbrl>'
66
67
68 def _legacy_nse_xbrl_snapshot(instrument_id, source: str, period: tuple[int, int, int]) -> ShareholdingSnapshot:
69 snapshot = _snapshot(instrument_id, source)
70 return snapshot.model_copy(update={
71 "period_end": datetime(*period, tzinfo=timezone.utc),
72 "source_type": "NSE_SHAREHOLDING_XBRL",
73 "source_identity_key": f"NSE_SHAREHOLDING:{source}",
74 "source_url": f"https://nsearchives.nseindia.com/corporate/xbrl/SHP_{source}.xml",
75 })
76
77
78 @pytest.mark.asyncio
79 async def test_nse_xbrl_fetch_uses_scoped_xml_headers_without_leaking_to_generic_fetches() -> None:
80 settings = Settings(research_max_retries=0)
81 requests: list[httpx.Request] = []
82
83 def handler(request: httpx.Request) -> httpx.Response:
84 requests.append(request)
85 if request.url.path.endswith(".xml"):
86 return httpx.Response(200, content=_nse_xbrl_fixture().encode(), headers={"content-type": "application/xml"}, request=request)
87 return httpx.Response(200, text="ordinary page", headers={"content-type": "text/plain"}, request=request)
88
89 client = httpx.AsyncClient(
90 transport=httpx.MockTransport(handler),
91 headers={"User-Agent": settings.research_user_agent, "Accept-Encoding": "gzip, deflate, br"},
92 )
93 fetcher = HttpResearchFetcher(settings, client=client)
94 try:
95 xml = await fetcher.fetch_nse_shareholding_xbrl("https://nsearchives.nseindia.com/corporate/xbrl/fixture.xml")
96 ordinary = await fetcher.fetch("https://example.com/ordinary.txt")
97 finally:
98 await client.aclose()
99
100 assert xml.content_type == "application/xml"
101 assert ordinary.content_type == "text/plain"
102 nse_headers, generic_headers = requests[0].headers, requests[1].headers
103 assert nse_headers["referer"] == "https://www.nseindia.com/"
104 assert nse_headers["accept"] == "application/xml,text/xml,*/*"
105 assert nse_headers["user-agent"].startswith("Mozilla/5.0")
106 assert generic_headers.get("referer") is None
107 assert generic_headers["user-agent"] == settings.research_user_agent
108
109
110 def test_nse_xbrl_normalizes_only_explicit_category_facts_with_provenance() -> None:
111 values = {value.category: value for value in parse_nse_shareholding_xbrl(_nse_xbrl_fixture())}
112
113 assert values[ShareholdingCategory.PROMOTER].percentage == Decimal("49.400")
114 assert values[ShareholdingCategory.DII].percentage == Decimal("0.6800")
115 assert values[ShareholdingCategory.MUTUAL_FUNDS].percentage == Decimal("0.1400")
116 assert values[ShareholdingCategory.INSURANCE].percentage == Decimal("0.300")
117 assert values[ShareholdingCategory.GOVERNMENT].percentage == Decimal("0")
118 assert values[ShareholdingCategory.PUBLIC_RETAIL].percentage == Decimal("24.7600")
119 assert values[ShareholdingCategory.OTHERS].percentage == Decimal("1.6800")
120 assert values[ShareholdingCategory.FII_FPI].percentage == Decimal("8.1600")
121 assert values[ShareholdingCategory.FII_FPI].metric_basis == "DERIVED_SUM_OF_MUTUALLY_EXCLUSIVE_FPI_CATEGORIES"
122 assert values[ShareholdingCategory.PROMOTER_PLEDGE].percentage == Decimal("44.7400")
123 assert values[ShareholdingCategory.PROMOTER_PLEDGE].metric_basis == "PERCENT_OF_PROMOTER_HOLDING"
124 assert values[ShareholdingCategory.PUBLIC_RETAIL].raw_source_label != "Public Shareholding"
125 assert all(value.source_locator and value.source_locator.startswith("nse-xbrl:") for value in values.values())
126 assert all(value.evidence_text for value in values.values())
127
128
129 def test_nse_xbrl_does_not_create_categories_when_explicit_facts_are_absent() -> None:
130 xml = _nse_xbrl_fixture().replace(
131 '<n:ShareholdingAsAPercentageOfTotalNumberOfShares contextRef="retail">0.2476</n:ShareholdingAsAPercentageOfTotalNumberOfShares>',
132 "",
133 ).replace(
134 '<n:ShareholdingAsAPercentageOfTotalNumberOfShares contextRef="fpi2">0.0027</n:ShareholdingAsAPercentageOfTotalNumberOfShares>',
135 "",
136 ).replace(
137 '<n:EncumberedShareUnderPledgedAsPercentageOfTotalNumberOfShares contextRef="promoter">0.4474</n:EncumberedShareUnderPledgedAsPercentageOfTotalNumberOfShares>',
138 "",
139 )
140 categories = {value.category for value in parse_nse_shareholding_xbrl(xml)}
141
142 assert ShareholdingCategory.PUBLIC_RETAIL not in categories
143 assert ShareholdingCategory.FII_FPI not in categories
144 assert ShareholdingCategory.PROMOTER_PLEDGE not in categories
145 assert ShareholdingCategory.PROMOTER in categories
146
147
148 def test_global_snapshot_persistence_deduplicates_source_and_values() -> None:
149 persistence = SqliteResearchPersistence()
150 instrument_id = uuid4()
151 assert persistence.upsert_shareholding_snapshot(_snapshot(instrument_id)) is True
152 assert persistence.upsert_shareholding_snapshot(_snapshot(instrument_id)) is False
153 loaded = persistence.load_shareholding_snapshots()
154 assert len(loaded) == 1
155 assert loaded[0].instrument_id == instrument_id
156 assert [(item.category, item.percentage) for item in loaded[0].values] == [(ShareholdingCategory.PROMOTER, Decimal("42.5"))]
157
158
159 def test_existing_official_snapshot_accepts_new_xbrl_category_values_once() -> None:
160 persistence = SqliteResearchPersistence()
161 instrument_id = uuid4()
162 snapshot = _snapshot(instrument_id, "NSE_SHAREHOLDING:12345")
163 snapshot.source_type = "NSE_SHAREHOLDING_XBRL"
164 assert persistence.upsert_shareholding_snapshot(snapshot) is True
165
166 enriched = snapshot.model_copy(update={"values": parse_nse_shareholding_xbrl(_nse_xbrl_fixture())})
167 assert persistence.upsert_shareholding_snapshot(enriched) is True
168 assert persistence.upsert_shareholding_snapshot(enriched) is False
169
170 loaded = persistence.load_shareholding_snapshots()
171 assert len(loaded) == 1
172 values = {value.category: value for value in loaded[0].values}
173 assert values[ShareholdingCategory.PROMOTER].percentage == Decimal("42.5")
174 assert values[ShareholdingCategory.FII_FPI].percentage == Decimal("8.1600")
175 assert values[ShareholdingCategory.PUBLIC_RETAIL].percentage == Decimal("24.7600")
176
177
178 def test_shareholding_persistence_accepts_native_postgres_uuid_rows_and_existing_id() -> None:
179 snapshot_id, instrument_id, value_id, existing_id = uuid4(), uuid4(), uuid4(), uuid4()
180 database_now = datetime(2026, 7, 20, 12, 0, 0)
181
182 class Cursor:
183 def __init__(self, *, rows=None, row=None) -> None:
184 self._rows = rows or []
185 self._row = row
186 def fetchall(self):
187 return self._rows
188 def fetchone(self):
189 return self._row
190
191 class NativeUuidConnection:
192 def execute(self, sql, _params=None):
193 if "global_shareholding_snapshot_values" in sql:
194 return Cursor(rows=[{
195 "id": value_id, "snapshot_id": snapshot_id, "category": "PROMOTER", "percentage": "49.40",
196 "metric_basis": None, "raw_source_label": "Promoter and Promoter Group",
197 "source_locator": "official:pr_and_prgrp", "evidence_text": "Promoter and Promoter Group: 49.40%",
198 "created_at": database_now,
199 }])
200 if "SELECT * FROM global_shareholding_snapshots" in sql:
201 return Cursor(rows=[{
202 "id": snapshot_id, "instrument_id": instrument_id, "period_end": datetime(2026, 6, 30),
203 "filing_basis": None, "source_provider": "NSE", "source_type": "NSE_SHAREHOLDING_XBRL",
204 "source_identity_key": "NSE_SHAREHOLDING:123", "source_url": "https://nsearchives.nseindia.com/xbrl.xml",
205 "research_document_id": None, "published_at": database_now, "retrieved_at": database_now, "confidence": "0.95",
206 "reliability_level": "LEVEL_A", "source_mode": "REAL", "created_at": database_now, "updated_at": database_now,
207 }])
208 if "SELECT id FROM global_shareholding_snapshots" in sql:
209 return Cursor(row={"id": existing_id})
210 raise AssertionError(f"unexpected SQL: {sql}")
211 def __enter__(self):
212 return self
213 def __exit__(self, *_args):
214 return False
215
216 persistence = SqliteResearchPersistence()
217 persistence._connection = NativeUuidConnection()
218 loaded = persistence.load_shareholding_snapshots()
219 assert len(loaded) == 1
220 assert loaded[0].id == snapshot_id
221 assert loaded[0].instrument_id == instrument_id
222 assert loaded[0].values[0].id == value_id
223 assert loaded[0].values[0].category == ShareholdingCategory.PROMOTER
224 assert loaded[0].retrieved_at == database_now.replace(tzinfo=timezone.utc)
225 assert loaded[0].period_end.tzinfo == timezone.utc
226
227 duplicate = _snapshot(instrument_id, "NSE_SHAREHOLDING:123")
228 assert persistence.upsert_shareholding_snapshot(duplicate) is False
229 assert duplicate.id == existing_id
230
231
232 def test_shareholding_freshness_uses_utc_normalized_database_timestamp_without_changing_ttl() -> None:
233 instrument_id = uuid4()
234 persisted_at = datetime(2026, 7, 20, 12, 0, 0, tzinfo=timezone.utc)
235 snapshot = _snapshot(instrument_id, "postgres-naive-timestamp")
236 snapshot.retrieved_at = persisted_at
237 repository = ResearchRepository(settings=Settings(research_shareholding_freshness_seconds=3600))
238 repository.shareholding_snapshots[snapshot.id] = snapshot
239
240 assert repository._category_is_fresh(
241 instrument_id, "SHAREHOLDING_PATTERN", persisted_at + timedelta(minutes=59)
242 ) is True
243 assert repository._category_is_fresh(
244 instrument_id, "SHAREHOLDING_PATTERN", persisted_at + timedelta(seconds=3601)
245 ) is False
246
247
248 def test_pledge_requires_explicit_basis_and_is_not_an_ownership_residual() -> None:
249 with pytest.raises(ValueError, match="PROMOTER_PLEDGE"):
250 ShareholdingSnapshot(instrument_id=uuid4(), period_end=datetime(2026, 6, 30, tzinfo=timezone.utc),
251 source_provider="NSE", source_type="EXCHANGE_ANNOUNCEMENT", source_identity_key="pledge-without-basis",
252 source_url="https://www.nseindia.com/file.pdf", confidence=Decimal("0.90"),
253 reliability_level=ReliabilityLevel.LEVEL_A, source_mode=SourceMode.REAL,
254 values=[ShareholdingSnapshotValue(category=ShareholdingCategory.PROMOTER_PLEDGE, percentage=Decimal("10"))])
255 pledge = ShareholdingSnapshotValue(category=ShareholdingCategory.PROMOTER_PLEDGE,
256 percentage=Decimal("10"), metric_basis="PERCENT_OF_PROMOTER_HOLDING")
257 snapshot = _snapshot(uuid4()).model_copy(update={"values": [_snapshot(uuid4()).values[0], pledge]})
258 assert snapshot.values[-1].metric_basis == "PERCENT_OF_PROMOTER_HOLDING"
259
260
261 def test_official_fixture_normalizes_real_categories_without_missing_zeroes() -> None:
262 instrument_id = uuid4()
263 document = ResearchDocument(
264 instrument_id=instrument_id, company_id=uuid4(), canonical_url="https://www.nseindia.com/file.pdf",
265 original_url="https://www.nseindia.com/file.pdf", source_type=SourceType.EXCHANGE_ANNOUNCEMENT,
266 source_classification=SourceClassification.EXCHANGE, source_name="NSE", content_type="application/pdf",
267 document_type=DocumentType.PDF_REFERENCE, content_hash="a" * 64, status=DocumentStatus.PROCESSED,
268 reliability_level=ReliabilityLevel.LEVEL_A, entity_resolution_confidence=0.99, source_mode=SourceMode.REAL,
269 normalized_text="Shareholding Pattern as on 30/06/2026. Promoter and Promoter Group: 42.50%. Foreign Portfolio Investors: 15.25%. Public Shareholders: 42.25%.",
270 )
271 snapshot = parse_official_shareholding(document)
272 assert snapshot is not None
273 values = {value.category: value.percentage for value in snapshot.values}
274 assert values[ShareholdingCategory.PROMOTER] == Decimal("42.50")
275 assert values[ShareholdingCategory.FII_FPI] == Decimal("15.25")
276 assert ShareholdingCategory.INSURANCE not in values
277
278
279 def test_invalid_percentages_are_rejected_not_persisted() -> None:
280 document = ResearchDocument(instrument_id=uuid4(), company_id=uuid4(), canonical_url="https://www.nseindia.com/file.pdf",
281 original_url="https://www.nseindia.com/file.pdf", source_type=SourceType.EXCHANGE_ANNOUNCEMENT,
282 source_classification=SourceClassification.EXCHANGE, source_name="NSE", content_type="application/pdf",
283 document_type=DocumentType.PDF_REFERENCE, content_hash="b" * 64, status=DocumentStatus.PROCESSED,
284 reliability_level=ReliabilityLevel.LEVEL_A, entity_resolution_confidence=0.99, source_mode=SourceMode.REAL,
285 normalized_text="Shareholding Pattern as on 30/06/2026. Promoter: 120%.")
286 assert parse_official_shareholding(document) is None
287
288
289 @pytest.mark.parametrize("title", [
290 "Shareholding Pattern",
291 "Shareholding Pattern for the quarter ended June 30, 2026",
292 "Share Holder Pattern",
293 "Regulation 31 - Shareholding Pattern",
294 ])
295 def test_nse_shareholding_classifier_accepts_only_explicit_pattern_filings(title: str) -> None:
296 assert _is_shareholding_announcement(title)
297
298
299 @pytest.mark.parametrize("title", [
300 "Change in Shareholding",
301 "Promoter Shareholding Update",
302 "Promoter Pledge",
303 "Outcome of Board Meeting",
304 "Financial Results",
305 "Investor Presentation",
306 "Acquisition resulting in change in shareholding",
307 ])
308 def test_nse_shareholding_classifier_rejects_non_pattern_announcements(title: str) -> None:
309 assert not _is_shareholding_announcement(title)
310
311
312 @pytest.mark.asyncio
313 async def test_official_discovery_excludes_unrelated_shareholding_rows() -> None:
314 profile = CompanyResearchProfile(instrument_id=uuid4(), company_id=uuid4(), company_name="Generic India Equity",
315 ticker="BROKER_ALIAS", exchange="NSE", mic="XNSE", country="IN", currency="INR",
316 provider_instrument_ids={"NSE": "VERIFIED_SYMBOL"})
317 rows = [
318 {"an_dt": "11-Aug-2026 13:52:28", "desc": "Change in Shareholding", "attchmntText": "Acquisition resulting in change in shareholding", "attchmntFile": "https://nsearchives.nseindia.com/corporate/unrelated.pdf"},
319 {"an_dt": "10-Aug-2026 13:52:28", "desc": "Regulation 31 - Shareholding Pattern", "attchmntText": "Shareholding Pattern for the quarter ended June 30, 2026", "attchmntFile": "https://nsearchives.nseindia.com/corporate/pattern.pdf"},
320 {"an_dt": "09-Aug-2026 13:52:28", "desc": "Promoter Pledge", "attchmntText": "Promoter shareholding update", "attchmntFile": "https://nsearchives.nseindia.com/corporate/pledge.pdf"},
321 ]
322 client = httpx.AsyncClient(transport=httpx.MockTransport(lambda request: httpx.Response(200, json=rows, request=request)))
323 try:
324 discovered = await OfficialFilingDiscovery(client).discover(profile, {"SHAREHOLDING_PATTERN"}, set())
325 finally:
326 await client.aclose()
327 assert [result.source.url for result in discovered] == ["https://nsearchives.nseindia.com/corporate/pattern.pdf"]
328
329
330 @pytest.mark.asyncio
331 async def test_dedicated_nse_shareholding_discovery_uses_trusted_mapping_and_official_xbrl_provenance() -> None:
332 instrument_id = uuid4()
333 profile = CompanyResearchProfile(instrument_id=instrument_id, company_id=uuid4(), company_name="Generic India Equity",
334 ticker="BROKER_ALIAS", exchange="NSE", mic="XNSE", country="IN", currency="INR",
335 provider_instrument_ids={"NSE": "OFFICIAL_SYMBOL"})
336 calls: list[httpx.Request] = []
337 rows = [{
338 "recordId": "12345", "symbol": "OFFICIAL_SYMBOL", "date": "30-JUN-2026",
339 "submissionDate": "20-JUL-2026", "xbrl": "https://nsearchives.nseindia.com/corporate/xbrl/SHP_12345.xml",
340 "pr_and_prgrp": "42.50", "public_val": "57.50", "employeeTrusts": "0",
341 }]
342 def handler(request: httpx.Request) -> httpx.Response:
343 calls.append(request)
344 return httpx.Response(200, json=rows, request=request)
345 client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
346 try:
347 snapshots = await OfficialNseShareholdingDiscovery(client).discover(profile)
348 finally:
349 await client.aclose()
350 assert len(calls) == 1
351 assert calls[0].url.params["symbol"] == "OFFICIAL_SYMBOL"
352 assert len(snapshots) == 1
353 snapshot = snapshots[0]
354 assert snapshot.instrument_id == instrument_id
355 assert snapshot.period_end == datetime(2026, 6, 30, tzinfo=timezone.utc)
356 assert snapshot.source_identity_key == "NSE_SHAREHOLDING:12345"
357 assert snapshot.source_url == "https://nsearchives.nseindia.com/corporate/xbrl/SHP_12345.xml"
358 assert {value.category: value.percentage for value in snapshot.values} == {
359 ShareholdingCategory.PROMOTER: Decimal("42.50"),
360 }
361 assert ShareholdingCategory.PUBLIC_RETAIL not in {value.category for value in snapshot.values}
362 assert len(snapshot.values) == 1
363
364
365 @pytest.mark.asyncio
366 async def test_dedicated_nse_shareholding_discovery_rejects_broker_alias_without_trusted_mapping() -> None:
367 profile = CompanyResearchProfile(instrument_id=uuid4(), company_id=uuid4(), company_name="Generic India Equity",
368 ticker="BROKER_ALIAS", exchange="NSE", mic="XNSE", country="IN", currency="INR")
369 calls = 0
370 def handler(request: httpx.Request) -> httpx.Response:
371 nonlocal calls
372 calls += 1
373 return httpx.Response(200, json=[], request=request)
374 client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
375 try:
376 assert await OfficialNseShareholdingDiscovery(client).discover(profile) == []
377 finally:
378 await client.aclose()
379 assert calls == 0
380
381
382 @pytest.mark.asyncio
383 async def test_dedicated_nse_shareholding_discovery_keeps_latest_four_distinct_official_periods() -> None:
384 profile = CompanyResearchProfile(instrument_id=uuid4(), company_id=uuid4(), company_name="Generic India Equity",
385 ticker="BROKER_ALIAS", exchange="NSE", mic="XNSE", country="IN", currency="INR",
386 provider_instrument_ids={"NSE": "OFFICIAL_SYMBOL"})
387 dates = ("30-JUN-2026", "31-MAR-2026", "31-DEC-2025", "30-SEP-2025", "30-JUN-2025")
388 rows = [{
389 "recordId": str(index), "symbol": "OFFICIAL_SYMBOL", "date": date,
390 "submissionDate": "20-JUL-2026", "xbrl": f"https://nsearchives.nseindia.com/corporate/xbrl/SHP_{index}.xml",
391 "pr_and_prgrp": "42.50", "public_val": "57.50",
392 } for index, date in enumerate(reversed(dates), start=1)]
393 client = httpx.AsyncClient(transport=httpx.MockTransport(lambda request: httpx.Response(200, json=rows, request=request)))
394 try:
395 snapshots = await OfficialNseShareholdingDiscovery(client).discover(profile)
396 finally:
397 await client.aclose()
398 assert [snapshot.period_end.date().isoformat() for snapshot in snapshots] == [
399 "2026-06-30", "2026-03-31", "2025-12-31", "2025-09-30",
400 ]
401
402
403 @pytest.mark.asyncio
404 async def test_dedicated_nse_shareholding_discovery_rejects_non_quarter_as_on_date_without_displacing_quarter() -> None:
405 profile = CompanyResearchProfile(instrument_id=uuid4(), company_id=uuid4(), company_name="Generic India Equity",
406 ticker="BROKER_ALIAS", exchange="NSE", mic="XNSE", country="IN", currency="INR",
407 provider_instrument_ids={"NSE": "OFFICIAL_SYMBOL"})
408 quarter_dates = ("30-JUN-2026", "31-MAR-2026", "31-DEC-2025", "30-SEP-2025", "30-JUN-2025")
409 rows = [{
410 "recordId": str(index), "symbol": "OFFICIAL_SYMBOL", "date": date,
411 "submissionDate": "20-JUL-2026", "xbrl": f"https://nsearchives.nseindia.com/corporate/xbrl/SHP_{index}.xml",
412 "pr_and_prgrp": "42.50", "public_val": "57.50",
413 } for index, date in enumerate(quarter_dates, start=1)]
414 rows.append({
415 # NSE's feed labels this generic field "As on Date". With no explicit
416 # quarterly period/type marker, it cannot be treated as Regulation 31
417 # quarterly evidence merely because it has a filing XBRL URL.
418 "recordId": "special", "symbol": "OFFICIAL_SYMBOL", "date": "18-FEB-2026",
419 "submissionDate": "26-FEB-2026", "broadcastDate": "26-FEB-2026 18:57:56",
420 "typeOfSubmission": None, "revisedData": "N",
421 "xbrl": "https://nsearchives.nseindia.com/corporate/xbrl/SHP_special.xml",
422 "pr_and_prgrp": "49.49", "public_val": "50.51",
423 })
424 client = httpx.AsyncClient(transport=httpx.MockTransport(lambda request: httpx.Response(200, json=rows, request=request)))
425 try:
426 snapshots = await OfficialNseShareholdingDiscovery(client).discover(profile)
427 finally:
428 await client.aclose()
429 assert [snapshot.period_end.date().isoformat() for snapshot in snapshots] == [
430 "2026-06-30", "2026-03-31", "2025-12-31", "2025-09-30",
431 ]
432 assert all(snapshot.source_identity_key != "NSE_SHAREHOLDING:special" for snapshot in snapshots)
433 assert all({value.category for value in snapshot.values} == {ShareholdingCategory.PROMOTER} for snapshot in snapshots)
434
435
436 @pytest.mark.asyncio
437 async def test_dedicated_nse_shareholding_retains_distinct_revisions_for_one_quarter() -> None:
438 profile = CompanyResearchProfile(instrument_id=uuid4(), company_id=uuid4(), company_name="Generic India Equity",
439 ticker="BROKER_ALIAS", exchange="NSE", mic="XNSE", country="IN", currency="INR",
440 provider_instrument_ids={"NSE": "OFFICIAL_SYMBOL"})
441 def row(record_id: str, submission_date: str, promoter: str) -> dict[str, str]:
442 return {
443 "recordId": record_id, "symbol": "OFFICIAL_SYMBOL", "date": "31-MAR-2026",
444 "submissionDate": submission_date, "xbrl": f"https://nsearchives.nseindia.com/corporate/xbrl/SHP_{record_id}.xml",
445 "pr_and_prgrp": promoter, "public_val": "50.00",
446 }
447 rows = [row("original", "20-APR-2026", "49.40"), row("revision", "22-APR-2026", "49.49")]
448 client = httpx.AsyncClient(transport=httpx.MockTransport(lambda request: httpx.Response(200, json=rows, request=request)))
449 try:
450 snapshots = await OfficialNseShareholdingDiscovery(client).discover(profile)
451 finally:
452 await client.aclose()
453 assert [snapshot.source_identity_key for snapshot in snapshots] == [
454 "NSE_SHAREHOLDING:revision", "NSE_SHAREHOLDING:original",
455 ]
456 assert all(snapshot.period_end == datetime(2026, 3, 31, tzinfo=timezone.utc) for snapshot in snapshots)
457 repository = ResearchRepository(settings=Settings())
458 for snapshot in snapshots:
459 assert repository.persist_shareholding_snapshot(snapshot) is True
460 latest = repository.shareholding_for(profile.instrument_id)
461 assert len(latest) == 1
462 assert latest[0].source_identity_key == "NSE_SHAREHOLDING:revision"
463
464
465 def test_existing_non_quarter_snapshot_does_not_pollute_quarterly_read_model() -> None:
466 repository = ResearchRepository(settings=Settings())
467 instrument_id = uuid4()
468 non_quarter = _snapshot(instrument_id, "legacy-special-filing")
469 non_quarter.period_end = datetime(2026, 2, 18, tzinfo=timezone.utc)
470 quarter = _snapshot(instrument_id, "quarterly-filing")
471 assert repository.persist_shareholding_snapshot(non_quarter) is True
472 assert repository.persist_shareholding_snapshot(quarter) is True
473 assert [snapshot.source_identity_key for snapshot in repository.shareholding_for(instrument_id)] == ["quarterly-filing"]
474
475
476 @pytest.mark.asyncio
477 async def test_dedicated_nse_shareholding_refresh_persists_once_and_reuses_durable_snapshot() -> None:
478 profile = CompanyResearchProfile(instrument_id=uuid4(), company_id=uuid4(), company_name="Generic India Equity",
479 ticker="BROKER_ALIAS", exchange="NSE", mic="XNSE", country="IN", currency="INR",
480 provider_instrument_ids={"NSE": "OFFICIAL_SYMBOL"})
481 rows = [{
482 "recordId": "12345", "symbol": "OFFICIAL_SYMBOL", "date": "30-JUN-2026",
483 "submissionDate": "20-JUL-2026", "xbrl": "https://nsearchives.nseindia.com/corporate/xbrl/SHP_12345.xml",
484 "pr_and_prgrp": "42.50", "public_val": "57.50",
485 }]
486 calls = 0
487 def handler(request: httpx.Request) -> httpx.Response:
488 nonlocal calls
489 calls += 1
490 return httpx.Response(200, json=rows, request=request)
491
492 class XbrlFetcher:
493 calls = 0
494
495 async def fetch(self, url: str) -> FetchResult:
496 self.calls += 1
497 return FetchResult(
498 final_url=url,
499 status_code=200,
500 content_type="application/xml",
501 text=_nse_xbrl_fixture(),
502 bytes_read=len(_nse_xbrl_fixture()),
503 )
504
505 fetcher = XbrlFetcher()
506 client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
507 try:
508 repository = ResearchRepository(
509 settings=Settings(research_live_enabled=True, research_search_enabled=False),
510 fetcher=fetcher,
511 official_filing_discovery=_EmptyOfficialFilingDiscovery(),
512 official_shareholding_discovery=OfficialNseShareholdingDiscovery(client),
513 persistence=SqliteResearchPersistence(),
514 )
515 repository.profiles = [profile]
516 await repository._refresh_targeted(profile, set())
517 await repository._refresh_targeted(profile, set())
518 finally:
519 await client.aclose()
520 # A fresh but incomplete history remains eligible for bounded
521 # reconciliation; durable source identity keeps the repeat idempotent.
522 assert calls == 2
523 assert fetcher.calls == 1
524 snapshots = repository.shareholding_for(profile.instrument_id)
525 assert len(snapshots) == 1
526 assert snapshots[0].source_mode == SourceMode.REAL
527 values = {value.category: value for value in snapshots[0].values}
528 assert values[ShareholdingCategory.PROMOTER].percentage == Decimal("49.400")
529 assert values[ShareholdingCategory.FII_FPI].percentage == Decimal("8.1600")
530 assert values[ShareholdingCategory.PUBLIC_RETAIL].percentage == Decimal("24.7600")
531 assert values[ShareholdingCategory.PROMOTER_PLEDGE].metric_basis == "PERCENT_OF_PROMOTER_HOLDING"
532 assert repository._category_is_fresh(profile.instrument_id, "SHAREHOLDING_PATTERN", datetime.now(timezone.utc))
533
534
535 @pytest.mark.asyncio
536 async def test_fresh_incomplete_shareholding_reconciles_missing_quarter_once_and_keeps_legacy_row() -> None:
537 profile = CompanyResearchProfile(instrument_id=uuid4(), company_id=uuid4(), company_name="Generic India Equity",
538 ticker="BROKER_ALIAS", exchange="NSE", mic="XNSE", country="IN", currency="INR",
539 provider_instrument_ids={"NSE": "OFFICIAL_SYMBOL"})
540 repository = ResearchRepository(settings=Settings(research_live_enabled=True, research_search_enabled=False))
541 repository.profiles = [profile]
542 for source, period in (("june", (2026, 6, 30)), ("march", (2026, 3, 31)), ("december", (2025, 12, 31))):
543 snapshot = _snapshot(profile.instrument_id, source)
544 snapshot.period_end = datetime(*period, tzinfo=timezone.utc)
545 assert repository.persist_shareholding_snapshot(snapshot) is True
546 legacy = _snapshot(profile.instrument_id, "legacy-non-quarter")
547 legacy.period_end = datetime(2026, 2, 18, tzinfo=timezone.utc)
548 assert repository.persist_shareholding_snapshot(legacy) is True
549 september = _snapshot(profile.instrument_id, "NSE_SHAREHOLDING:203088")
550 september.period_end = datetime(2025, 9, 30, tzinfo=timezone.utc)
551
552 class ShareholdingProvider:
553 def __init__(self) -> None:
554 self.calls = 0
555 async def discover(self, _profile):
556 self.calls += 1
557 return [september]
558 provider = ShareholdingProvider()
559 repository._official_filing_discovery = _EmptyOfficialFilingDiscovery()
560 repository._official_shareholding_discovery = provider
561
562 await repository._refresh_targeted(profile, set(), now=datetime(2026, 8, 15, tzinfo=timezone.utc))
563 await repository._refresh_targeted(profile, set(), now=datetime(2026, 8, 16, tzinfo=timezone.utc))
564
565 assert provider.calls == 1
566 assert [snapshot.period_end.date().isoformat() for snapshot in repository.shareholding_for(profile.instrument_id)] == [
567 "2026-06-30", "2026-03-31", "2025-12-31", "2025-09-30",
568 ]
569 assert any(snapshot.source_identity_key == "legacy-non-quarter" for snapshot in repository.shareholding_snapshots.values())
570 assert len([snapshot for snapshot in repository.shareholding_snapshots.values()
571 if snapshot.source_identity_key == "NSE_SHAREHOLDING:203088"]) == 1
572
573
574 @pytest.mark.asyncio
575 async def test_fresh_complete_shareholding_does_not_call_structured_reconciliation_provider() -> None:
576 profile = CompanyResearchProfile(instrument_id=uuid4(), company_id=uuid4(), company_name="Generic India Equity",
577 ticker="BROKER_ALIAS", exchange="NSE", mic="XNSE", country="IN", currency="INR",
578 provider_instrument_ids={"NSE": "OFFICIAL_SYMBOL"})
579 repository = ResearchRepository(settings=Settings(research_live_enabled=True, research_search_enabled=False))
580 repository.profiles = [profile]
581 for source, period in (("june", (2026, 6, 30)), ("march", (2026, 3, 31)), ("december", (2025, 12, 31)), ("september", (2025, 9, 30))):
582 snapshot = _snapshot(profile.instrument_id, source)
583 snapshot.period_end = datetime(*period, tzinfo=timezone.utc)
584 assert repository.persist_shareholding_snapshot(snapshot) is True
585 class Provider:
586 calls = 0
587 async def discover(self, _profile):
588 self.calls += 1
589 return []
590 provider = Provider()
591 repository._official_filing_discovery = _EmptyOfficialFilingDiscovery()
592 repository._official_shareholding_discovery = provider
593 await repository._refresh_targeted(profile, set(), now=datetime(2026, 8, 15, tzinfo=timezone.utc))
594 assert provider.calls == 0
595
596
597 @pytest.mark.asyncio
598 async def test_fresh_complete_legacy_nse_snapshots_are_enriched_once_without_duplicate_periods() -> None:
599 profile = CompanyResearchProfile(instrument_id=uuid4(), company_id=uuid4(), company_name="Generic India Equity",
600 ticker="BROKER_ALIAS", exchange="NSE", mic="XNSE", country="IN", currency="INR",
601 provider_instrument_ids={"NSE": "OFFICIAL_SYMBOL"})
602 repository = ResearchRepository(
603 settings=Settings(research_live_enabled=True, research_search_enabled=False),
604 persistence=SqliteResearchPersistence(),
605 official_filing_discovery=_EmptyOfficialFilingDiscovery(),
606 )
607 repository.profiles = [profile]
608 periods = ((2026, 6, 30), (2026, 3, 31), (2025, 12, 31), (2025, 9, 30))
609 legacy = [_legacy_nse_xbrl_snapshot(profile.instrument_id, str(index), period) for index, period in enumerate(periods, start=1)]
610 for snapshot in legacy:
611 assert repository.persist_shareholding_snapshot(snapshot) is True
612
613 class Provider:
614 calls = 0
615 async def discover(self, _profile):
616 self.calls += 1
617 return [snapshot.model_copy(update={"values": snapshot.values[:]}) for snapshot in legacy]
618
619 class XbrlFetcher:
620 calls: list[str] = []
621 async def fetch_nse_shareholding_xbrl(self, url: str) -> FetchResult:
622 self.calls.append(url)
623 return FetchResult(url, 200, "application/xml", _nse_xbrl_fixture(), len(_nse_xbrl_fixture()))
624
625 async def fetch(self, _url: str) -> FetchResult:
626 raise AssertionError("NSE XBRL enrichment must use the scoped fetch path")
627
628 provider, fetcher = Provider(), XbrlFetcher()
629 repository._official_shareholding_discovery = provider
630 repository._fetcher = fetcher
631
632 await repository._refresh_targeted(profile, set())
633
634 assert provider.calls == 1
635 assert len(fetcher.calls) == 4
636 snapshots = repository.shareholding_for(profile.instrument_id)
637 assert [snapshot.source_identity_key for snapshot in snapshots] == [
638 "NSE_SHAREHOLDING:1", "NSE_SHAREHOLDING:2", "NSE_SHAREHOLDING:3", "NSE_SHAREHOLDING:4",
639 ]
640 assert all(ShareholdingCategory.FII_FPI in {value.category for value in snapshot.values} for snapshot in snapshots)
641 assert all(any((value.source_locator or "").startswith("nse-xbrl:") for value in snapshot.values) for snapshot in snapshots)
642 assert len(repository.shareholding_snapshots) == 4
643
644 # XBRL provenance, not an all-categories requirement, prevents another
645 # fresh reconciliation even though optional official categories may be absent.
646 await repository._refresh_targeted(profile, set())
647 assert provider.calls == 1
648 assert len(fetcher.calls) == 4
649
650
651 @pytest.mark.asyncio
652 async def test_legacy_xbrl_enrichment_fetch_failure_keeps_existing_promoter_values_retryable() -> None:
653 profile = CompanyResearchProfile(instrument_id=uuid4(), company_id=uuid4(), company_name="Generic India Equity",
654 ticker="BROKER_ALIAS", exchange="NSE", mic="XNSE", country="IN", currency="INR",
655 provider_instrument_ids={"NSE": "OFFICIAL_SYMBOL"})
656 repository = ResearchRepository(
657 settings=Settings(research_live_enabled=True, research_search_enabled=False),
658 persistence=SqliteResearchPersistence(),
659 official_filing_discovery=_EmptyOfficialFilingDiscovery(),
660 )
661 repository.profiles = [profile]
662 legacy = [_legacy_nse_xbrl_snapshot(profile.instrument_id, str(index), period) for index, period in enumerate(
663 ((2026, 6, 30), (2026, 3, 31), (2025, 12, 31), (2025, 9, 30)), start=1)]
664 for snapshot in legacy:
665 assert repository.persist_shareholding_snapshot(snapshot) is True
666
667 class Provider:
668 calls = 0
669 async def discover(self, _profile):
670 self.calls += 1
671 return legacy
672
673 class FailingFetcher:
674 calls = 0
675 async def fetch(self, _url: str):
676 self.calls += 1
677 raise FetchError("fixture XBRL unavailable")
678
679 provider, fetcher = Provider(), FailingFetcher()
680 repository._official_shareholding_discovery = provider
681 repository._fetcher = fetcher
682 await repository._refresh_targeted(profile, set())
683
684 assert provider.calls == 1
685 assert fetcher.calls == 4
686 assert len(repository.shareholding_snapshots) == 4
687 assert all(
688 [(value.category, value.percentage) for value in snapshot.values] == [(ShareholdingCategory.PROMOTER, Decimal("42.5"))]
689 for snapshot in repository.shareholding_for(profile.instrument_id)
690 )
691
692
693 @pytest.mark.asyncio
694 async def test_stale_complete_shareholding_rechecks_nse_and_rolls_latest_four_without_deleting_history() -> None:
695 profile = CompanyResearchProfile(instrument_id=uuid4(), company_id=uuid4(), company_name="Generic India Equity",
696 ticker="BROKER_ALIAS", exchange="NSE", mic="XNSE", country="IN", currency="INR",
697 provider_instrument_ids={"NSE": "OFFICIAL_SYMBOL"})
698 settings = Settings(research_live_enabled=True, research_search_enabled=False, research_shareholding_freshness_seconds=3600)
699 repository = ResearchRepository(settings=settings)
700 repository.profiles = [profile]
701 stale_at = datetime.now(timezone.utc) - timedelta(seconds=settings.research_shareholding_freshness_seconds + 1)
702 for source, period in (("june", (2026, 6, 30)), ("march", (2026, 3, 31)), ("december", (2025, 12, 31)), ("september", (2025, 9, 30))):
703 snapshot = _snapshot(profile.instrument_id, source)
704 snapshot.period_end = datetime(*period, tzinfo=timezone.utc)
705 snapshot.retrieved_at = stale_at
706 assert repository.persist_shareholding_snapshot(snapshot) is True
707 legacy = _snapshot(profile.instrument_id, "legacy-non-quarter")
708 legacy.period_end = datetime(2026, 2, 18, tzinfo=timezone.utc)
709 legacy.retrieved_at = stale_at
710 assert repository.persist_shareholding_snapshot(legacy) is True
711 new_june = _snapshot(profile.instrument_id, "NSE_SHAREHOLDING:new-june")
712 new_june.period_end = datetime(2026, 9, 30, tzinfo=timezone.utc)
713
714 class Provider:
715 def __init__(self) -> None:
716 self.calls = 0
717 self.results: list[ShareholdingSnapshot] = []
718 async def discover(self, _profile):
719 self.calls += 1
720 return self.results
721 provider = Provider()
722 repository._official_filing_discovery = _EmptyOfficialFilingDiscovery()
723 repository._official_shareholding_discovery = provider
724
725 # Stale + complete must query NSE even when it has no new quarter.
726 await repository._refresh_targeted(profile, set())
727 assert provider.calls == 1
728 assert [snapshot.period_end.date().isoformat() for snapshot in repository.shareholding_for(profile.instrument_id)] == [
729 "2026-06-30", "2026-03-31", "2025-12-31", "2025-09-30",
730 ]
731
732 # Simulate the next normal refresh after the category freshness interval.
733 repository._category_refresh[(profile.instrument_id, "SHAREHOLDING_PATTERN")] = datetime.now(timezone.utc) - timedelta(days=4)
734 provider.results = [new_june]
735 await repository._refresh_targeted(profile, set())
736 assert provider.calls == 2
737 assert [snapshot.period_end.date().isoformat() for snapshot in repository.shareholding_for(profile.instrument_id)] == [
738 "2026-09-30", "2026-06-30", "2026-03-31", "2025-12-31",
739 ]
740 assert len([snapshot for snapshot in repository.shareholding_snapshots.values()
741 if (snapshot.period_end.month, snapshot.period_end.day) in {(3, 31), (6, 30), (9, 30), (12, 31)}]) == 5
742 assert any(snapshot.source_identity_key == "september" for snapshot in repository.shareholding_snapshots.values())
743 assert all(snapshot.period_end.date().isoformat() != "2026-02-18" for snapshot in repository.shareholding_for(profile.instrument_id))
744
745 # A later stale reconciliation may rediscover the same filing, but upsert
746 # preserves one durable official source identity.
747 repository._category_refresh[(profile.instrument_id, "SHAREHOLDING_PATTERN")] = datetime.now(timezone.utc) - timedelta(days=4)
748 await repository._refresh_targeted(profile, set(), now=datetime(2026, 12, 5, tzinfo=timezone.utc))
749 assert provider.calls == 3
750 assert len([snapshot for snapshot in repository.shareholding_snapshots.values()
751 if snapshot.source_identity_key == "NSE_SHAREHOLDING:new-june"]) == 1
752
753
754 @pytest.mark.asyncio
755 async def test_stale_complete_xbrl_shareholding_uses_master_check_without_redownloading_unchanged_records() -> None:
756 profile = CompanyResearchProfile(instrument_id=uuid4(), company_id=uuid4(), company_name="Generic India Equity",
757 ticker="BROKER_ALIAS", exchange="NSE", mic="XNSE", country="IN", currency="INR",
758 provider_instrument_ids={"NSE": "OFFICIAL_SYMBOL"})
759 settings = Settings(research_live_enabled=True, research_search_enabled=False, research_shareholding_freshness_seconds=3600)
760 repository = ResearchRepository(settings=settings)
761 repository.profiles = [profile]
762 stale_at = datetime.now(timezone.utc) - timedelta(seconds=settings.research_shareholding_freshness_seconds + 1)
763 snapshots = []
764 for source, period in (("june", (2026, 6, 30)), ("march", (2026, 3, 31)), ("december", (2025, 12, 31)), ("september", (2025, 9, 30))):
765 snapshot = _legacy_nse_xbrl_snapshot(profile.instrument_id, source, period)
766 snapshot.retrieved_at = stale_at
767 snapshot.values = [ShareholdingSnapshotValue(
768 category=ShareholdingCategory.PROMOTER,
769 percentage=Decimal("42.5"),
770 source_locator="nse-xbrl:explicit:promoter",
771 )]
772 assert repository.persist_shareholding_snapshot(snapshot) is True
773 snapshots.append(snapshot)
774
775 class Provider:
776 calls = 0
777 async def discover(self, _profile):
778 self.calls += 1
779 return snapshots
780
781 class Fetcher:
782 calls = 0
783 async def fetch_nse_shareholding_xbrl(self, _url):
784 self.calls += 1
785 raise AssertionError("unchanged XBRL must not be downloaded")
786
787 provider, fetcher = Provider(), Fetcher()
788 repository._official_filing_discovery = _EmptyOfficialFilingDiscovery()
789 repository._official_shareholding_discovery = provider
790 repository._fetcher = fetcher
791
792 await repository._refresh_targeted(profile, set())
793
794 assert provider.calls == 1
795 assert fetcher.calls == 0
796 assert {snapshot.source_identity_key for snapshot in repository.shareholding_snapshots.values()} == {
797 "NSE_SHAREHOLDING:june", "NSE_SHAREHOLDING:march", "NSE_SHAREHOLDING:december", "NSE_SHAREHOLDING:september",
798 }
799 assert repository._category_refresh[(profile.instrument_id, "SHAREHOLDING_PATTERN")] > stale_at
800
801
802 @pytest.mark.asyncio
803 async def test_quarterly_shareholding_gate_skips_before_next_window_then_checks_metadata_without_changing_evidence() -> None:
804 profile = CompanyResearchProfile(instrument_id=uuid4(), company_id=uuid4(), company_name="Generic India Equity",
805 ticker="BROKER_ALIAS", exchange="NSE", mic="XNSE", country="IN", currency="INR",
806 provider_instrument_ids={"NSE": "OFFICIAL_SYMBOL"})
807 repository = ResearchRepository(settings=Settings(research_live_enabled=True, research_search_enabled=False))
808 repository.profiles = [profile]
809 snapshots = []
810 for source, period in (("june", (2026, 6, 30)), ("march", (2026, 3, 31)), ("december", (2025, 12, 31)), ("september", (2025, 9, 30))):
811 snapshot = _legacy_nse_xbrl_snapshot(profile.instrument_id, source, period)
812 snapshot.retrieved_at = datetime(2026, 7, 2, tzinfo=timezone.utc)
813 snapshot.values = [ShareholdingSnapshotValue(
814 category=ShareholdingCategory.PROMOTER, percentage=Decimal("42.5"), source_locator="nse-xbrl:explicit:promoter",
815 )]
816 assert repository.persist_shareholding_snapshot(snapshot)
817 snapshots.append(snapshot)
818
819 class Provider:
820 calls = 0
821 async def discover(self, _profile):
822 self.calls += 1
823 return snapshots
824
825 provider = Provider()
826 repository._official_filing_discovery = _EmptyOfficialFilingDiscovery()
827 repository._official_shareholding_discovery = provider
828 evidence_at = repository.shareholding_for(profile.instrument_id, limit=1)[0].retrieved_at
829
830 await repository._refresh_targeted(profile, set(), now=datetime(2026, 8, 15, tzinfo=timezone.utc))
831 assert provider.calls == 0
832
833 checked_at = datetime(2026, 9, 2, tzinfo=timezone.utc)
834 await repository._refresh_targeted(profile, set(), now=checked_at)
835 assert provider.calls == 1
836 assert repository._category_refresh[(profile.instrument_id, "SHAREHOLDING_PATTERN")] == checked_at
837 assert repository.shareholding_for(profile.instrument_id, limit=1)[0].retrieved_at == evidence_at
838
839 await repository._refresh_targeted(profile, set(), now=checked_at + timedelta(days=1))
840 assert provider.calls == 1
841
842
843 @pytest.mark.asyncio
844 async def test_incomplete_fresh_shareholding_provider_failure_preserves_existing_data_and_non_nse_is_skipped() -> None:
845 profile = CompanyResearchProfile(instrument_id=uuid4(), company_id=uuid4(), company_name="Generic India Equity",
846 ticker="BROKER_ALIAS", exchange="NSE", mic="XNSE", country="IN", currency="INR",
847 provider_instrument_ids={"NSE": "OFFICIAL_SYMBOL"})
848 repository = ResearchRepository(settings=Settings(research_live_enabled=True, research_search_enabled=False))
849 repository.profiles = [profile]
850 for source, period in (("june", (2026, 6, 30)), ("march", (2026, 3, 31)), ("december", (2025, 12, 31))):
851 snapshot = _snapshot(profile.instrument_id, source)
852 snapshot.period_end = datetime(*period, tzinfo=timezone.utc)
853 assert repository.persist_shareholding_snapshot(snapshot) is True
854 class UnavailableProvider:
855 calls = 0
856 async def discover(self, _profile):
857 self.calls += 1
858 raise SearchProviderError("NSE_SHAREHOLDING_OFFICIAL_UNAVAILABLE")
859 provider = UnavailableProvider()
860 repository._official_filing_discovery = _EmptyOfficialFilingDiscovery()
861 repository._official_shareholding_discovery = provider
862 await repository._refresh_targeted(profile, set())
863 assert provider.calls == 1
864 assert len(repository.shareholding_for(profile.instrument_id)) == 3
865 assert repository._category_is_fresh(profile.instrument_id, "SHAREHOLDING_PATTERN", datetime.now(timezone.utc))
866
867 non_nse = profile.model_copy(update={"instrument_id": uuid4(), "exchange": "NYSE", "country": "US", "provider_instrument_ids": {}})
868 repository.profiles.append(non_nse)
869 for source, period in (("us-june", (2026, 6, 30)), ("us-march", (2026, 3, 31)), ("us-december", (2025, 12, 31))):
870 snapshot = _snapshot(non_nse.instrument_id, source)
871 snapshot.period_end = datetime(*period, tzinfo=timezone.utc)
872 assert repository.persist_shareholding_snapshot(snapshot) is True
873 await repository._refresh_targeted(non_nse, set())
874 assert provider.calls == 1
875
876
877 @pytest.mark.asyncio
878 async def test_dedicated_nse_shareholding_empty_or_invalid_rows_remain_retryable_without_snapshot() -> None:
879 profile = CompanyResearchProfile(instrument_id=uuid4(), company_id=uuid4(), company_name="Generic India Equity",
880 ticker="BROKER_ALIAS", exchange="NSE", mic="XNSE", country="IN", currency="INR",
881 provider_instrument_ids={"NSE": "OFFICIAL_SYMBOL"})
882 rows = [{
883 "recordId": "bad", "symbol": "OFFICIAL_SYMBOL", "date": "30-JUN-2026",
884 "xbrl": "https://nsearchives.nseindia.com/corporate/xbrl/SHP_bad.xml", "pr_and_prgrp": "101",
885 }]
886 client = httpx.AsyncClient(transport=httpx.MockTransport(lambda request: httpx.Response(200, json=rows, request=request)))
887 try:
888 assert await OfficialNseShareholdingDiscovery(client).discover(profile) == []
889 finally:
890 await client.aclose()
891
892
893 @pytest.mark.asyncio
894 async def test_persisted_real_official_document_reuses_before_network_after_recreation(tmp_path) -> None:
895 database = tmp_path / "research.db"
896 settings = Settings(research_live_enabled=True, research_persistence_enabled=True,
897 research_database_backend="sqlite", research_database_name=str(database),
898 research_official_document_max_attempts_per_refresh=1)
899 first = ResearchRepository(settings=settings)
900 profile = next(profile for profile in first.list_profiles() if profile.ticker == "RELIANCE")
901 source = RegisteredResearchSource(source_id="nse-shareholding-reuse", instrument_id=profile.instrument_id,
902 url="https://nsearchives.nseindia.com/corporate/shareholding-reuse.pdf",
903 source_type=SourceType.EXCHANGE_ANNOUNCEMENT, source_classification=SourceClassification.EXCHANGE,
904 source_name="NSE corporate announcements", publisher="NSE", reliability_level=ReliabilityLevel.LEVEL_A,
905 company_id=profile.company_id, discovery_method="NSE_OFFICIAL_API", categories=("SHAREHOLDING_PATTERN",))
906 persisted = first.ingest_fixture(original_url=source.url, source_type=source.source_type,
907 source_classification=source.source_classification, source_name=source.source_name, publisher=source.publisher,
908 content_type="text/html", body="<title>Reliance Industries Limited Shareholding Pattern</title><main>Reliance Industries Limited RELIANCE INE002A01018 Shareholding Pattern as on 30/06/2026. Promoter: 42.5%.</main>",
909 reliability=source.reliability_level, source_mode=SourceMode.REAL, expected_profile=profile)
910 assert persisted.status == DocumentStatus.PROCESSED
911
912 class NoNetworkFetcher:
913 calls = 0
914 async def fetch(self, _url):
915 self.calls += 1
916 raise AssertionError("durable reuse must happen before network fetch")
917
918 second = ResearchRepository(settings=settings)
919 fetcher = NoNetworkFetcher()
920 second._fetcher = fetcher
921 restored_profile = second.profile(profile.instrument_id)
922 records: list[logging.LogRecord] = []
923 handler = logging.Handler()
924 handler.emit = records.append
925 logger = logging.getLogger("app.repository")
926 previous_level = logger.level
927 logger.setLevel(logging.INFO)
928 logger.addHandler(handler)
929 try:
930 await second._fetch_official_filings(restored_profile, [DiscoveryResult("SHAREHOLDING_PATTERN", source)], set())
931 finally:
932 logger.removeHandler(handler)
933 logger.setLevel(previous_level)
934 assert fetcher.calls == 0
935 assert len(second.documents_for(profile.instrument_id, source_mode=SourceMode.REAL)) == 1
936 assert second._reusable_official_document(profile.instrument_id, source.url) is not None
937 assert any("outcome=REUSED reason=ALREADY_PERSISTED" in record.getMessage() for record in records)
938
939
940 @pytest.mark.asyncio
941 async def test_usable_historical_parsed_document_reuses_before_network_and_freshens_financial_results(tmp_path) -> None:
942 database = tmp_path / "research.db"
943 settings = Settings(research_live_enabled=True, research_persistence_enabled=True,
944 research_database_backend="sqlite", research_database_name=str(database))
945 seed = ResearchRepository(settings=Settings())
946 profile = next(profile for profile in seed.list_profiles() if profile.ticker == "RELIANCE")
947 source = RegisteredResearchSource(source_id="nse-parsed-reuse", instrument_id=profile.instrument_id,
948 url="https://nsearchives.nseindia.com/corporate/parsed-reuse.pdf",
949 source_type=SourceType.EXCHANGE_ANNOUNCEMENT, source_classification=SourceClassification.EXCHANGE,
950 source_name="NSE corporate announcements", publisher="NSE", reliability_level=ReliabilityLevel.LEVEL_A,
951 company_id=profile.company_id, discovery_method="NSE_OFFICIAL_API", categories=("FINANCIAL_RESULTS",))
952 parsed = ResearchDocument(instrument_id=profile.instrument_id, company_id=profile.company_id,
953 canonical_url=source.url, original_url=source.url, source_type=source.source_type,
954 source_classification=source.source_classification, source_name=source.source_name, publisher=source.publisher,
955 title="Reliance Industries Limited Financial Results", content_type="application/pdf",
956 document_type=DocumentType.PDF_REFERENCE, content_hash="c" * 64, status=DocumentStatus.PARSED,
957 reliability_level=ReliabilityLevel.LEVEL_A, entity_resolution_confidence=0.99, source_mode=SourceMode.REAL,
958 normalized_text="Reliance Industries Limited quarterly financial results revenue 1000 crore PAT 100 crore.")
959 SqliteResearchPersistence(database).upsert_document(parsed)
960
961 class NoNetworkFetcher:
962 calls = 0
963 async def fetch(self, _url):
964 self.calls += 1
965 raise AssertionError("usable PARSED document must be reused before network fetch")
966
967 repository = ResearchRepository(settings=settings)
968 repository._fetcher = NoNetworkFetcher()
969 assert repository._reusable_official_document(profile.instrument_id, source.url) is not None
970 records: list[logging.LogRecord] = []
971 handler = logging.Handler()
972 handler.emit = records.append
973 logger = logging.getLogger("app.repository")
974 previous_level = logger.level
975 logger.setLevel(logging.INFO)
976 logger.addHandler(handler)
977 try:
978 await repository._fetch_official_filings(profile, [DiscoveryResult("FINANCIAL_RESULTS", source)], set())
979 finally:
980 logger.removeHandler(handler)
981 logger.setLevel(previous_level)
982 assert repository._fetcher.calls == 0
983 assert repository._qualifying_category_evidence(profile.instrument_id, "FINANCIAL_RESULTS") is not None
984 assert any("outcome=REUSED reason=ALREADY_PERSISTED" in record.getMessage() for record in records)
985
986
987 def test_empty_parsed_document_is_not_reusable_or_financial_result_evidence() -> None:
988 repository = ResearchRepository(settings=Settings(research_live_enabled=True))
989 profile = next(profile for profile in repository.list_profiles() if profile.ticker == "RELIANCE")
990 document = ResearchDocument(instrument_id=profile.instrument_id, company_id=profile.company_id,
991 canonical_url="https://nsearchives.nseindia.com/corporate/empty-parsed.pdf",
992 original_url="https://nsearchives.nseindia.com/corporate/empty-parsed.pdf", source_type=SourceType.EXCHANGE_ANNOUNCEMENT,
993 source_classification=SourceClassification.EXCHANGE, source_name="NSE", content_type="application/pdf",
994 document_type=DocumentType.PDF_REFERENCE, content_hash="d" * 64, status=DocumentStatus.PARSED,
995 reliability_level=ReliabilityLevel.LEVEL_A, entity_resolution_confidence=0.99, source_mode=SourceMode.REAL,
996 normalized_text="")
997 repository.documents[document.document_id] = document
998 assert repository._reusable_official_document(profile.instrument_id, document.canonical_url) is None
999 assert repository._qualifying_category_evidence(profile.instrument_id, "FINANCIAL_RESULTS") is None
1000
1001
1002 @pytest.mark.asyncio
1003 async def test_official_fetch_round_robin_prevents_category_starvation_with_bounded_budget() -> None:
1004 settings = Settings(research_live_enabled=True, research_official_document_max_attempts_per_refresh=3)
1005 repository = ResearchRepository(settings=settings)
1006 profile = next(profile for profile in repository.list_profiles() if profile.ticker == "RELIANCE")
1007
1008 def source(category: str, suffix: str) -> RegisteredResearchSource:
1009 return RegisteredResearchSource(source_id=f"nse-{category}-{suffix}", instrument_id=profile.instrument_id,
1010 url=f"https://nsearchives.nseindia.com/corporate/{suffix}.pdf", source_type=SourceType.EXCHANGE_ANNOUNCEMENT,
1011 source_classification=SourceClassification.EXCHANGE, source_name="NSE", publisher="NSE",
1012 reliability_level=ReliabilityLevel.LEVEL_A, company_id=profile.company_id, discovery_method="NSE_OFFICIAL_API",
1013 categories=(category,))
1014
1015 financial = [source("FINANCIAL_RESULTS", f"financial-{index}") for index in range(3)]
1016 shareholding = [source("SHAREHOLDING_PATTERN", f"shareholding-{index}") for index in range(2)]
1017
1018 class FailingFetcher:
1019 def __init__(self) -> None:
1020 self.urls: list[str] = []
1021 async def fetch(self, url):
1022 self.urls.append(url)
1023 raise FetchError("fixture failure")
1024
1025 fetcher = FailingFetcher()
1026 repository._fetcher = fetcher
1027 await repository._fetch_official_filings(profile,
1028 [*(DiscoveryResult("FINANCIAL_RESULTS", item) for item in financial), *(DiscoveryResult("SHAREHOLDING_PATTERN", item) for item in shareholding)], set())
1029 assert fetcher.urls == [financial[0].url, shareholding[0].url, financial[1].url]
1030 assert len(fetcher.urls) == settings.research_official_document_max_attempts_per_refresh
1031
1032
1033 def test_official_filing_scheduler_preserves_newest_first_within_each_category() -> None:
1034 profile = ResearchRepository(settings=Settings()).list_profiles()[0]
1035 def result(category: str, suffix: str) -> DiscoveryResult:
1036 source = RegisteredResearchSource(source_id=f"{category}-{suffix}", instrument_id=profile.instrument_id,
1037 url=f"https://nsearchives.nseindia.com/corporate/{suffix}.pdf", source_type=SourceType.EXCHANGE_ANNOUNCEMENT,
1038 source_classification=SourceClassification.EXCHANGE, source_name="NSE", publisher="NSE",
1039 reliability_level=ReliabilityLevel.LEVEL_A, company_id=profile.company_id, categories=(category,))
1040 return DiscoveryResult(category, source)
1041 financial_new, financial_old = result("FINANCIAL_RESULTS", "financial-new"), result("FINANCIAL_RESULTS", "financial-old")
1042 share_new, share_old = result("SHAREHOLDING_PATTERN", "share-new"), result("SHAREHOLDING_PATTERN", "share-old")
1043 scheduled = _fair_official_filing_order([financial_new, financial_old, share_new, share_old])
1044 urls = [item.source.url for item in scheduled]
1045 assert urls == [financial_new.source.url, share_new.source.url, financial_old.source.url, share_old.source.url]
1046
1047
1048 @pytest.mark.asyncio
1049 async def test_reusable_official_documents_do_not_consume_category_fair_network_budget() -> None:
1050 settings = Settings(research_live_enabled=True, research_official_document_max_attempts_per_refresh=3)
1051 repository = ResearchRepository(settings=settings)
1052 profile = next(profile for profile in repository.list_profiles() if profile.ticker == "RELIANCE")
1053 def source(category: str, suffix: str) -> RegisteredResearchSource:
1054 return RegisteredResearchSource(source_id=f"nse-{category}-{suffix}", instrument_id=profile.instrument_id,
1055 url=f"https://nsearchives.nseindia.com/corporate/{suffix}.pdf", source_type=SourceType.EXCHANGE_ANNOUNCEMENT,
1056 source_classification=SourceClassification.EXCHANGE, source_name="NSE", publisher="NSE",
1057 reliability_level=ReliabilityLevel.LEVEL_A, company_id=profile.company_id, discovery_method="NSE_OFFICIAL_API",
1058 categories=(category,))
1059 financial = [source("FINANCIAL_RESULTS", f"reuse-financial-{index}") for index in range(3)]
1060 shareholding = [source("SHAREHOLDING_PATTERN", f"reuse-shareholding-{index}") for index in range(2)]
1061 for index in (0, 1):
1062 document = ResearchDocument(instrument_id=profile.instrument_id, company_id=profile.company_id,
1063 canonical_url=financial[index].url, original_url=financial[index].url, source_type=SourceType.EXCHANGE_ANNOUNCEMENT,
1064 source_classification=SourceClassification.EXCHANGE, source_name="NSE", content_type="application/pdf",
1065 document_type=DocumentType.PDF_REFERENCE, content_hash=(str(index + 7) * 64), status=DocumentStatus.PROCESSED,
1066 reliability_level=ReliabilityLevel.LEVEL_A, entity_resolution_confidence=0.99, source_mode=SourceMode.REAL)
1067 repository.documents[document.document_id] = document
1068
1069 class FailingFetcher:
1070 def __init__(self) -> None:
1071 self.urls: list[str] = []
1072 async def fetch(self, url):
1073 self.urls.append(url)
1074 raise FetchError("fixture failure")
1075
1076 fetcher = FailingFetcher()
1077 repository._fetcher = fetcher
1078 await repository._fetch_official_filings(profile,
1079 [*(DiscoveryResult("FINANCIAL_RESULTS", item) for item in financial), *(DiscoveryResult("SHAREHOLDING_PATTERN", item) for item in shareholding)], set())
1080 assert fetcher.urls == [shareholding[0].url, shareholding[1].url, financial[2].url]
1081 assert len(fetcher.urls) == settings.research_official_document_max_attempts_per_refresh
1082
1083
1084 @pytest.mark.asyncio
1085 async def test_official_discovery_requires_verified_nse_mapping() -> None:
1086 profile = CompanyResearchProfile(instrument_id=uuid4(), company_id=uuid4(), company_name="Generic India Equity",
1087 ticker="BROKER_ALIAS", exchange="NSE", mic="XNSE", country="IN", currency="INR")
1088 discovery = OfficialFilingDiscovery()
1089 assert await discovery.discover(profile, {"SHAREHOLDING_PATTERN"}, set()) == []
1090
1091
1092 def test_broker_derived_nse_alias_is_not_hydrated_as_trusted_identity() -> None:
1093 instrument_id = uuid4()
1094 instrument = _global_master_instrument({
1095 "canonicalName": "Generic India Equity", "assetType": "EQUITY", "country": "IN", "currency": "INR",
1096 "primarySymbol": "BROKER_ALIAS", "primaryExchange": "NSE",
1097 "providerMappings": [{"provider": "NSE", "providerSymbol": "BROKER_ALIAS", "status": "VERIFIED",
1098 "resolutionSource": "BROKER_IMPORT_IDENTITY"}],
1099 }, instrument_id)
1100 assert "nseSymbol" not in instrument
1101
1102
1103 def test_durable_snapshot_freshness_survives_repository_recreation(tmp_path) -> None:
1104 database = tmp_path / "research.db"
1105 settings = Settings(research_persistence_enabled=True, research_database_backend="sqlite", research_database_name=str(database))
1106 first = ResearchRepository(settings=settings)
1107 profile = first.list_profiles()[0]
1108 assert first.persist_shareholding_snapshot(_snapshot(profile.instrument_id)) is True
1109 second = ResearchRepository(settings=settings)
1110 assert second.shareholding_for(profile.instrument_id)
1111 assert second._qualifying_category_evidence(profile.instrument_id, "SHAREHOLDING_PATTERN") is not None
1112 assert second._category_is_fresh(profile.instrument_id, "SHAREHOLDING_PATTERN", datetime.now(timezone.utc))
1113
1114
1115 def test_summary_returns_latest_four_global_periods_with_provenance() -> None:
1116 repo = ResearchRepository(settings=Settings())
1117 profile = repo.list_profiles()[0]
1118 for index, period_end in enumerate((
1119 datetime(2026, 3, 31, tzinfo=timezone.utc),
1120 datetime(2026, 6, 30, tzinfo=timezone.utc),
1121 datetime(2026, 9, 30, tzinfo=timezone.utc),
1122 datetime(2026, 12, 31, tzinfo=timezone.utc),
1123 datetime(2027, 3, 31, tzinfo=timezone.utc),
1124 ), start=1):
1125 snapshot = _snapshot(profile.instrument_id, f"nse-{index}")
1126 snapshot.period_end = period_end
1127 repo.persist_shareholding_snapshot(snapshot)
1128 snapshots = repo.summary(profile.instrument_id).shareholding_snapshots
1129 assert len(snapshots) == 4
1130 assert all(snapshot.source_url and snapshot.values for snapshot in snapshots)