| 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) |