| 1 | """Normalization of explicitly labelled Indian shareholding filing values. |
| 2 | |
| 3 | This deliberately accepts only source labels adjacent to a percentage; it does not |
| 4 | invent categories, residual public holdings, or a pledge basis. |
| 5 | """ |
| 6 | from __future__ import annotations |
| 7 | |
| 8 | import re |
| 9 | import xml.etree.ElementTree as ET |
| 10 | from datetime import datetime, timezone |
| 11 | from decimal import Decimal, InvalidOperation |
| 12 | |
| 13 | from app.models import ( |
| 14 | DocumentStatus, ResearchDocument, ShareholdingCategory, ShareholdingSnapshot, |
| 15 | ShareholdingSnapshotValue, SourceClassification, SourceMode, |
| 16 | ) |
| 17 | |
| 18 | _PERIOD = re.compile(r"(?:as\s+on|quarter\s+ended|ended)\s*[:\-]?\s*(\d{1,2}[/-]\d{1,2}[/-]\d{4}|\d{1,2}\s+[A-Za-z]+\s+\d{4})", re.I) |
| 19 | _VALUE = re.compile(r"(?P<label>[A-Za-z/&() .-]{3,80})\s*[:\-]?\s*(?P<value>\d{1,3}(?:\.\d{1,4})?)\s*%", re.I) |
| 20 | _LABELS: tuple[tuple[ShareholdingCategory, tuple[str, ...]], ...] = ( |
| 21 | (ShareholdingCategory.PROMOTER, ("promoter and promoter group", "promoter holding", "promoter")), |
| 22 | (ShareholdingCategory.FII_FPI, ("foreign institutional investors", "foreign portfolio investors", "fii/fpi", "fii", "fpi")), |
| 23 | (ShareholdingCategory.DII, ("domestic institutional investors", "dii")), |
| 24 | (ShareholdingCategory.MUTUAL_FUNDS, ("mutual funds",)), |
| 25 | (ShareholdingCategory.INSURANCE, ("insurance companies", "insurance")), |
| 26 | (ShareholdingCategory.GOVERNMENT, ("government",)), |
| 27 | (ShareholdingCategory.PUBLIC_RETAIL, ("public shareholders", "public/retail", "public holding", "retail")), |
| 28 | (ShareholdingCategory.OTHERS, ("others", "other non-promoter")), |
| 29 | ) |
| 30 | |
| 31 | |
| 32 | def parse_nse_shareholding_xbrl(xml_text: str) -> list[ShareholdingSnapshotValue]: |
| 33 | """Normalize only explicit NSE shareholding-taxonomy facts. |
| 34 | |
| 35 | The SHP taxonomy reports ownership ratios as fractions (for example, |
| 36 | ``0.494`` for 49.4%). Context members identify the source category; no |
| 37 | aggregate public-shareholding fact is used as a retail proxy. |
| 38 | """ |
| 39 | try: |
| 40 | root = ET.fromstring(xml_text) |
| 41 | except ET.ParseError: |
| 42 | return [] |
| 43 | contexts: dict[str, str] = {} |
| 44 | for context in root.iter(): |
| 45 | if _xml_local_name(context.tag) != "context": |
| 46 | continue |
| 47 | member = next((node.text or "" for node in context.iter() if _xml_local_name(node.tag) == "explicitMember"), "") |
| 48 | contexts[context.attrib.get("id", "")] = member.rsplit(":", 1)[-1] |
| 49 | |
| 50 | facts: dict[tuple[str, str], str] = {} |
| 51 | for node in root.iter(): |
| 52 | context = node.attrib.get("contextRef") |
| 53 | if context and node.text is not None: |
| 54 | facts[(contexts.get(context, ""), _xml_local_name(node.tag))] = node.text.strip() |
| 55 | |
| 56 | values: list[ShareholdingSnapshotValue] = [] |
| 57 | direct = ( |
| 58 | ("ShareholdingOfPromoterAndPromoterGroupMember", "PROMOTER", ShareholdingCategory.PROMOTER), |
| 59 | ("InstitutionsDomesticMember", "Institutions Domestic", ShareholdingCategory.DII), |
| 60 | ("MutualFundsOrUTIMember", "Mutual Funds or UTI", ShareholdingCategory.MUTUAL_FUNDS), |
| 61 | ("InsuranceCompaniesMember", "Insurance Companies", ShareholdingCategory.INSURANCE), |
| 62 | ("GovernmentsMember", "Governments", ShareholdingCategory.GOVERNMENT), |
| 63 | ("ResidentIndividualShareholdersHoldingNominalShareCapitalUpToRsTwoLakhMember", "Resident Individual Shareholders Holding Nominal Share Capital Up To Rs Two Lakh", ShareholdingCategory.PUBLIC_RETAIL), |
| 64 | ("OtherNonInstitutionsMember", "Other Non-Institutions", ShareholdingCategory.OTHERS), |
| 65 | ) |
| 66 | for member, label, category in direct: |
| 67 | value = _xbrl_ratio_percentage(facts.get((member, "ShareholdingAsAPercentageOfTotalNumberOfShares"))) |
| 68 | if value is not None: |
| 69 | values.append(_xbrl_value(category, label, member, "ShareholdingAsAPercentageOfTotalNumberOfShares", value)) |
| 70 | |
| 71 | fpi_members = ( |
| 72 | "InstitutionsForeignPortfolioInvestorCategoryOneMember", |
| 73 | "InstitutionsForeignPortfolioInvestorCategoryTwoMember", |
| 74 | ) |
| 75 | fpi_values = [_xbrl_ratio_percentage(facts.get((member, "ShareholdingAsAPercentageOfTotalNumberOfShares"))) for member in fpi_members] |
| 76 | if all(value is not None for value in fpi_values): |
| 77 | values.append(ShareholdingSnapshotValue( |
| 78 | category=ShareholdingCategory.FII_FPI, percentage=sum(fpi_values, Decimal("0")), |
| 79 | metric_basis="DERIVED_SUM_OF_MUTUALLY_EXCLUSIVE_FPI_CATEGORIES", |
| 80 | raw_source_label="Foreign Portfolio Investor Category I + Category II", |
| 81 | source_locator="nse-xbrl:context=" + "+".join(fpi_members) + ";tag=ShareholdingAsAPercentageOfTotalNumberOfShares", |
| 82 | evidence_text=f"NSE XBRL FPI Category I + II: {fpi_values[0]}% + {fpi_values[1]}%", |
| 83 | )) |
| 84 | |
| 85 | pledge = _xbrl_ratio_percentage(facts.get(("ShareholdingOfPromoterAndPromoterGroupMember", "EncumberedShareUnderPledgedAsPercentageOfTotalNumberOfShares"))) |
| 86 | if pledge is not None: |
| 87 | values.append(ShareholdingSnapshotValue( |
| 88 | category=ShareholdingCategory.PROMOTER_PLEDGE, percentage=pledge, |
| 89 | metric_basis="PERCENT_OF_PROMOTER_HOLDING", |
| 90 | raw_source_label="Encumbered Share Under Pledged (Promoter and Promoter Group)", |
| 91 | source_locator="nse-xbrl:context=ShareholdingOfPromoterAndPromoterGroupMember;tag=EncumberedShareUnderPledgedAsPercentageOfTotalNumberOfShares", |
| 92 | evidence_text=f"NSE XBRL promoter-group pledged-share ratio: {pledge}% of promoter holding", |
| 93 | )) |
| 94 | return values |
| 95 | |
| 96 | |
| 97 | def _xml_local_name(tag: str) -> str: |
| 98 | return tag.rsplit("}", 1)[-1] |
| 99 | |
| 100 | |
| 101 | def _xbrl_ratio_percentage(value: str | None) -> Decimal | None: |
| 102 | if value is None: |
| 103 | return None |
| 104 | try: |
| 105 | percentage = Decimal(value) * Decimal("100") |
| 106 | except InvalidOperation: |
| 107 | return None |
| 108 | return percentage if Decimal("0") <= percentage <= Decimal("100") else None |
| 109 | |
| 110 | |
| 111 | def _xbrl_value(category: ShareholdingCategory, label: str, member: str, tag: str, percentage: Decimal) -> ShareholdingSnapshotValue: |
| 112 | return ShareholdingSnapshotValue( |
| 113 | category=category, percentage=percentage, raw_source_label=label, |
| 114 | source_locator=f"nse-xbrl:context={member};tag={tag}", |
| 115 | evidence_text=f"NSE XBRL {label}: {percentage}%", |
| 116 | ) |
| 117 | |
| 118 | |
| 119 | def parse_official_shareholding(document: ResearchDocument) -> ShareholdingSnapshot | None: |
| 120 | if document.source_mode != SourceMode.REAL or document.source_classification not in { |
| 121 | SourceClassification.EXCHANGE, SourceClassification.REGULATORY, SourceClassification.OFFICIAL_COMPANY |
| 122 | } or document.status not in {DocumentStatus.PARSED, DocumentStatus.PROCESSED} or not document.instrument_id: |
| 123 | return None |
| 124 | text = document.normalized_text or document.raw_text or "" |
| 125 | period_match = _PERIOD.search(text) |
| 126 | if not period_match: |
| 127 | return None |
| 128 | period_end = _parse_period(period_match.group(1)) |
| 129 | if period_end is None: |
| 130 | return None |
| 131 | values: list[ShareholdingSnapshotValue] = [] |
| 132 | seen: set[ShareholdingCategory] = set() |
| 133 | for match in _VALUE.finditer(text): |
| 134 | label = " ".join(match.group("label").lower().split()) |
| 135 | category = _category(label) |
| 136 | if category is None or category in seen: |
| 137 | continue |
| 138 | percentage = _percentage(match.group("value")) |
| 139 | if percentage is None: |
| 140 | continue |
| 141 | values.append(ShareholdingSnapshotValue(category=category, percentage=percentage, |
| 142 | raw_source_label=match.group("label").strip(), source_locator=f"text:{match.start()}", |
| 143 | evidence_text=match.group(0).strip())) |
| 144 | seen.add(category) |
| 145 | pledge = _pledge_value(text) |
| 146 | if pledge is not None: |
| 147 | values.append(pledge) |
| 148 | if not values: |
| 149 | return None |
| 150 | return ShareholdingSnapshot( |
| 151 | instrument_id=document.instrument_id, period_end=period_end, filing_basis=_filing_basis(text), |
| 152 | source_provider="NSE" if document.source_classification == SourceClassification.EXCHANGE else document.source_name, |
| 153 | source_type=str(document.source_type), source_identity_key=str(document.document_id), source_url=document.canonical_url, |
| 154 | research_document_id=document.document_id, published_at=document.published_at, retrieved_at=document.retrieved_at, |
| 155 | confidence=Decimal("0.90"), reliability_level=document.reliability_level, source_mode=document.source_mode, |
| 156 | values=values, |
| 157 | ) |
| 158 | |
| 159 | |
| 160 | def _category(label: str) -> ShareholdingCategory | None: |
| 161 | for category, labels in _LABELS: |
| 162 | if any(candidate in label for candidate in labels): |
| 163 | return category |
| 164 | return None |
| 165 | |
| 166 | |
| 167 | def _pledge_value(text: str) -> ShareholdingSnapshotValue | None: |
| 168 | match = re.search(r"(?P<label>promoter(?:s)?[^.\n]{0,70}?(?:pledged|encumbered)[^.\n]{0,70})\s*[:\-]?\s*(?P<value>\d{1,3}(?:\.\d{1,4})?)\s*%", text, re.I) |
| 169 | if not match: |
| 170 | return None |
| 171 | lowered = match.group("label").lower() |
| 172 | basis = "PERCENT_OF_PROMOTER_HOLDING" if "promoter holding" in lowered else ( |
| 173 | "PERCENT_OF_TOTAL_SHARES" if "total shares" in lowered else None |
| 174 | ) |
| 175 | percentage = _percentage(match.group("value")) |
| 176 | if basis is None or percentage is None: |
| 177 | return None |
| 178 | return ShareholdingSnapshotValue(category=ShareholdingCategory.PROMOTER_PLEDGE, percentage=percentage, |
| 179 | metric_basis=basis, raw_source_label=match.group("label").strip(), source_locator=f"text:{match.start()}", |
| 180 | evidence_text=match.group(0).strip()) |
| 181 | |
| 182 | |
| 183 | def _percentage(value: str) -> Decimal | None: |
| 184 | try: |
| 185 | parsed = Decimal(value) |
| 186 | except InvalidOperation: |
| 187 | return None |
| 188 | return parsed if Decimal("0") <= parsed <= Decimal("100") else None |
| 189 | |
| 190 | |
| 191 | def _parse_period(value: str) -> datetime | None: |
| 192 | for fmt in ("%d/%m/%Y", "%d-%m-%Y", "%d %B %Y", "%d %b %Y"): |
| 193 | try: |
| 194 | return datetime.strptime(value, fmt).replace(tzinfo=timezone.utc) |
| 195 | except ValueError: |
| 196 | continue |
| 197 | return None |
| 198 | |
| 199 | |
| 200 | def _filing_basis(text: str) -> str | None: |
| 201 | lowered = text.lower() |
| 202 | if "consolidated" in lowered: |
| 203 | return "CONSOLIDATED" |
| 204 | if "standalone" in lowered: |
| 205 | return "STANDALONE" |
| 206 | return None |