| 1 | from __future__ import annotations |
| 2 | |
| 3 | import logging |
| 4 | import re |
| 5 | import time |
| 6 | from bisect import bisect_left |
| 7 | from datetime import date, datetime, timezone |
| 8 | from decimal import Decimal, InvalidOperation |
| 9 | from enum import Enum |
| 10 | from dataclasses import dataclass, field |
| 11 | from urllib.parse import urlparse |
| 12 | |
| 13 | from app.models import ( |
| 14 | PortfolioResearchCompany, |
| 15 | FinancialStatementPeriod, |
| 16 | FinancialResultPeriod, |
| 17 | ProvenancedValue, |
| 18 | QuarterlyResult, |
| 19 | ResearchDocument, |
| 20 | ResearchEvent, |
| 21 | ResearchEventType, |
| 22 | ShareholdingChange, |
| 23 | ShareholdingSnapshot, |
| 24 | ShareholdingSnapshotValue, |
| 25 | SourceClassification, |
| 26 | SourceDiversity, |
| 27 | SourceMode, |
| 28 | ValuationAssessment, |
| 29 | ValuationBenchmark, |
| 30 | ValuationStateEvidence, |
| 31 | ) |
| 32 | from app.fact_precedence import FinancialFact, FactSourceTier, SUPPORTED_FINANCIAL_SOURCE_TIERS, fact_source_authority |
| 33 | |
| 34 | |
| 35 | logger = logging.getLogger(__name__) |
| 36 | |
| 37 | |
| 38 | PERIOD_RE = re.compile(r"\b(Q[1-4]\s*(?:FY)?\s*\d{2,4}|(?:quarter|three months) ended\s+(?:\d{1,2}\s+)?[A-Za-z]+(?:\s+\d{1,2})?,?\s+\d{4})\b", re.I) |
| 39 | NUMBER = r"([-+]?\d[\d,]*(?:\.\d+)?)" |
| 40 | _NSE_QUARTER_HEADING = re.compile(r"\b(?:quarter|quaiter|three\s+months)(?:\s+and\s+year)?\s+ended\b", re.I) |
| 41 | _NSE_DATE = re.compile(r"\b(\d{1,2})(?:[lI]?st|nd|rd|th|tli)?\s*[\"'.,-]*\s*([A-Za-z\s]{3,12}?)\s+([2Z][0-9Z]{3})\b", re.I) |
| 42 | _NSE_MONTH_DAY_DATE = re.compile(r"\b([A-Za-z]{3,12})\s+(\d{1,2})(?:[lI]?st|nd|rd|th|tli)?\s*,?\s+([2Z][0-9Z]{3})\b", re.I) |
| 43 | _NSE_NUMERIC_DATE = re.compile(r"\b(\d{1,2})\s*[./-]\s*(\d{1,2})\s*[./-]\s*([2Z][0-9Z]{3})\b") |
| 44 | _NSE_DAY_MONTH = re.compile(r"\b(\d{1,2})(?:[lI]?st|nd|rd|th|tli)?\s*[\"'.,-]*\s*([A-Za-z\s]{3,12}?)(?=\s+\d|\s+Z?\d{3}|\s+\d{1,2}(?:[lI]?st|nd|rd|th|tli)?|\s*$)", re.I) |
| 45 | _MONTHS = {name: index for index, name in enumerate(("january", "february", "march", "april", "may", "june", "july", "august", "september", "october", "november", "december"), 1)} |
| 46 | _STATEMENT_HEADING = re.compile(r"\b(?:extract\s+of\s+)?(?:stateme[no]t\s+of\s+)?(?:(?:(?:unaud(?:\.?ited|[il]ted)|umaudtoed|aud[il]ted|standalone|consolidated)\s+){0,2})(?:f[il]nanc[il]al|fl\s+nclal)\s+results\b|\bquarterly\s+financial\s+results\b", re.I) |
| 47 | _STATEMENT_END = re.compile(r"\bstatement\s+of\s+(?:assets|financial\s+position|cash\s+flows?)\b|\bcash\s+flow\s+statement\b", re.I) |
| 48 | _GROUP_RE = re.compile(r"\b(quarter|nine\s+months?|half\s+year|year)\s+ended\b", re.I) |
| 49 | _PARTICULARS_RE = re.compile(r"\b(?:particulars|partlculars|parriculars|p8mculars)\b", re.I) |
| 50 | # Some PDF text extractors place the column groups immediately before the |
| 51 | # serial-number and ``Particulars`` cells, while leaving the date cells after |
| 52 | # them. Match only the contiguous table-header suffix; this cannot pull an |
| 53 | # earlier narrative reference to a reporting period into the candidate. |
| 54 | _PRE_PARTICULARS_GROUPS_RE = re.compile( |
| 55 | r"(?:(?:\b(?:quarter|nine\s+months?|half\s+year|year)\s+ended\b)\s*)+(?:(?:s[il]|no\.?)\s*)?$", |
| 56 | re.I, |
| 57 | ) |
| 58 | _ROW_REFERENCE_FORMULA_RE = re.compile( |
| 59 | r"^\s*(?:(?:\(\s*\d+\s*\)(?:\s*[+\-*/•]\s*\(\s*\d+\s*\))+)|(?:\(\s*\d+(?:\s*[+\-*/•]\s*\d+)+\s*\)))\s*" |
| 60 | ) |
| 61 | _BALANCE_SHEET_HEADING = re.compile(r"\b(?:balance\s+sheet|statement\s+of\s+(?:financial\s+position|assets\s+and\s+liabilities))\b", re.I) |
| 62 | _CASH_FLOW_HEADING = re.compile(r"\b(?:cash\s+fl\s*ows?\s+statement|statement\s+(?:of|for\s+the)\s+cash\s+fl\s*ows?)\b", re.I) |
| 63 | |
| 64 | |
| 65 | @dataclass(frozen=True) |
| 66 | class FinancialColumn: |
| 67 | index: int |
| 68 | period_end: str |
| 69 | period_type: str |
| 70 | header_group: str |
| 71 | |
| 72 | |
| 73 | @dataclass(frozen=True) |
| 74 | class FinancialStatement: |
| 75 | region: str |
| 76 | columns: tuple[FinancialColumn, ...] |
| 77 | current_column: FinancialColumn |
| 78 | unit_context: str |
| 79 | |
| 80 | |
| 81 | @dataclass(frozen=True) |
| 82 | class FinancialRow: |
| 83 | normalized_label: str |
| 84 | numeric_cells: tuple[Decimal, ...] |
| 85 | |
| 86 | |
| 87 | @dataclass(frozen=True) |
| 88 | class ParsedIncomeStatementPeriod: |
| 89 | """One explicitly reported NSE income-statement column. |
| 90 | |
| 91 | This deliberately stays smaller than ``QuarterlyResult``: it is the |
| 92 | normalized persistence read model for the G1 income-statement metrics. |
| 93 | """ |
| 94 | period_end: str |
| 95 | period_type: str |
| 96 | reporting_basis: str | None |
| 97 | metrics: tuple[tuple[str, ProvenancedValue], ...] |
| 98 | |
| 99 | |
| 100 | @dataclass(frozen=True) |
| 101 | class ParsedBalanceSheetPeriod: |
| 102 | period_end: str |
| 103 | period_type: str |
| 104 | reporting_basis: str | None |
| 105 | metrics: tuple[tuple[str, ProvenancedValue], ...] |
| 106 | |
| 107 | |
| 108 | @dataclass(frozen=True) |
| 109 | class ParsedCashFlowPeriod: |
| 110 | period_end: str |
| 111 | period_type: str |
| 112 | reporting_basis: str | None |
| 113 | metrics: tuple[tuple[str, ProvenancedValue], ...] |
| 114 | |
| 115 | |
| 116 | @dataclass(frozen=True) |
| 117 | class HeaderToken: |
| 118 | kind: str |
| 119 | value: str |
| 120 | position: int |
| 121 | |
| 122 | |
| 123 | def enrich_company_research( |
| 124 | company: PortfolioResearchCompany, |
| 125 | documents: list[ResearchDocument], |
| 126 | events: list[ResearchEvent], |
| 127 | ownership_threshold: Decimal, |
| 128 | financial_facts: list[FinancialFact] | None = None, |
| 129 | ) -> PortfolioResearchCompany: |
| 130 | company.source_diversity = source_diversity(documents) |
| 131 | company.financial_result_history = financial_result_history_from_facts(financial_facts or []) |
| 132 | company.balance_sheet_history = financial_statement_history_from_facts( |
| 133 | financial_facts or [], period_type={"AS_AT", "QUARTERLY", "ANNUAL"}, metrics={ |
| 134 | "total_assets", "total_liabilities", "total_equity", "equity", |
| 135 | "cash_and_cash_equivalents", "cash_and_equivalents", "total_debt", |
| 136 | "debt_or_borrowings", "current_assets", "current_liabilities", |
| 137 | }, |
| 138 | ) |
| 139 | company.cash_flow_history = financial_statement_history_from_facts( |
| 140 | financial_facts or [], period_type={"QUARTERLY", "ANNUAL"}, metrics={ |
| 141 | "operating_cash_flow", "investing_cash_flow", "financing_cash_flow", |
| 142 | "cash_flow_from_operating_activities", "cash_flow_from_investing_activities", |
| 143 | "cash_flow_from_financing_activities", |
| 144 | }, |
| 145 | ) |
| 146 | company.latest_quarterly_result = latest_quarterly_result_from_facts(financial_facts or []) |
| 147 | if company.latest_quarterly_result is None: |
| 148 | company.latest_quarterly_result = latest_quarterly_result(documents) |
| 149 | if company.latest_quarterly_result and financial_facts: |
| 150 | _apply_normalized_official_facts(company.latest_quarterly_result, financial_facts) |
| 151 | company.quarterly_result_status = "EXTRACTED" if company.latest_quarterly_result else ( |
| 152 | "PDF_SCANNED_OCR_REQUIRED" if any(_is_failed_pdf_reference(document) for document in documents) |
| 153 | or "PDF_SCANNED_OCR_REQUIRED" in str(company.safe_error_code or "") |
| 154 | or "PDF_SCANNED_OCR_REQUIRED" in str(company.safe_error_message or "") else "NOT_AVAILABLE" |
| 155 | ) |
| 156 | # Persisted NSE/XBRL snapshots retain the actual distinct reporting |
| 157 | # periods. Prefer them over the legacy document-text parser, whose two |
| 158 | # period matches can originate from the same filing. |
| 159 | company.shareholding_changes = ( |
| 160 | shareholding_changes_from_snapshots(company.shareholding_snapshots) |
| 161 | or shareholding_changes(documents) |
| 162 | ) |
| 163 | company.ownership_increases = sorted({ |
| 164 | change.category |
| 165 | for change in company.shareholding_changes |
| 166 | if change.category in {"PROMOTER", "FII_FPI", "DII"} |
| 167 | and change.change_percentage_points >= ownership_threshold |
| 168 | }) |
| 169 | company.valuation = valuation_assessment(documents) |
| 170 | company.current_quarter_catalysts = current_quarter_catalysts(events) |
| 171 | return company |
| 172 | |
| 173 | |
| 174 | def latest_quarterly_result_from_facts(facts: list[FinancialFact]) -> QuarterlyResult | None: |
| 175 | """Build the newest usable quarterly result from persisted facts. |
| 176 | |
| 177 | The financial period is selected before field completeness. A result may |
| 178 | therefore be partial, but all populated fields come from one reporting |
| 179 | basis: an explicit basis never silently absorbs an ``UNKNOWN``-basis fact. |
| 180 | """ |
| 181 | history = financial_result_history_from_facts(facts, period_type="QUARTERLY") |
| 182 | if not history: |
| 183 | return None |
| 184 | result = history[0] |
| 185 | return QuarterlyResult( |
| 186 | period=result.period, reporting_basis=result.reporting_basis, |
| 187 | result_date=result.published_at or result.retrieved_at, |
| 188 | revenue=result.revenue, pat=result.pat, eps=result.eps, |
| 189 | source_name=result.source_name, source_url=result.source_url, source_type=result.source_type, |
| 190 | published_at=result.published_at, retrieved_at=result.retrieved_at, confidence=result.confidence, |
| 191 | ) |
| 192 | |
| 193 | |
| 194 | def financial_result_history_from_facts( |
| 195 | facts: list[FinancialFact], *, period_type: str | None = None, |
| 196 | ) -> list[FinancialResultPeriod]: |
| 197 | """Return up to four newest explicit periods per selected compatible basis. |
| 198 | |
| 199 | The newest period chooses the same authoritative basis as the legacy latest |
| 200 | selector; subsequent rows are limited to that basis, so no series can be |
| 201 | manufactured by composing consolidated, standalone, or unknown facts. |
| 202 | """ |
| 203 | allowed_types = {period_type} if period_type else {"QUARTERLY", "ANNUAL"} |
| 204 | grouped: dict[str, dict[tuple[str, str | None], dict[str, FinancialFact]]] = {} |
| 205 | for fact in facts: |
| 206 | if ( |
| 207 | fact.source_mode != SourceMode.REAL |
| 208 | or fact.source_tier not in SUPPORTED_FINANCIAL_SOURCE_TIERS |
| 209 | or fact.key.period_type not in allowed_types |
| 210 | or not fact.key.period_end |
| 211 | or fact.key.metric not in {"revenue", "operating_income", "ebit", "ebitda", "pat", "eps"} |
| 212 | ): |
| 213 | continue |
| 214 | values = grouped.setdefault(fact.key.period_type, {}).setdefault((fact.key.period_end, fact.key.reporting_basis), {}) |
| 215 | existing = values.get(fact.key.metric) |
| 216 | if existing is None or fact_source_authority(fact.source_tier) > fact_source_authority(existing.source_tier): |
| 217 | values[fact.key.metric] = fact |
| 218 | def basis_rank(item: tuple[str | None, dict[str, FinancialFact]]) -> tuple[int, int, int]: |
| 219 | reporting_basis, values = item |
| 220 | # Source authority remains meaningful when the same latest period has |
| 221 | # multiple independently reported bases. Explicit source bases then |
| 222 | # win over UNKNOWN; CONSOLIDATED retains the existing deterministic |
| 223 | # preference when authority is otherwise equal. |
| 224 | return ( |
| 225 | max(fact_source_authority(fact.source_tier) for fact in values.values()), |
| 226 | int(reporting_basis not in {None, "", "UNKNOWN"}), |
| 227 | int(reporting_basis == "CONSOLIDATED"), |
| 228 | ) |
| 229 | |
| 230 | result: list[FinancialResultPeriod] = [] |
| 231 | for kind, groups in grouped.items(): |
| 232 | newest = max(period for period, _basis in groups) |
| 233 | basis, _ = max(((basis, values) for (period, basis), values in groups.items() if period == newest), key=basis_rank) |
| 234 | selected = 0 |
| 235 | # ``reporting_basis`` is intentionally optional. Sort the explicit |
| 236 | # reporting period first and use a stable text projection only as a |
| 237 | # tie-breaker; do not normalize the stored basis identity. |
| 238 | for (period, candidate_basis), values in sorted( |
| 239 | groups.items(), key=lambda item: (item[0][0], item[0][1] or ""), reverse=True |
| 240 | ): |
| 241 | if candidate_basis != basis: |
| 242 | continue |
| 243 | if selected == 4: |
| 244 | break |
| 245 | primary = max(values.values(), key=lambda fact: fact_source_authority(fact.source_tier)) |
| 246 | result.append(FinancialResultPeriod( |
| 247 | period=period, period_type=kind, reporting_basis=basis, |
| 248 | revenue=values.get("revenue").value if values.get("revenue") else None, |
| 249 | operating_income=values.get("operating_income").value if values.get("operating_income") else None, |
| 250 | ebit=values.get("ebit").value if values.get("ebit") else None, |
| 251 | ebitda=values.get("ebitda").value if values.get("ebitda") else None, |
| 252 | pat=values.get("pat").value if values.get("pat") else None, |
| 253 | eps=values.get("eps").value if values.get("eps") else None, |
| 254 | source_name=primary.value.source_name, source_url=primary.value.source_url, |
| 255 | source_type=primary.value.source_type, published_at=primary.value.as_of_date, |
| 256 | retrieved_at=primary.value.retrieved_at, confidence=primary.value.confidence or 0.82, |
| 257 | )) |
| 258 | selected += 1 |
| 259 | return sorted(result, key=lambda item: (item.period_type, item.period), reverse=True) if period_type else result |
| 260 | |
| 261 | |
| 262 | def financial_statement_history_from_facts( |
| 263 | facts: list[FinancialFact], *, period_type: str | set[str], metrics: set[str], limit: int = 4, |
| 264 | ) -> list[FinancialStatementPeriod]: |
| 265 | """Project persisted statement facts without crossing reporting bases. |
| 266 | |
| 267 | This is deliberately a read projection: it neither derives financial values |
| 268 | nor changes durable precedence. For a selected period/basis, each metric |
| 269 | keeps the highest-authority persisted fact. |
| 270 | """ |
| 271 | period_types = {period_type} if isinstance(period_type, str) else period_type |
| 272 | grouped: dict[tuple[str, str, str | None], dict[str, FinancialFact]] = {} |
| 273 | for fact in facts: |
| 274 | if ( |
| 275 | fact.source_mode != SourceMode.REAL |
| 276 | or fact.source_tier not in SUPPORTED_FINANCIAL_SOURCE_TIERS |
| 277 | or fact.key.period_type not in period_types |
| 278 | or not fact.key.period_end |
| 279 | or fact.key.metric not in metrics |
| 280 | ): |
| 281 | continue |
| 282 | values = grouped.setdefault((fact.key.period_end, fact.key.period_type, fact.key.reporting_basis), {}) |
| 283 | existing = values.get(fact.key.metric) |
| 284 | if existing is None or fact_source_authority(fact.source_tier) > fact_source_authority(existing.source_tier): |
| 285 | values[fact.key.metric] = fact |
| 286 | |
| 287 | if not grouped: |
| 288 | return [] |
| 289 | |
| 290 | def basis_rank(item: tuple[str | None, dict[str, FinancialFact]]) -> tuple[int, int, int]: |
| 291 | basis, values = item |
| 292 | return ( |
| 293 | max(fact_source_authority(fact.source_tier) for fact in values.values()), |
| 294 | int(basis not in {None, "", "UNKNOWN"}), |
| 295 | int(basis == "CONSOLIDATED"), |
| 296 | ) |
| 297 | |
| 298 | selected_basis_by_period: dict[tuple[str, str], str | None] = {} |
| 299 | for period, kind, _basis in grouped: |
| 300 | candidates = [(basis, values) for (candidate_period, candidate_kind, basis), values in grouped.items() if candidate_period == period and candidate_kind == kind] |
| 301 | selected_basis_by_period[(period, kind)] = max(candidates, key=basis_rank)[0] |
| 302 | |
| 303 | result: list[FinancialStatementPeriod] = [] |
| 304 | # A caller asking for both annual and quarterly series gets a bounded |
| 305 | # history for each series; do not let the more recent quarterlies erase |
| 306 | # the annual view (or vice versa). |
| 307 | selected_by_kind: dict[str, int] = {} |
| 308 | for period, kind in sorted(selected_basis_by_period, reverse=True): |
| 309 | if selected_by_kind.get(kind, 0) >= limit: |
| 310 | continue |
| 311 | basis = selected_basis_by_period[(period, kind)] |
| 312 | values = grouped[(period, kind, basis)] |
| 313 | result.append(FinancialStatementPeriod( |
| 314 | period=period, |
| 315 | period_type=kind, |
| 316 | reporting_basis=basis, |
| 317 | metrics={metric: fact.value for metric, fact in values.items()}, |
| 318 | )) |
| 319 | selected_by_kind[kind] = selected_by_kind.get(kind, 0) + 1 |
| 320 | return result |
| 321 | |
| 322 | |
| 323 | def _apply_normalized_official_facts(result: QuarterlyResult, facts: list[FinancialFact]) -> None: |
| 324 | for fact in facts: |
| 325 | if (fact.source_tier not in SUPPORTED_FINANCIAL_SOURCE_TIERS |
| 326 | or fact.key.period_type != "QUARTERLY" |
| 327 | or fact.key.period_end not in {result.period, _canonical_quarter_end(result.period)}): |
| 328 | continue |
| 329 | if result.reporting_basis and fact.key.reporting_basis != result.reporting_basis: |
| 330 | continue |
| 331 | if fact.key.metric in {"revenue", "pat", "eps", "ebitda", "debt_or_borrowings"}: |
| 332 | if fact.source_tier == FactSourceTier.YAHOO and getattr(result, fact.key.metric) is not None: |
| 333 | continue |
| 334 | setattr(result, fact.key.metric, fact.value) |
| 335 | |
| 336 | |
| 337 | def _is_failed_pdf_reference(document: ResearchDocument) -> bool: |
| 338 | return (_enum_or_string_name(document.status) == "FAILED" |
| 339 | and _enum_or_string_name(document.document_type) == "PDF_REFERENCE") |
| 340 | |
| 341 | |
| 342 | def _enum_or_string_name(value: object) -> str: |
| 343 | """Return the stable domain name from persisted strings or enum instances.""" |
| 344 | if isinstance(value, str): |
| 345 | return value.upper() |
| 346 | if isinstance(value, Enum): |
| 347 | return value.name.upper() |
| 348 | return "" |
| 349 | |
| 350 | |
| 351 | def source_diversity(documents: list[ResearchDocument]) -> SourceDiversity: |
| 352 | unique = {document.canonical_url: document for document in documents}.values() |
| 353 | domains = {urlparse(document.canonical_url).hostname for document in unique if urlparse(document.canonical_url).hostname} |
| 354 | classifications = [document.source_classification for document in unique] |
| 355 | return SourceDiversity( |
| 356 | sources_found=len(list(unique)), |
| 357 | domains_found=len(domains), |
| 358 | official_sources=sum(value in {SourceClassification.REGULATORY, SourceClassification.EXCHANGE, SourceClassification.OFFICIAL_COMPANY} for value in classifications), |
| 359 | exchange_sources=sum(value == SourceClassification.EXCHANGE for value in classifications), |
| 360 | company_sources=sum(value == SourceClassification.OFFICIAL_COMPANY for value in classifications), |
| 361 | secondary_sources=sum(value not in {SourceClassification.REGULATORY, SourceClassification.EXCHANGE, SourceClassification.OFFICIAL_COMPANY} for value in classifications), |
| 362 | ) |
| 363 | |
| 364 | |
| 365 | @dataclass |
| 366 | class _FinancialParserDiagnostics: |
| 367 | document_id: str | None |
| 368 | source_name: str | None |
| 369 | evidence_at: datetime | None |
| 370 | text_length: int |
| 371 | candidate_discovery_calls: int = 0 |
| 372 | candidate_parse_calls: int = 0 |
| 373 | candidate_parse_elapsed_ms: float = 0.0 |
| 374 | boundary_lookup_elapsed_ms: float = 0.0 |
| 375 | basis_lookup_calls: int = 0 |
| 376 | basis_lookup_elapsed_ms: float = 0.0 |
| 377 | region_build_calls: int = 0 |
| 378 | region_build_elapsed_ms: float = 0.0 |
| 379 | region_chars: int = 0 |
| 380 | max_region_chars: int = 0 |
| 381 | header_parse_calls: int = 0 |
| 382 | header_parse_elapsed_ms: float = 0.0 |
| 383 | header_input_calls: int = 0 |
| 384 | header_input_keys: set[tuple[str, datetime | None]] = field(default_factory=set) |
| 385 | header_normalize_calls: int = 0 |
| 386 | header_normalize_elapsed_ms: float = 0.0 |
| 387 | first_row_search_calls: int = 0 |
| 388 | first_row_search_elapsed_ms: float = 0.0 |
| 389 | tokenize_header_calls: int = 0 |
| 390 | tokenize_header_requests: int = 0 |
| 391 | tokenize_header_cache_hits: int = 0 |
| 392 | tokenize_header_elapsed_ms: float = 0.0 |
| 393 | fallback_group_calls: int = 0 |
| 394 | fallback_group_elapsed_ms: float = 0.0 |
| 395 | header_cell_region_calls: int = 0 |
| 396 | header_cell_region_elapsed_ms: float = 0.0 |
| 397 | header_dates_calls: int = 0 |
| 398 | header_dates_elapsed_ms: float = 0.0 |
| 399 | financial_columns_calls: int = 0 |
| 400 | financial_columns_elapsed_ms: float = 0.0 |
| 401 | headingless_validation_calls: int = 0 |
| 402 | headingless_validation_elapsed_ms: float = 0.0 |
| 403 | candidate_rejection_counts: dict[str, int] = field(default_factory=dict) |
| 404 | statement_quality_calls: int = 0 |
| 405 | statement_quality_elapsed_ms: float = 0.0 |
| 406 | |
| 407 | |
| 408 | @dataclass(frozen=True) |
| 409 | class _StatementBoundaryIndex: |
| 410 | """Document-local locations for the existing candidate-boundary regexes.""" |
| 411 | |
| 412 | statement_headings: tuple[tuple[int, int], ...] |
| 413 | particulars: tuple[tuple[int, int], ...] |
| 414 | statement_ends: tuple[tuple[int, int], ...] |
| 415 | |
| 416 | @classmethod |
| 417 | def from_text(cls, text: str) -> _StatementBoundaryIndex: |
| 418 | return cls( |
| 419 | statement_headings=tuple((match.start(), match.end()) for match in _STATEMENT_HEADING.finditer(text)), |
| 420 | particulars=tuple((match.start(), match.end()) for match in _PARTICULARS_RE.finditer(text)), |
| 421 | statement_ends=tuple((match.start(), match.end()) for match in _STATEMENT_END.finditer(text)), |
| 422 | ) |
| 423 | |
| 424 | @staticmethod |
| 425 | def _at_or_after(spans: tuple[tuple[int, int], ...], offset: int) -> tuple[int, int] | None: |
| 426 | index = bisect_left(spans, (offset, -1)) |
| 427 | return spans[index] if index < len(spans) else None |
| 428 | |
| 429 | def statement_heading_at_or_after(self, offset: int) -> tuple[int, int] | None: |
| 430 | return self._at_or_after(self.statement_headings, offset) |
| 431 | |
| 432 | def particulars_at_or_after(self, offset: int) -> tuple[int, int] | None: |
| 433 | return self._at_or_after(self.particulars, offset) |
| 434 | |
| 435 | def statement_end_at_or_after(self, offset: int) -> tuple[int, int] | None: |
| 436 | return self._at_or_after(self.statement_ends, offset) |
| 437 | |
| 438 | |
| 439 | @dataclass |
| 440 | class _StatementCandidateParseContext: |
| 441 | """Mutable caches scoped to one document's candidate-discovery pass.""" |
| 442 | |
| 443 | boundary_index: _StatementBoundaryIndex |
| 444 | tokenized_headers: dict[tuple[str, datetime], tuple[tuple[str, ...], tuple[str, ...]]] = field(default_factory=dict) |
| 445 | |
| 446 | |
| 447 | def latest_quarterly_result(documents: list[ResearchDocument]) -> QuarterlyResult | None: |
| 448 | started_at = time.perf_counter() |
| 449 | total_candidate_discovery_calls = 0 |
| 450 | total_statement_quality_calls = 0 |
| 451 | total_statement_quality_elapsed_ms = 0.0 |
| 452 | |
| 453 | def log_total(extra_diagnostics: _FinancialParserDiagnostics | None = None) -> None: |
| 454 | logger.info( |
| 455 | "financial_parser_diag stage=TOTAL documentCount=%s durationMs=%s candidateDiscoveryCalls=%s " |
| 456 | "statementQualityCalls=%s statementQualityDurationMs=%s", |
| 457 | len(documents), |
| 458 | round((time.perf_counter() - started_at) * 1000), |
| 459 | total_candidate_discovery_calls + (extra_diagnostics.candidate_discovery_calls if extra_diagnostics else 0), |
| 460 | total_statement_quality_calls + (extra_diagnostics.statement_quality_calls if extra_diagnostics else 0), |
| 461 | round(total_statement_quality_elapsed_ms + (extra_diagnostics.statement_quality_elapsed_ms if extra_diagnostics else 0)), |
| 462 | ) |
| 463 | |
| 464 | candidates = [] |
| 465 | for document in documents: |
| 466 | text = document.normalized_text or document.raw_text or "" |
| 467 | evidence_at = document.published_at or document.retrieved_at |
| 468 | diagnostics = _FinancialParserDiagnostics( |
| 469 | document_id=document.document_id, |
| 470 | source_name=document.source_name, |
| 471 | evidence_at=evidence_at, |
| 472 | text_length=len(text), |
| 473 | ) |
| 474 | document_started_at = time.perf_counter() |
| 475 | statement_candidates: list[FinancialStatement] = [] |
| 476 | try: |
| 477 | statement_candidates = _nse_statement_candidates( |
| 478 | text, |
| 479 | evidence_at, |
| 480 | diagnostics=diagnostics, |
| 481 | diagnostic_pass="LATEST", |
| 482 | ) |
| 483 | statement = _select_nse_income_statement(statement_candidates, diagnostics=diagnostics) |
| 484 | period = statement.current_column.period_end if statement else _financial_result_period(text) |
| 485 | if _STATEMENT_HEADING.search(text) and statement is None: |
| 486 | continue |
| 487 | if not period or not any(term in text.lower() for term in ("revenue", "income", "pat", "profit after tax", "ebitda", "eps")): |
| 488 | continue |
| 489 | official = document.source_classification in { |
| 490 | SourceClassification.EXCHANGE, SourceClassification.REGULATORY, SourceClassification.OFFICIAL_COMPANY |
| 491 | } |
| 492 | basis = _reporting_basis(statement.region if statement else text) |
| 493 | candidates.append((1 if official else 0, int(basis == "CONSOLIDATED"), _period_key(period), |
| 494 | evidence_at, document, period, statement, statement_candidates)) |
| 495 | finally: |
| 496 | total_candidate_discovery_calls += diagnostics.candidate_discovery_calls |
| 497 | total_statement_quality_calls += diagnostics.statement_quality_calls |
| 498 | total_statement_quality_elapsed_ms += diagnostics.statement_quality_elapsed_ms |
| 499 | logger.info( |
| 500 | "financial_parser_diag stage=DOCUMENT documentId=%s source=%s evidenceAt=%s textLength=%s " |
| 501 | "durationMs=%s statementCandidateCount=%s candidateDiscoveryCalls=%s " |
| 502 | "statementQualityCalls=%s statementQualityDurationMs=%s", |
| 503 | diagnostics.document_id, |
| 504 | diagnostics.source_name, |
| 505 | diagnostics.evidence_at, |
| 506 | diagnostics.text_length, |
| 507 | round((time.perf_counter() - document_started_at) * 1000), |
| 508 | len(statement_candidates), |
| 509 | diagnostics.candidate_discovery_calls, |
| 510 | diagnostics.statement_quality_calls, |
| 511 | round(diagnostics.statement_quality_elapsed_ms), |
| 512 | ) |
| 513 | if not candidates: |
| 514 | log_total() |
| 515 | return None |
| 516 | _, _, _, _, document, period, statement, statement_candidates = max(candidates, key=lambda item: (item[0], item[1], item[2], item[3])) |
| 517 | text = document.normalized_text or document.raw_text or "" |
| 518 | provenance = lambda value, unit=None: _provenance(document, value, unit=unit, period=period, confidence=0.82) |
| 519 | revenue_yoy = _percent_metric(text, ("revenue", "total income"), "yoy", provenance) |
| 520 | pat_yoy = _percent_metric(text, ("profit after tax", "pat"), "yoy", provenance) |
| 521 | yoy_parts = [] |
| 522 | if revenue_yoy: |
| 523 | yoy_parts.append(f"Revenue {revenue_yoy.value}% YoY") |
| 524 | if pat_yoy: |
| 525 | yoy_parts.append(f"PAT {pat_yoy.value}% YoY") |
| 526 | selected_diagnostics = _FinancialParserDiagnostics( |
| 527 | document_id=document.document_id, |
| 528 | source_name=document.source_name, |
| 529 | evidence_at=document.published_at or document.retrieved_at, |
| 530 | text_length=len(text), |
| 531 | ) |
| 532 | aligned = sorted( |
| 533 | (candidate for candidate in statement_candidates if candidate.current_column.period_end == period), |
| 534 | key=lambda candidate: _statement_quality(candidate, diagnostics=selected_diagnostics), |
| 535 | reverse=True, |
| 536 | ) |
| 537 | logger.info( |
| 538 | "financial_parser_diag stage=QUALITY_SORT documentId=%s source=%s statementQualityCalls=%s " |
| 539 | "statementQualityDurationMs=%s", |
| 540 | selected_diagnostics.document_id, |
| 541 | selected_diagnostics.source_name, |
| 542 | selected_diagnostics.statement_quality_calls, |
| 543 | round(selected_diagnostics.statement_quality_elapsed_ms), |
| 544 | ) |
| 545 | log_total(selected_diagnostics) |
| 546 | def table_metric(labels): |
| 547 | for candidate in aligned: |
| 548 | if value := _statement_metric(candidate, labels, provenance): |
| 549 | return value |
| 550 | return _metric(text, labels, provenance) if not statement else None |
| 551 | def revenue_metric(): |
| 552 | for candidate in aligned: |
| 553 | if value := _statement_revenue(candidate, provenance): |
| 554 | return value |
| 555 | return None |
| 556 | def eps_metric(): |
| 557 | for candidate in aligned: |
| 558 | if value := _statement_eps(candidate, provenance): |
| 559 | return value |
| 560 | return None |
| 561 | return QuarterlyResult( |
| 562 | period=period.upper().replace(" ", " "), |
| 563 | document_title=document.title, |
| 564 | reporting_basis=_reporting_basis(statement.region if statement else text), |
| 565 | result_date=document.published_at, |
| 566 | revenue=revenue_metric() if statement else table_metric(("revenue", "total income")), |
| 567 | revenue_yoy_percent=revenue_yoy, |
| 568 | revenue_qoq_percent=_percent_metric(text, ("revenue", "total income"), "qoq", provenance), |
| 569 | ebitda=table_metric(("ebitda",)), |
| 570 | ebitda_margin=_percent_metric(text, ("ebitda margin",), None, provenance), |
| 571 | ebitda_yoy_percent=_percent_metric(text, ("ebitda",), "yoy", provenance), |
| 572 | pat=table_metric(("semantic:PAT",)), |
| 573 | pat_yoy_percent=pat_yoy, |
| 574 | pat_qoq_percent=_percent_metric(text, ("profit after tax", "pat"), "qoq", provenance), |
| 575 | eps=eps_metric() if statement else table_metric(("earning per share", "earnings per share", "eps")), |
| 576 | # Balance-sheet rows have their own reporting dates; do not attach them |
| 577 | # to a quarterly P&L table without a separately aligned parser. |
| 578 | debt_or_borrowings=None if statement else table_metric(("total debt", "borrowings")), |
| 579 | exceptional_items=_sentence_containing(text, ("exceptional item", "exceptional items")), |
| 580 | segment_information=_sentence_containing(text, ("segment revenue", "segment result")), |
| 581 | management_commentary=_commentary_highlights(text), |
| 582 | yoy_summary="; ".join(yoy_parts) or None, |
| 583 | nim=_percent_metric(text, ("net interest margin", "nim"), None, provenance), |
| 584 | roa=_percent_metric(text, ("return on assets", "roa"), None, provenance), |
| 585 | roe=_percent_metric(text, ("return on equity", "roe"), None, provenance), |
| 586 | gross_npa=_percent_metric(text, ("gross npa", "gnpa"), None, provenance), |
| 587 | net_npa=_percent_metric(text, ("net npa", "nnpa"), None, provenance), |
| 588 | deposits=_metric(text, ("deposits", "total deposits"), provenance), |
| 589 | advances=_metric(text, ("advances", "gross advances"), provenance), |
| 590 | capital_adequacy=_percent_metric(text, ("capital adequacy", "capital adequacy ratio", "crar"), None, provenance), |
| 591 | credit_cost=_percent_metric(text, ("credit cost",), None, provenance), |
| 592 | source_name=document.source_name, |
| 593 | source_url=document.canonical_url, |
| 594 | source_type=str(document.source_type), |
| 595 | published_at=document.published_at, |
| 596 | retrieved_at=document.retrieved_at, |
| 597 | confidence=0.82 if document.source_classification in {SourceClassification.EXCHANGE, SourceClassification.REGULATORY, SourceClassification.OFFICIAL_COMPANY} else 0.65, |
| 598 | ) |
| 599 | |
| 600 | |
| 601 | def parsed_nse_income_statement_periods(documents: list[ResearchDocument]) -> list[ParsedIncomeStatementPeriod]: |
| 602 | """Return explicit quarterly/annual NSE income-statement columns only. |
| 603 | |
| 604 | Values are read from the table's aligned cells; no annual value is summed |
| 605 | and no quarterly value is derived from a cumulative column. |
| 606 | """ |
| 607 | periods: dict[tuple[str, str, str | None], dict[str, ProvenancedValue]] = {} |
| 608 | for document in documents: |
| 609 | document_periods: dict[tuple[str, str, str | None], dict[str, ProvenancedValue]] = {} |
| 610 | metric_rankings: dict[tuple[tuple[str, str, str | None], str], tuple[int, int, int, int, int]] = {} |
| 611 | text = document.normalized_text or document.raw_text or "" |
| 612 | evidence_at = document.published_at or document.retrieved_at |
| 613 | candidates = sorted( |
| 614 | _nse_statement_candidates(text, evidence_at, require_quarterly=False), key=_statement_quality, reverse=True, |
| 615 | ) |
| 616 | for statement in candidates: |
| 617 | basis = _reporting_basis(statement.region) |
| 618 | unit = _reported_unit(statement.unit_context[:400], statement.unit_context[:800]) |
| 619 | eps_cells, eps_direct_columns = _statement_eps_candidate(statement) |
| 620 | rows = { |
| 621 | "revenue": _statement_revenue_cells(statement), |
| 622 | "pat": _classified_row_cells(statement, "PAT"), |
| 623 | "eps": eps_cells, |
| 624 | } |
| 625 | for column in statement.columns: |
| 626 | if column.period_type not in {"QUARTERLY", "ANNUAL"}: |
| 627 | continue |
| 628 | values = { |
| 629 | metric: cells[column.index] |
| 630 | for metric, cells in rows.items() |
| 631 | if cells is not None and cells[column.index] is not None |
| 632 | } |
| 633 | if not values: |
| 634 | continue |
| 635 | key = (column.period_end, column.period_type, basis) |
| 636 | rank = _statement_quality(statement) |
| 637 | for metric, value in values.items(): |
| 638 | # For the same document and fact key, a direct aligned |
| 639 | # EPS row is structurally stronger than the legacy |
| 640 | # preceding-token fallback. Table quality only breaks |
| 641 | # ties within the same extraction kind. |
| 642 | metric_rank = (int(metric != "eps" or eps_direct_columns[column.index]), *rank) |
| 643 | ranking_key = (key, metric) |
| 644 | if metric_rank > metric_rankings.get(ranking_key, (-1, -1, -1, -1, -1)): |
| 645 | document_periods.setdefault(key, {})[metric] = _provenance( |
| 646 | document, |
| 647 | value, |
| 648 | unit="INR per share" if metric == "eps" else unit, |
| 649 | period=column.period_end, |
| 650 | confidence=0.82, |
| 651 | ) |
| 652 | metric_rankings[ranking_key] = metric_rank |
| 653 | # Preserve the existing conservative cross-document first-source |
| 654 | # behavior; candidate precedence above is only within one document. |
| 655 | for key, metrics in document_periods.items(): |
| 656 | target = periods.setdefault(key, {}) |
| 657 | for metric, value in metrics.items(): |
| 658 | target.setdefault(metric, value) |
| 659 | return [ |
| 660 | ParsedIncomeStatementPeriod(period_end, period_type, basis, tuple(metrics.items())) |
| 661 | for (period_end, period_type, basis), metrics in sorted( |
| 662 | periods.items(), key=lambda item: (item[0][1], item[0][2] or "", item[0][0]), reverse=True |
| 663 | ) |
| 664 | ] |
| 665 | |
| 666 | |
| 667 | def parsed_nse_balance_sheet_periods(documents: list[ResearchDocument]) -> list[ParsedBalanceSheetPeriod]: |
| 668 | """Return explicit NSE balance-sheet ``As at`` columns without derivation.""" |
| 669 | periods: dict[tuple[str, str, str | None], dict[str, ProvenancedValue]] = {} |
| 670 | for document in documents: |
| 671 | text = document.normalized_text or document.raw_text or "" |
| 672 | for statement in _nse_balance_sheet_candidates(text, document.published_at or document.retrieved_at): |
| 673 | basis = _reporting_basis(statement.region) |
| 674 | unit = _reported_unit(statement.unit_context[:400], statement.unit_context[:800]) |
| 675 | rows = { |
| 676 | "total_assets": _aligned_row_cells(statement, r"\btotal\s+assets\b"), |
| 677 | "total_equity": _aligned_row_cells(statement, r"\btotal\s+equity\b"), |
| 678 | "total_liabilities": _aligned_row_cells(statement, r"\btotal\s+liabilities\b(?!\s+and\s+equity)"), |
| 679 | "current_assets": _aligned_row_cells(statement, r"\btotal\s+current\s+assets\b"), |
| 680 | "current_liabilities": _aligned_row_cells(statement, r"\btotal\s+current\s+liabilities\b"), |
| 681 | "cash_and_cash_equivalents": _aligned_row_cells(statement, r"\bcash\s+and\s+cash\s+equivalents?\b"), |
| 682 | "total_debt": _aligned_row_cells(statement, r"\b(?:total\s+debt|total\s+borrowings?|outstanding\s+debt)\b"), |
| 683 | } |
| 684 | for column in statement.columns: |
| 685 | values = { |
| 686 | metric: cells[column.index] for metric, cells in rows.items() |
| 687 | if cells is not None and cells[column.index] is not None |
| 688 | } |
| 689 | if not values: |
| 690 | continue |
| 691 | key = (column.period_end, "AS_AT", basis) |
| 692 | target = periods.setdefault(key, {}) |
| 693 | for metric, value in values.items(): |
| 694 | target.setdefault(metric, _provenance(document, value, unit=unit, period=column.period_end, confidence=0.82)) |
| 695 | return [ |
| 696 | ParsedBalanceSheetPeriod(period_end, period_type, basis, tuple(metrics.items())) |
| 697 | for (period_end, period_type, basis), metrics in sorted(periods.items(), reverse=True) |
| 698 | ] |
| 699 | |
| 700 | |
| 701 | def parsed_nse_cash_flow_periods(documents: list[ResearchDocument]) -> list[ParsedCashFlowPeriod]: |
| 702 | """Return only explicitly headed annual NSE cash-flow columns.""" |
| 703 | periods: dict[tuple[str, str, str | None], dict[str, ProvenancedValue]] = {} |
| 704 | for document in documents: |
| 705 | text = document.normalized_text or document.raw_text or "" |
| 706 | for statement in _nse_cash_flow_candidates(text, document.published_at or document.retrieved_at): |
| 707 | basis = _reporting_basis(statement.region) |
| 708 | unit = _reported_unit(statement.unit_context[:400], statement.unit_context[:800]) |
| 709 | rows = { |
| 710 | "cash_flow_from_operating_activities": _aligned_row_cells(statement, r"\bnet\s+cash\s+(?:(?:flow\s*/?\s*\(?used\)?\s+in|flow\s+from|generated\s+from|from))\s+(?:operating|operations)\s+activities\b"), |
| 711 | "cash_flow_from_investing_activities": _aligned_row_cells(statement, r"\bnet\s+cash\s+(?:(?:flow\s*/?\s*\(?used\)?\s+in|flow\s+from|used\s+in|from))\s+investing\s+activities\b"), |
| 712 | "cash_flow_from_financing_activities": _aligned_row_cells(statement, r"\bnet\s+cash\s+(?:(?:flow\s*/?\s*\(?used\)?\s+in|flow\s+from|used\s+in|from))\s+financing\s+activities\b"), |
| 713 | "net_change_in_cash": _aligned_row_cells(statement, r"\bnet\s+(?:increase|decrease|change)\s+in\s+cash(?:\s+and\s+cash\s+equivalents?)?\b"), |
| 714 | } |
| 715 | for column in statement.columns: |
| 716 | if column.period_type != "ANNUAL": |
| 717 | continue |
| 718 | values = {metric: cells[column.index] for metric, cells in rows.items() |
| 719 | if cells is not None and cells[column.index] is not None} |
| 720 | if not values: |
| 721 | continue |
| 722 | key = (column.period_end, column.period_type, basis) |
| 723 | target = periods.setdefault(key, {}) |
| 724 | for metric, value in values.items(): |
| 725 | target.setdefault(metric, _provenance(document, value, unit=unit, period=column.period_end, confidence=0.82)) |
| 726 | return [ParsedCashFlowPeriod(period_end, period_type, basis, tuple(metrics.items())) |
| 727 | for (period_end, period_type, basis), metrics in sorted(periods.items(), reverse=True)] |
| 728 | |
| 729 | |
| 730 | def _reporting_basis(text: str) -> str | None: |
| 731 | lowered = text.lower() |
| 732 | if "consolidated" in lowered: |
| 733 | return "CONSOLIDATED" |
| 734 | if "standalone" in lowered: |
| 735 | return "STANDALONE" |
| 736 | return None |
| 737 | |
| 738 | |
| 739 | def shareholding_changes(documents: list[ResearchDocument]) -> list[ShareholdingChange]: |
| 740 | results = [] |
| 741 | labels = { |
| 742 | "PROMOTER": ("promoter holding", "promoters"), |
| 743 | "FII_FPI": ("fii/fpi", "fii", "fpi"), |
| 744 | "DII": ("dii", "domestic institutional"), |
| 745 | "PUBLIC": ("public holding", "public shareholders"), |
| 746 | "PROMOTER_PLEDGED": ("promoter pledge", "pledged shares"), |
| 747 | } |
| 748 | for document in sorted(documents, key=lambda value: value.published_at or value.retrieved_at, reverse=True): |
| 749 | if document.source_classification not in { |
| 750 | SourceClassification.EXCHANGE, SourceClassification.REGULATORY, SourceClassification.OFFICIAL_COMPANY |
| 751 | }: |
| 752 | continue |
| 753 | text = document.normalized_text or document.raw_text or "" |
| 754 | periods = PERIOD_RE.findall(text) |
| 755 | if len(periods) < 2: |
| 756 | continue |
| 757 | for category, aliases in labels.items(): |
| 758 | values = _two_percentages(text, aliases) |
| 759 | if not values: |
| 760 | continue |
| 761 | current, previous = values |
| 762 | results.append(ShareholdingChange( |
| 763 | category=category, |
| 764 | current=_provenance(document, current, unit="PERCENT", period=periods[0], confidence=0.80), |
| 765 | previous=_provenance(document, previous, unit="PERCENT", period=periods[1], confidence=0.80), |
| 766 | current_period=periods[0], |
| 767 | previous_period=periods[1], |
| 768 | change_percentage_points=current - previous, |
| 769 | source_date=document.published_at, |
| 770 | )) |
| 771 | if results: |
| 772 | break |
| 773 | return results |
| 774 | |
| 775 | |
| 776 | def shareholding_changes_from_snapshots( |
| 777 | snapshots: list[ShareholdingSnapshot], |
| 778 | ) -> list[ShareholdingChange]: |
| 779 | """Compare the two latest distinct structured shareholding periods. |
| 780 | |
| 781 | Snapshot values are already normalized semantic categories. They are |
| 782 | deliberately selected, never aggregated: XBRL parent/child rows are not |
| 783 | additive ownership categories. |
| 784 | """ |
| 785 | valid = [snapshot for snapshot in snapshots |
| 786 | if snapshot.source_mode == SourceMode.REAL and snapshot.values] |
| 787 | latest_by_period: dict[date, ShareholdingSnapshot] = {} |
| 788 | for snapshot in sorted( |
| 789 | valid, |
| 790 | key=lambda value: ( |
| 791 | value.period_end, |
| 792 | value.published_at or datetime.min.replace(tzinfo=timezone.utc), |
| 793 | value.retrieved_at, |
| 794 | ), |
| 795 | reverse=True, |
| 796 | ): |
| 797 | latest_by_period.setdefault(snapshot.period_end.date(), snapshot) |
| 798 | periods = sorted(latest_by_period, reverse=True) |
| 799 | if len(periods) < 2: |
| 800 | return [] |
| 801 | |
| 802 | current_snapshot = latest_by_period[periods[0]] |
| 803 | previous_snapshot = latest_by_period[periods[1]] |
| 804 | current_values = _snapshot_values_by_category(current_snapshot) |
| 805 | previous_values = _snapshot_values_by_category(previous_snapshot) |
| 806 | results: list[ShareholdingChange] = [] |
| 807 | for category, current_value in current_values.items(): |
| 808 | previous_value = previous_values.get(category) |
| 809 | if previous_value is None: |
| 810 | continue |
| 811 | results.append(ShareholdingChange( |
| 812 | category=str(category), |
| 813 | current=_snapshot_provenance(current_snapshot, current_value), |
| 814 | previous=_snapshot_provenance(previous_snapshot, previous_value), |
| 815 | current_period=current_snapshot.period_end.date().isoformat(), |
| 816 | previous_period=previous_snapshot.period_end.date().isoformat(), |
| 817 | change_percentage_points=current_value.percentage - previous_value.percentage, |
| 818 | source_date=current_snapshot.published_at, |
| 819 | )) |
| 820 | return results |
| 821 | |
| 822 | |
| 823 | def _snapshot_values_by_category( |
| 824 | snapshot: ShareholdingSnapshot, |
| 825 | ) -> dict: |
| 826 | """Return one explicit source value per category without aggregation.""" |
| 827 | values = {} |
| 828 | for value in snapshot.values: |
| 829 | values.setdefault(value.category, value) |
| 830 | return values |
| 831 | |
| 832 | |
| 833 | def _snapshot_provenance( |
| 834 | snapshot: ShareholdingSnapshot, |
| 835 | value: ShareholdingSnapshotValue, |
| 836 | ) -> ProvenancedValue: |
| 837 | return ProvenancedValue( |
| 838 | value=value.percentage, |
| 839 | unit="PERCENT", |
| 840 | as_of_date=snapshot.period_end, |
| 841 | period=snapshot.period_end.date().isoformat(), |
| 842 | source_url=snapshot.source_url, |
| 843 | source_name=snapshot.source_provider, |
| 844 | source_type=snapshot.source_type, |
| 845 | published_at=snapshot.published_at, |
| 846 | retrieved_at=snapshot.retrieved_at, |
| 847 | confidence=float(snapshot.confidence), |
| 848 | ) |
| 849 | |
| 850 | |
| 851 | def valuation_assessment(documents: list[ResearchDocument]) -> ValuationAssessment: |
| 852 | metrics = {} |
| 853 | for document in sorted(documents, key=lambda value: value.published_at or value.retrieved_at, reverse=True): |
| 854 | text = document.normalized_text or document.raw_text or "" |
| 855 | for key, labels in { |
| 856 | "current_pe": ("current p/e", "p/e ratio", "pe ratio"), |
| 857 | "sector_pe": ("sector p/e", "industry p/e"), |
| 858 | "peer_pe": ("peer p/e", "peer median p/e"), |
| 859 | "historical_pe": ("historical p/e", "5 year p/e", "five year p/e"), |
| 860 | "roe": ("roe", "return on equity"), |
| 861 | "roce": ("roce", "return on capital employed"), |
| 862 | }.items(): |
| 863 | extractor = _percentage_label_number if key in {"roe", "roce"} else _label_number |
| 864 | if key not in metrics and (value := extractor(text, labels)) is not None: |
| 865 | metrics[key] = _provenance(document, value, unit="PERCENT" if key in {"roe", "roce"} else "RATIO", confidence=0.65) |
| 866 | current = _decimal_value(metrics.get("current_pe")) |
| 867 | benchmark_metrics = [(kind, metrics.get(key)) for kind, key in (("SECTOR_PE", "sector_pe"), ("PEER_PE", "peer_pe"), ("HISTORICAL_PE", "historical_pe"))] |
| 868 | benchmarks = [ValuationBenchmark(kind=kind, value=value) for kind, value in benchmark_metrics if value is not None and (_decimal_value(value) or Decimal(0)) > 0] |
| 869 | comparisons = [_decimal_value(benchmark.value) for benchmark in benchmarks] |
| 870 | state = "UNKNOWN" |
| 871 | reason = "Insufficient comparable public valuation evidence." |
| 872 | state_evidence = None |
| 873 | if current and comparisons: |
| 874 | benchmark = sum(comparisons) / Decimal(len(comparisons)) |
| 875 | ratio = current / benchmark |
| 876 | if ratio <= Decimal("0.80"): |
| 877 | state, reason = "CHEAP", "Current P/E is at least 20% below available sector, peer, or historical context." |
| 878 | elif ratio >= Decimal("1.20"): |
| 879 | state, reason = "EXPENSIVE", "Current P/E is at least 20% above available sector, peer, or historical context." |
| 880 | else: |
| 881 | state, reason = "FAIR", "Current P/E is within 20% of available sector, peer, or historical context." |
| 882 | state_evidence = ValuationStateEvidence(primary_metric="CURRENT_PE", current_value=metrics["current_pe"], benchmarks=benchmarks, benchmark_value=benchmark, comparison_ratio=ratio, comparison_method="CURRENT_PE_VS_AVAILABLE_PE_BENCHMARK_MEAN", explanation=reason) |
| 883 | return ValuationAssessment(state=state, reason=reason, state_evidence=state_evidence, **metrics) |
| 884 | |
| 885 | |
| 886 | def current_quarter_catalysts(events: list[ResearchEvent]) -> list[ResearchEvent]: |
| 887 | now = datetime.now(timezone.utc) |
| 888 | quarter = (now.month - 1) // 3 |
| 889 | aliases = { |
| 890 | ResearchEventType.NEW_ORDER: ResearchEventType.ORDER_WIN, |
| 891 | ResearchEventType.MAJOR_CONTRACT: ResearchEventType.NEW_CONTRACT, |
| 892 | ResearchEventType.GOVERNMENT_CONTRACT: ResearchEventType.NEW_CONTRACT, |
| 893 | ResearchEventType.NEW_CUSTOMER: ResearchEventType.CLIENT_WIN, |
| 894 | ResearchEventType.CUSTOMER_EXPANSION: ResearchEventType.CLIENT_WIN, |
| 895 | ResearchEventType.MAJOR_CUSTOMER: ResearchEventType.CLIENT_WIN, |
| 896 | ResearchEventType.FACTORY_EXPANSION: ResearchEventType.NEW_PLANT, |
| 897 | ResearchEventType.NEW_FACILITY: ResearchEventType.NEW_PLANT, |
| 898 | ResearchEventType.DEBT_CHANGE: ResearchEventType.BORROWING_CHANGE, |
| 899 | ResearchEventType.FUNDING: ResearchEventType.BORROWING_CHANGE, |
| 900 | ResearchEventType.GUIDANCE_RAISED: ResearchEventType.MANAGEMENT_GUIDANCE, |
| 901 | ResearchEventType.GUIDANCE_LOWERED: ResearchEventType.MANAGEMENT_GUIDANCE, |
| 902 | ResearchEventType.GUIDANCE_MAINTAINED: ResearchEventType.MANAGEMENT_GUIDANCE, |
| 903 | ResearchEventType.GUIDANCE_CUT: ResearchEventType.MANAGEMENT_GUIDANCE, |
| 904 | ResearchEventType.REVENUE_GUIDANCE: ResearchEventType.MANAGEMENT_GUIDANCE, |
| 905 | ResearchEventType.MARGIN_GUIDANCE: ResearchEventType.MANAGEMENT_GUIDANCE, |
| 906 | ResearchEventType.REGULATORY_EVENT: ResearchEventType.MAJOR_CORPORATE_ANNOUNCEMENT, |
| 907 | } |
| 908 | current = [] |
| 909 | for event in events: |
| 910 | event_time = event.event_date or event.published_at or event.detected_at |
| 911 | if event_time.year == now.year and (event_time.month - 1) // 3 == quarter: |
| 912 | current.append(event.model_copy(update={"event_type": aliases.get(event.event_type, event.event_type)})) |
| 913 | return current |
| 914 | |
| 915 | |
| 916 | def _metric(text, labels, factory): |
| 917 | for label in labels: |
| 918 | match = re.search(rf"\b{re.escape(label)}\b[^\d+-]{{0,24}}{NUMBER}([^.;\n]{{0,40}})", text, re.I) |
| 919 | if not match: |
| 920 | continue |
| 921 | value = _decimal(match.group(1)) |
| 922 | if value is None: |
| 923 | continue |
| 924 | reported_unit = _reported_unit(match.group(0), text[max(match.start() - 50, 0):match.end() + 50]) |
| 925 | return factory(value, reported_unit) |
| 926 | return None |
| 927 | |
| 928 | |
| 929 | def _financial_result_period(text: str) -> str | None: |
| 930 | if _NSE_QUARTER_HEADING.search(text): |
| 931 | return _nse_table_period_end(text) |
| 932 | return match.group(1) if (match := PERIOD_RE.search(text)) else None |
| 933 | |
| 934 | |
| 935 | def _parse_nse_income_statement( |
| 936 | text: str, |
| 937 | evidence_at: datetime, |
| 938 | *, |
| 939 | diagnostics: _FinancialParserDiagnostics | None = None, |
| 940 | ) -> FinancialStatement | None: |
| 941 | candidates = _nse_statement_candidates( |
| 942 | text, |
| 943 | evidence_at, |
| 944 | diagnostics=diagnostics, |
| 945 | diagnostic_pass="PARSE_INCOME", |
| 946 | ) |
| 947 | return _select_nse_income_statement(candidates, diagnostics=diagnostics) |
| 948 | |
| 949 | |
| 950 | def _select_nse_income_statement( |
| 951 | candidates: list[FinancialStatement], |
| 952 | *, |
| 953 | diagnostics: _FinancialParserDiagnostics | None = None, |
| 954 | ) -> FinancialStatement | None: |
| 955 | return max(candidates, key=lambda candidate: _statement_quality(candidate, diagnostics=diagnostics)) if candidates else None |
| 956 | |
| 957 | |
| 958 | def _nse_statement_candidates( |
| 959 | text: str, |
| 960 | evidence_at: datetime, |
| 961 | *, |
| 962 | require_quarterly: bool = True, |
| 963 | diagnostics: _FinancialParserDiagnostics | None = None, |
| 964 | diagnostic_pass: str | None = None, |
| 965 | ) -> list[FinancialStatement]: |
| 966 | started_at = time.perf_counter() if diagnostics else None |
| 967 | candidates = [] |
| 968 | boundary_index = _StatementBoundaryIndex.from_text(text) |
| 969 | parse_context = _StatementCandidateParseContext(boundary_index) |
| 970 | heading_starts = [(start, True) for start, _ in boundary_index.statement_headings] |
| 971 | particulars_starts = [(start, False) for start, _ in boundary_index.particulars] |
| 972 | starts = heading_starts + particulars_starts |
| 973 | for start, has_heading in starts: |
| 974 | candidate = _parse_statement_candidate( |
| 975 | text, |
| 976 | evidence_at, |
| 977 | start, |
| 978 | has_heading, |
| 979 | require_quarterly=require_quarterly, |
| 980 | boundary_index=boundary_index, |
| 981 | parse_context=parse_context, |
| 982 | diagnostics=diagnostics, |
| 983 | ) |
| 984 | if candidate is not None: |
| 985 | candidates.append(candidate) |
| 986 | if diagnostics: |
| 987 | diagnostics.candidate_discovery_calls += 1 |
| 988 | logger.info( |
| 989 | "financial_parser_diag stage=CANDIDATE_DISCOVERY pass=%s documentId=%s source=%s textLength=%s " |
| 990 | "statementHeadingCount=%s particularsCount=%s candidateStartCount=%s acceptedCandidateCount=%s durationMs=%s", |
| 991 | diagnostic_pass, |
| 992 | diagnostics.document_id, |
| 993 | diagnostics.source_name, |
| 994 | diagnostics.text_length, |
| 995 | len(heading_starts), |
| 996 | len(particulars_starts), |
| 997 | len(starts), |
| 998 | len(candidates), |
| 999 | round((time.perf_counter() - started_at) * 1000), |
| 1000 | ) |
| 1001 | logger.info( |
| 1002 | "financial_parser_diag stage=CANDIDATE_INTERNALS pass=%s documentId=%s candidateParseCalls=%s " |
| 1003 | "candidateParseDurationMs=%s boundaryLookupDurationMs=%s basisLookupCalls=%s basisLookupDurationMs=%s " |
| 1004 | "regionBuildCalls=%s regionBuildDurationMs=%s regionChars=%s maxRegionChars=%s " |
| 1005 | "headerParseCalls=%s headerParseDurationMs=%s headinglessValidationCalls=%s " |
| 1006 | "headinglessValidationDurationMs=%s rejectionCounts=%s", |
| 1007 | diagnostic_pass, |
| 1008 | diagnostics.document_id, |
| 1009 | diagnostics.candidate_parse_calls, |
| 1010 | round(diagnostics.candidate_parse_elapsed_ms), |
| 1011 | round(diagnostics.boundary_lookup_elapsed_ms), |
| 1012 | diagnostics.basis_lookup_calls, |
| 1013 | round(diagnostics.basis_lookup_elapsed_ms), |
| 1014 | diagnostics.region_build_calls, |
| 1015 | round(diagnostics.region_build_elapsed_ms), |
| 1016 | diagnostics.region_chars, |
| 1017 | diagnostics.max_region_chars, |
| 1018 | diagnostics.header_parse_calls, |
| 1019 | round(diagnostics.header_parse_elapsed_ms), |
| 1020 | diagnostics.headingless_validation_calls, |
| 1021 | round(diagnostics.headingless_validation_elapsed_ms), |
| 1022 | diagnostics.candidate_rejection_counts, |
| 1023 | ) |
| 1024 | logger.info( |
| 1025 | "financial_parser_diag stage=HEADER_INTERNALS pass=%s documentId=%s headerInputCalls=%s " |
| 1026 | "uniqueHeaderInputs=%s reusedHeaderInputCalls=%s headerNormalizeCalls=%s headerNormalizeDurationMs=%s " |
| 1027 | "firstRowSearchCalls=%s firstRowSearchDurationMs=%s tokenizeHeaderRequests=%s tokenizeHeaderCalls=%s " |
| 1028 | "tokenizeHeaderCacheHits=%s tokenizeHeaderDurationMs=%s " |
| 1029 | "fallbackGroupCalls=%s fallbackGroupDurationMs=%s headerCellRegionCalls=%s " |
| 1030 | "headerCellRegionDurationMs=%s headerDatesCalls=%s headerDatesDurationMs=%s " |
| 1031 | "financialColumnsCalls=%s financialColumnsDurationMs=%s", |
| 1032 | diagnostic_pass, |
| 1033 | diagnostics.document_id, |
| 1034 | diagnostics.header_input_calls, |
| 1035 | len(diagnostics.header_input_keys), |
| 1036 | diagnostics.header_input_calls - len(diagnostics.header_input_keys), |
| 1037 | diagnostics.header_normalize_calls, |
| 1038 | round(diagnostics.header_normalize_elapsed_ms), |
| 1039 | diagnostics.first_row_search_calls, |
| 1040 | round(diagnostics.first_row_search_elapsed_ms), |
| 1041 | diagnostics.tokenize_header_requests, |
| 1042 | diagnostics.tokenize_header_calls, |
| 1043 | diagnostics.tokenize_header_cache_hits, |
| 1044 | round(diagnostics.tokenize_header_elapsed_ms), |
| 1045 | diagnostics.fallback_group_calls, |
| 1046 | round(diagnostics.fallback_group_elapsed_ms), |
| 1047 | diagnostics.header_cell_region_calls, |
| 1048 | round(diagnostics.header_cell_region_elapsed_ms), |
| 1049 | diagnostics.header_dates_calls, |
| 1050 | round(diagnostics.header_dates_elapsed_ms), |
| 1051 | diagnostics.financial_columns_calls, |
| 1052 | round(diagnostics.financial_columns_elapsed_ms), |
| 1053 | ) |
| 1054 | return candidates |
| 1055 | |
| 1056 | |
| 1057 | def _nse_balance_sheet_candidates(text: str, evidence_at: datetime) -> list[FinancialStatement]: |
| 1058 | candidates = [] |
| 1059 | for heading in _BALANCE_SHEET_HEADING.finditer(text): |
| 1060 | particulars = _PARTICULARS_RE.search(text, heading.end()) |
| 1061 | if particulars is None or particulars.start() - heading.end() > 800: |
| 1062 | continue |
| 1063 | next_heading = _BALANCE_SHEET_HEADING.search(text, particulars.end()) |
| 1064 | next_income = _STATEMENT_HEADING.search(text, particulars.end()) |
| 1065 | end = min((match.start() for match in (next_heading, next_income) if match is not None), default=heading.start() + 8000) |
| 1066 | region_start = heading.start() |
| 1067 | basis_prefix = re.search(r"\b(?:standalone|consolidated)\s*$", text[max(0, heading.start() - 40):heading.start()], re.I) |
| 1068 | if basis_prefix: |
| 1069 | region_start = max(0, heading.start() - 40 + basis_prefix.start()) |
| 1070 | region = text[region_start:end] |
| 1071 | local_particulars = _PARTICULARS_RE.search(region) |
| 1072 | if local_particulars is None: |
| 1073 | continue |
| 1074 | header = _normalize_nse_header_ocr(region[local_particulars.end():local_particulars.end() + 500]) |
| 1075 | first_row = re.search(r"\b(?:total\s+assets|total\s+equity|cash\s+and\s+cash|current\s+assets|liabilities)\b", header, re.I) |
| 1076 | if first_row: |
| 1077 | header = header[:first_row.start()] |
| 1078 | dates = _header_dates(header, evidence_at, ["annual"]) |
| 1079 | columns = _financial_columns(["annual"], dates) |
| 1080 | if not columns: |
| 1081 | continue |
| 1082 | candidates.append(FinancialStatement(region, tuple(columns), max(columns, key=lambda column: column.period_end), region[:500])) |
| 1083 | return candidates |
| 1084 | |
| 1085 | |
| 1086 | def _compact_year_ended_dates(header: str) -> list[str]: |
| 1087 | """Tokenize a compact OCR ``YearEnded31March20Z5`` header structurally.""" |
| 1088 | compact = header.translate(str.maketrans({"Z": "2", "z": "2"})) |
| 1089 | values = [] |
| 1090 | for match in re.finditer(r"year\s*ended\D{0,8}(\d{1,2})\s*([A-Za-z]+)\s*(20\d{2})", compact, re.I): |
| 1091 | parsed = _nse_date(match.group(1), match.group(2), match.group(3)) |
| 1092 | if parsed is not None: |
| 1093 | values.append(parsed.isoformat()) |
| 1094 | return values |
| 1095 | |
| 1096 | |
| 1097 | def _nse_cash_flow_candidates(text: str, evidence_at: datetime) -> list[FinancialStatement]: |
| 1098 | candidates = [] |
| 1099 | for heading in _CASH_FLOW_HEADING.finditer(text): |
| 1100 | particulars = _PARTICULARS_RE.search(text, heading.end()) |
| 1101 | if particulars is None or particulars.start() - heading.end() > 800: |
| 1102 | continue |
| 1103 | next_heading = _CASH_FLOW_HEADING.search(text, particulars.end()) |
| 1104 | next_income = _STATEMENT_HEADING.search(text, particulars.end()) |
| 1105 | next_balance = _BALANCE_SHEET_HEADING.search(text, particulars.end()) |
| 1106 | end = min((match.start() for match in (next_heading, next_income, next_balance) if match is not None), default=heading.start() + 8000) |
| 1107 | region_start = heading.start() |
| 1108 | basis_prefix = re.search(r"\b(?:standalone|consolidated)\s*$", text[max(0, heading.start() - 40):heading.start()], re.I) |
| 1109 | if basis_prefix: |
| 1110 | region_start = max(0, heading.start() - 40 + basis_prefix.start()) |
| 1111 | region = text[region_start:end] |
| 1112 | local_particulars = _PARTICULARS_RE.search(region) |
| 1113 | if local_particulars is None: |
| 1114 | continue |
| 1115 | header = _normalize_nse_header_ocr(region[local_particulars.end():local_particulars.end() + 500]) |
| 1116 | # A March date alone is not a cash-flow period classification. |
| 1117 | if not re.search(r"year\s*ended", header, re.I): |
| 1118 | continue |
| 1119 | first_row = re.search(r"\b(?:net\s+cash|cash\s+flow)\b", header, re.I) |
| 1120 | if first_row: |
| 1121 | header = header[:first_row.start()] |
| 1122 | groups = [match.group(1).lower() for match in _GROUP_RE.finditer(header)] |
| 1123 | dates = _header_dates(header, evidence_at, groups) |
| 1124 | columns = _financial_columns(groups, dates) |
| 1125 | if not columns: |
| 1126 | continue |
| 1127 | candidates.append(FinancialStatement(region, tuple(columns), max(columns, key=lambda column: column.period_end), region[:500])) |
| 1128 | return candidates |
| 1129 | |
| 1130 | |
| 1131 | def _parse_statement_candidate( |
| 1132 | text: str, |
| 1133 | evidence_at: datetime, |
| 1134 | start: int, |
| 1135 | has_heading: bool, |
| 1136 | *, |
| 1137 | require_quarterly: bool = True, |
| 1138 | boundary_index: _StatementBoundaryIndex | None = None, |
| 1139 | parse_context: _StatementCandidateParseContext | None = None, |
| 1140 | diagnostics: _FinancialParserDiagnostics | None = None, |
| 1141 | ) -> FinancialStatement | None: |
| 1142 | candidate_started_at = time.perf_counter() if diagnostics else None |
| 1143 | |
| 1144 | def finish(value: FinancialStatement | None, rejection: str | None = None) -> FinancialStatement | None: |
| 1145 | if diagnostics: |
| 1146 | diagnostics.candidate_parse_calls += 1 |
| 1147 | diagnostics.candidate_parse_elapsed_ms += (time.perf_counter() - candidate_started_at) * 1000 |
| 1148 | if rejection: |
| 1149 | diagnostics.candidate_rejection_counts[rejection] = diagnostics.candidate_rejection_counts.get(rejection, 0) + 1 |
| 1150 | return value |
| 1151 | |
| 1152 | boundary_started_at = time.perf_counter() if diagnostics else None |
| 1153 | parse_context = parse_context or _StatementCandidateParseContext( |
| 1154 | boundary_index or _StatementBoundaryIndex.from_text(text), |
| 1155 | ) |
| 1156 | boundary_index = parse_context.boundary_index |
| 1157 | heading = boundary_index.statement_heading_at_or_after(start) if has_heading else None |
| 1158 | if has_heading and (heading is None or heading[0] != start): |
| 1159 | return finish(None, "HEADING_OFFSET_MISMATCH") |
| 1160 | particulars_match = boundary_index.particulars_at_or_after(start) |
| 1161 | if particulars_match is None: |
| 1162 | return finish(None, "NO_PARTICULARS") |
| 1163 | particulars_start = particulars_match[0] |
| 1164 | if not has_heading and particulars_start != start: |
| 1165 | return finish(None, "PARTICULARS_OFFSET_MISMATCH") |
| 1166 | start = heading[0] if heading else 0 |
| 1167 | next_statement = boundary_index.statement_heading_at_or_after(heading[1]) if heading else boundary_index.statement_heading_at_or_after(start + 1) |
| 1168 | statement_end = next_statement[0] if next_statement else None |
| 1169 | if next_statement and heading: |
| 1170 | basis_prefix = re.search(r"\b(?:standalone|consolidated)\s*$", text[max(heading[1], next_statement[0] - 40):next_statement[0]], re.I) |
| 1171 | if basis_prefix: |
| 1172 | statement_end = max(heading[1], next_statement[0] - 40) + basis_prefix.start() |
| 1173 | next_particulars = boundary_index.particulars_at_or_after(particulars_match[1]) |
| 1174 | next_particulars_start = next_particulars[0] if next_particulars else None |
| 1175 | statement_end_match = boundary_index.statement_end_at_or_after(heading[1] if heading else start) |
| 1176 | end_candidates = [value for value in (statement_end_match[0] if statement_end_match else None, statement_end, next_particulars_start) if value is not None] |
| 1177 | end = min(end_candidates) if end_candidates else start + 8000 |
| 1178 | if diagnostics: |
| 1179 | diagnostics.boundary_lookup_elapsed_ms += (time.perf_counter() - boundary_started_at) * 1000 |
| 1180 | # Do not let a later statement re-read metric rows from the previous PDF |
| 1181 | # page. Keep only an immediately preceding reporting-basis label. |
| 1182 | basis_started_at = time.perf_counter() if diagnostics else None |
| 1183 | basis_context = re.search(r"\b(?:standalone|consolidated)\s*$", text[max(0, start - 40):start], re.I) if heading else None |
| 1184 | if diagnostics and heading: |
| 1185 | diagnostics.basis_lookup_calls += 1 |
| 1186 | diagnostics.basis_lookup_elapsed_ms += (time.perf_counter() - basis_started_at) * 1000 |
| 1187 | region_start = max(0, start - 40 + basis_context.start()) if basis_context else start |
| 1188 | region_started_at = time.perf_counter() if diagnostics else None |
| 1189 | region = text[region_start:end] |
| 1190 | if diagnostics: |
| 1191 | diagnostics.region_build_calls += 1 |
| 1192 | diagnostics.region_build_elapsed_ms += (time.perf_counter() - region_started_at) * 1000 |
| 1193 | diagnostics.region_chars += len(region) |
| 1194 | diagnostics.max_region_chars = max(diagnostics.max_region_chars, len(region)) |
| 1195 | particulars = boundary_index.particulars_at_or_after(start) |
| 1196 | if particulars is None or particulars[1] > end: |
| 1197 | return finish(None, "NO_PARTICULARS_IN_REGION") |
| 1198 | header_started_at = time.perf_counter() if diagnostics else None |
| 1199 | try: |
| 1200 | header_offset = particulars[1] - region_start |
| 1201 | header = region[header_offset:header_offset + 700] |
| 1202 | # Preserve only an immediately adjacent group-label run such as |
| 1203 | # ``Quarter ended Year ended SI Particulars``. The selected text is |
| 1204 | # already bounded by this statement candidate and the anchored match |
| 1205 | # prevents prose or a previous table from becoming header context. |
| 1206 | pre_particulars = region[:particulars[0] - region_start] |
| 1207 | preceding_groups = _PRE_PARTICULARS_GROUPS_RE.search(pre_particulars) |
| 1208 | if preceding_groups: |
| 1209 | header = f"{preceding_groups.group()} {header}" |
| 1210 | if diagnostics: |
| 1211 | diagnostics.header_input_calls += 1 |
| 1212 | diagnostics.header_input_keys.add((header, evidence_at)) |
| 1213 | header = _normalize_nse_header_ocr(header, diagnostics=diagnostics) |
| 1214 | first_row_started_at = time.perf_counter() if diagnostics else None |
| 1215 | first_row = re.search(r"\b(?:revenue\s+from\s+operations|total\s+revenue|total\s+income|net\s+profit|profit\s+after\s+tax|earning(?:s)?\s+per\s+share)\b", header, re.I) |
| 1216 | if diagnostics: |
| 1217 | diagnostics.first_row_search_calls += 1 |
| 1218 | diagnostics.first_row_search_elapsed_ms += (time.perf_counter() - first_row_started_at) * 1000 |
| 1219 | if first_row: |
| 1220 | header = header[:first_row.start()] |
| 1221 | if diagnostics: |
| 1222 | diagnostics.tokenize_header_requests += 1 |
| 1223 | tokenization_key = (header, evidence_at) |
| 1224 | tokenized_header = parse_context.tokenized_headers.get(tokenization_key) |
| 1225 | if tokenized_header is None: |
| 1226 | tokenize_started_at = time.perf_counter() if diagnostics else None |
| 1227 | groups, dates = _tokenize_financial_header(header, evidence_at) |
| 1228 | tokenized_header = (tuple(groups), tuple(dates)) |
| 1229 | parse_context.tokenized_headers[tokenization_key] = tokenized_header |
| 1230 | if diagnostics: |
| 1231 | diagnostics.tokenize_header_calls += 1 |
| 1232 | diagnostics.tokenize_header_elapsed_ms += (time.perf_counter() - tokenize_started_at) * 1000 |
| 1233 | elif diagnostics: |
| 1234 | diagnostics.tokenize_header_cache_hits += 1 |
| 1235 | token_groups, token_dates = tokenized_header |
| 1236 | if token_groups: |
| 1237 | groups = token_groups |
| 1238 | else: |
| 1239 | fallback_groups_started_at = time.perf_counter() if diagnostics else None |
| 1240 | groups = _deduplicate_header_groups([match.group(1).lower() for match in _GROUP_RE.finditer(header)]) |
| 1241 | if diagnostics: |
| 1242 | diagnostics.fallback_group_calls += 1 |
| 1243 | diagnostics.fallback_group_elapsed_ms += (time.perf_counter() - fallback_groups_started_at) * 1000 |
| 1244 | header_cell_region_started_at = time.perf_counter() if diagnostics else None |
| 1245 | header = _header_cell_region(header, groups) |
| 1246 | if diagnostics: |
| 1247 | diagnostics.header_cell_region_calls += 1 |
| 1248 | diagnostics.header_cell_region_elapsed_ms += (time.perf_counter() - header_cell_region_started_at) * 1000 |
| 1249 | if not heading: |
| 1250 | validation_started_at = time.perf_counter() if diagnostics else None |
| 1251 | has_income_row = re.search(r"\b(?:revenue\s+from\s+operations|total\s+revenue|total\s+income|profit\s+(?:after\s+tax|for\s+the\s+period)|net\s+profit)\b", region, re.I) |
| 1252 | if diagnostics: |
| 1253 | diagnostics.headingless_validation_calls += 1 |
| 1254 | diagnostics.headingless_validation_elapsed_ms += (time.perf_counter() - validation_started_at) * 1000 |
| 1255 | if not has_income_row: |
| 1256 | return finish(None, "HEADINGLESS_NO_INCOME_ROW") |
| 1257 | dates = token_dates or _header_dates(header, evidence_at, groups, diagnostics=diagnostics) |
| 1258 | # A quarter-and-nine-month filing can lose the ``nine months ended`` |
| 1259 | # table-header token during PDF extraction and leave a spurious |
| 1260 | # ``year ended`` label nearby. Its five date cells still describe |
| 1261 | # three quarterly and two cumulative nine-month columns, not annual |
| 1262 | # results. The explicit report heading is stronger evidence than |
| 1263 | # that damaged table-group token. Do not apply this repair to a |
| 1264 | # six-column table, which may genuinely contain an annual column. |
| 1265 | if len(dates) == 5 and _quarter_nine_month_reporting_semantics(region): |
| 1266 | groups = ["quarter", "nine month"] |
| 1267 | columns = _financial_columns(groups, dates, diagnostics=diagnostics) |
| 1268 | finally: |
| 1269 | if diagnostics: |
| 1270 | diagnostics.header_parse_calls += 1 |
| 1271 | diagnostics.header_parse_elapsed_ms += (time.perf_counter() - header_started_at) * 1000 |
| 1272 | quarterly = [column for column in columns if column.period_type == "QUARTERLY"] |
| 1273 | annual = [column for column in columns if column.period_type == "ANNUAL"] |
| 1274 | if (require_quarterly and not quarterly) or (not quarterly and not annual): |
| 1275 | return finish(None, "NO_SUPPORTED_COLUMNS") |
| 1276 | current_columns = quarterly or annual |
| 1277 | return finish(FinancialStatement(region, tuple(columns), max(current_columns, key=lambda column: column.period_end), |
| 1278 | text[max(0, start - 250):end])) |
| 1279 | |
| 1280 | |
| 1281 | def _statement_quality( |
| 1282 | statement: FinancialStatement, |
| 1283 | *, |
| 1284 | diagnostics: _FinancialParserDiagnostics | None = None, |
| 1285 | ) -> tuple[int, int, int, int]: |
| 1286 | started_at = time.perf_counter() if diagnostics else None |
| 1287 | metrics = sum(_aligned_row_value(statement, pattern) is not None for pattern in ( |
| 1288 | r"(?<!total )\brevenue\s+from\s+operations\b", r"\btotal\s+revenue\s+from\s+operations\b", |
| 1289 | r"\bnet\s+profit\b", r"\bprofit\s+after\s+tax\b", r"\bprofit\s+for\s+the\s+period\b", |
| 1290 | r"\bbasic\b")) |
| 1291 | # Prefer a statement with a valid after-tax/PAT row when the general |
| 1292 | # metric count ties. Deliberately exclude before-tax profit rows. |
| 1293 | has_pat = any(_aligned_row_value(statement, rf"\b{re.escape(label)}\b") is not None for label in ( |
| 1294 | "net profit for the period after tax", "profit for the period from continuing operations", |
| 1295 | "profit after tax", "profit for the period", "pat")) |
| 1296 | result = metrics, int(has_pat), len(statement.columns), int(bool(_STATEMENT_HEADING.search(statement.region))) |
| 1297 | if diagnostics: |
| 1298 | diagnostics.statement_quality_calls += 1 |
| 1299 | diagnostics.statement_quality_elapsed_ms += (time.perf_counter() - started_at) * 1000 |
| 1300 | return result |
| 1301 | |
| 1302 | |
| 1303 | def _normalize_nse_header_ocr( |
| 1304 | header: str, |
| 1305 | *, |
| 1306 | diagnostics: _FinancialParserDiagnostics | None = None, |
| 1307 | ) -> str: |
| 1308 | """Repair only observed date/header tokens, never financial-value cells.""" |
| 1309 | started_at = time.perf_counter() if diagnostics else None |
| 1310 | for corrupted, repaired in ( |
| 1311 | ("3lst", "31st"), |
| 1312 | ("3Lst", "31st"), |
| 1313 | ("315t", "31st"), |
| 1314 | ("Deceli`I]er", "December"), |
| 1315 | ("De[embei", "December"), |
| 1316 | ("Se|)tember", "September"), |
| 1317 | ("DΓé¼cembe]", "December"), |
| 1318 | ("D▒cembe]", "December"), |
| 1319 | ("D€cembe]", "December"), |
| 1320 | ("Slat March", "31st March"), |
| 1321 | ("Quar`er", "Quarter"), |
| 1322 | ("Quar`er€nded", "Quarter Ended"), |
| 1323 | ("Nlne Momh", "Nine Month"), |
| 1324 | ("Year [rrded", "Year Ended"), |
| 1325 | ("so September", "30 September"), |
| 1326 | ("2o25", "2025"), |
| 1327 | ("Decembe.", "December"), |
| 1328 | ("Man:h", "March"), |
| 1329 | ): |
| 1330 | header = header.replace(corrupted, repaired) |
| 1331 | if diagnostics: |
| 1332 | diagnostics.header_normalize_calls += 1 |
| 1333 | diagnostics.header_normalize_elapsed_ms += (time.perf_counter() - started_at) * 1000 |
| 1334 | return header |
| 1335 | |
| 1336 | |
| 1337 | def _header_cell_region(header: str, groups: list[str]) -> str: |
| 1338 | """Keep the header through its expected year cells, excluding table rows.""" |
| 1339 | expected = {("quarter", "year"): 5, ("quarter", "nine month", "year"): 6, |
| 1340 | ("quarter", "nine months", "year"): 6}.get(tuple(groups)) |
| 1341 | if not expected: |
| 1342 | return header |
| 1343 | years = list(re.finditer(r"\b[2Z][0-9Z]{3}\b", header)) |
| 1344 | return header[:years[expected - 1].end()] if len(years) >= expected else header |
| 1345 | |
| 1346 | |
| 1347 | def _nse_header_dates(header: str) -> list[date | None]: |
| 1348 | """Read explicit day-month-year header cells in their source order.""" |
| 1349 | parsed: list[tuple[int, date | None]] = [] |
| 1350 | parsed.extend((match.start(), _nse_date(*match.groups())) for match in _NSE_DATE.finditer(header)) |
| 1351 | parsed.extend((match.start(), _nse_date(match.group(2), match.group(1), match.group(3))) |
| 1352 | for match in _NSE_MONTH_DAY_DATE.finditer(header)) |
| 1353 | parsed.extend((match.start(), _nse_numeric_date(*match.groups())) for match in _NSE_NUMERIC_DATE.finditer(header)) |
| 1354 | return [value for _offset, value in sorted(parsed)] |
| 1355 | |
| 1356 | |
| 1357 | def _nse_numeric_date(day: str, month: str, year: str) -> date | None: |
| 1358 | try: |
| 1359 | return date(int(year.replace("Z", "2")), int(month), int(day)) |
| 1360 | except ValueError: |
| 1361 | return None |
| 1362 | |
| 1363 | |
| 1364 | def _tokenize_financial_header(header: str, evidence_at: datetime) -> tuple[list[str], list[str]]: |
| 1365 | """Interpret compact table headers as structural tokens, never value text.""" |
| 1366 | raw = list(re.finditer(r"\S+", header[:700])) |
| 1367 | words = [re.sub(r"[^a-z]", "", item.group().lower()) for item in raw] |
| 1368 | groups: list[str] = [] |
| 1369 | for index in range(len(words)): |
| 1370 | phrase = "".join(words[index:index + 3]) |
| 1371 | if _edit_distance(phrase[:len("quarterended")], "quarterended") <= 2: |
| 1372 | groups.append("quarter") |
| 1373 | elif _edit_distance(phrase[:len("ninemonthended")], "ninemonthended") <= 3: |
| 1374 | groups.append("nine month") |
| 1375 | elif _edit_distance(phrase[:len("yearended")], "yearended") <= 2: |
| 1376 | groups.append("year") |
| 1377 | groups = _deduplicate_header_groups(groups) |
| 1378 | expected = {("quarter", "year"): 5, ("quarter", "nine month", "year"): 6}.get(tuple(groups), 0) |
| 1379 | if not expected: |
| 1380 | return [], [] |
| 1381 | months = [] |
| 1382 | for word in words: |
| 1383 | match = next((name for name in _MONTHS if _edit_distance(word, name) <= 2), None) |
| 1384 | months.append(match) |
| 1385 | pairs = [] |
| 1386 | for index, month in enumerate(months): |
| 1387 | if not month or index == 0: |
| 1388 | continue |
| 1389 | day = re.sub(r"(?i)(st|nd|rd|th)$", "", raw[index - 1].group()) |
| 1390 | day = day.translate(str.maketrans({"s": "3", "S": "3", "o": "0", "O": "0", "l": "1", "I": "1"})) |
| 1391 | digits = re.sub(r"\D", "", day) |
| 1392 | if digits and 1 <= int(digits) <= 31: |
| 1393 | pairs.append((int(digits), month)) |
| 1394 | years = [] |
| 1395 | for item in raw: |
| 1396 | normalized = item.group().translate(str.maketrans({"Z": "2", "z": "2", "o": "0", "O": "0", "l": "1", "I": "1"})) |
| 1397 | digits = re.sub(r"\D", "", normalized) |
| 1398 | if len(digits) == 4 and digits.startswith("20"): |
| 1399 | years.append(int(digits)) |
| 1400 | if len(pairs) < expected or len(years) < expected: |
| 1401 | return [], [] |
| 1402 | try: |
| 1403 | dates = [date(year, _MONTHS[month], day).isoformat() for (day, month), year in zip(pairs[:expected], years[:expected])] |
| 1404 | except ValueError: |
| 1405 | # An OCR-flattened header can pair a real day token with the wrong |
| 1406 | # month. Reject this tokenized header rather than manufacturing a |
| 1407 | # calendar date or aborting the surrounding document reconciliation. |
| 1408 | return [], [] |
| 1409 | floor, ceiling = evidence_at.year - 3, evidence_at.year + 1 |
| 1410 | return (groups, dates) if all(floor <= int(value[:4]) <= ceiling for value in dates) else ([], []) |
| 1411 | |
| 1412 | |
| 1413 | def _deduplicate_header_groups(groups: list[str]) -> list[str]: |
| 1414 | """OCR may duplicate an adjacent header group without adding columns.""" |
| 1415 | result = [] |
| 1416 | for group in groups: |
| 1417 | if not result or result[-1] != group: |
| 1418 | result.append(group) |
| 1419 | return result |
| 1420 | |
| 1421 | |
| 1422 | def _header_dates( |
| 1423 | header: str, |
| 1424 | evidence_at: datetime, |
| 1425 | groups: list[str], |
| 1426 | *, |
| 1427 | diagnostics: _FinancialParserDiagnostics | None = None, |
| 1428 | ) -> list[str]: |
| 1429 | started_at = time.perf_counter() if diagnostics else None |
| 1430 | try: |
| 1431 | header = _normalize_nse_header_ocr(header, diagnostics=diagnostics) |
| 1432 | groups = _deduplicate_header_groups(groups) |
| 1433 | direct = _nse_header_dates(header) |
| 1434 | candidates = [[value for value in direct if value is not None]] |
| 1435 | fragments = list(_NSE_DAY_MONTH.finditer(header)) |
| 1436 | month_day_fragments = [ |
| 1437 | match for match in re.finditer( |
| 1438 | r"\b([A-Za-z]{3,12})\s+(\d{1,2})(?:[lI]?st|nd|rd|th|tli)?(?:\s*,?\s*([2Z][0-9Z]{3}))?\b", |
| 1439 | header, |
| 1440 | ) |
| 1441 | if match.group(1).lower() in _MONTHS |
| 1442 | ] |
| 1443 | years = re.findall(r"\b([2Z][0-9Z]{3})\b", header) |
| 1444 | if len(fragments) == len(years): |
| 1445 | split = [_nse_date(match.group(1), match.group(2), year.replace("Z", "2")) |
| 1446 | for match, year in zip(fragments, years)] |
| 1447 | if all(value is not None for value in split): |
| 1448 | candidates.insert(0, split) |
| 1449 | explicit_years = [match.group(3) for match in month_day_fragments if match.group(3)] |
| 1450 | deferred_years = list(years) |
| 1451 | for explicit in explicit_years: |
| 1452 | deferred_years.remove(explicit) |
| 1453 | if len(month_day_fragments) == len(explicit_years) + len(deferred_years): |
| 1454 | deferred = iter(deferred_years) |
| 1455 | split = [ |
| 1456 | _nse_date(match.group(2), match.group(1), (match.group(3) or next(deferred)).replace("Z", "2")) |
| 1457 | for match in month_day_fragments |
| 1458 | ] |
| 1459 | if all(value is not None for value in split): |
| 1460 | candidates.insert(0, split) |
| 1461 | dates = next((candidate for candidate in candidates if _financial_columns(groups, [value.isoformat() for value in candidate], diagnostics=diagnostics)), []) |
| 1462 | if not dates: |
| 1463 | return [] |
| 1464 | floor = evidence_at.year - 3 |
| 1465 | ceiling = evidence_at.year + 1 |
| 1466 | if any(value.year < floor or value.year > ceiling for value in dates): |
| 1467 | return [] |
| 1468 | return [value.isoformat() for value in dates] |
| 1469 | finally: |
| 1470 | if diagnostics: |
| 1471 | diagnostics.header_dates_calls += 1 |
| 1472 | diagnostics.header_dates_elapsed_ms += (time.perf_counter() - started_at) * 1000 |
| 1473 | |
| 1474 | |
| 1475 | def _financial_columns( |
| 1476 | groups: list[str], |
| 1477 | dates: list[str], |
| 1478 | *, |
| 1479 | diagnostics: _FinancialParserDiagnostics | None = None, |
| 1480 | ) -> list[FinancialColumn]: |
| 1481 | """Associate date cells to explicit header groups for common NSE layouts.""" |
| 1482 | started_at = time.perf_counter() if diagnostics else None |
| 1483 | try: |
| 1484 | groups = _deduplicate_header_groups(groups) |
| 1485 | normalized = ["QUARTERLY" if group == "quarter" else "NINE_MONTH" if group.startswith("nine") |
| 1486 | else "HALF_YEAR" if group.startswith("half") else "ANNUAL" for group in groups] |
| 1487 | count = len(dates) |
| 1488 | if normalized == ["QUARTERLY", "ANNUAL"] and count == 3: |
| 1489 | widths = [2, 1] |
| 1490 | elif normalized == ["QUARTERLY", "ANNUAL"] and count in {4, 5}: |
| 1491 | widths = [3, count - 3] |
| 1492 | elif normalized == ["QUARTERLY", "NINE_MONTH", "ANNUAL"] and count == 6: |
| 1493 | widths = [3, 2, 1] |
| 1494 | elif normalized == ["QUARTERLY", "NINE_MONTH"] and count == 5: |
| 1495 | widths = [3, 2] |
| 1496 | elif normalized == ["ANNUAL", "QUARTERLY"] and count in {4, 5}: |
| 1497 | widths = [count - 3, 3] |
| 1498 | elif normalized == ["QUARTERLY"] and count in {2, 3}: |
| 1499 | widths = [count] |
| 1500 | elif normalized == ["ANNUAL"] and count in {1, 2, 3, 4}: |
| 1501 | widths = [count] |
| 1502 | else: |
| 1503 | return [] |
| 1504 | result, index = [], 0 |
| 1505 | for group, width in zip(normalized, widths): |
| 1506 | for _ in range(width): |
| 1507 | result.append(FinancialColumn(index, dates[index], group, group)) |
| 1508 | index += 1 |
| 1509 | return result |
| 1510 | finally: |
| 1511 | if diagnostics: |
| 1512 | diagnostics.financial_columns_calls += 1 |
| 1513 | diagnostics.financial_columns_elapsed_ms += (time.perf_counter() - started_at) * 1000 |
| 1514 | |
| 1515 | |
| 1516 | def _statement_metric(statement: FinancialStatement | None, labels, factory): |
| 1517 | if statement is None: |
| 1518 | return None |
| 1519 | for label in labels: |
| 1520 | if label.startswith("semantic:"): |
| 1521 | value = _classified_row_value(statement, label.split(":", 1)[1]) |
| 1522 | if value is not None: |
| 1523 | return factory(value, _reported_unit(statement.unit_context[:400], statement.unit_context[:800])) |
| 1524 | continue |
| 1525 | pattern = label[3:] if label.startswith("re:") else rf"\b{re.escape(label)}\b" |
| 1526 | value = _aligned_row_value(statement, pattern) |
| 1527 | if value is not None: |
| 1528 | return factory(value, _reported_unit(statement.unit_context[:400], statement.unit_context[:800])) |
| 1529 | return None |
| 1530 | |
| 1531 | |
| 1532 | def _row_semantic_tokens(label: str) -> frozenset[str]: |
| 1533 | """Classify OCR-damaged accounting labels without touching numeric cells.""" |
| 1534 | words = [re.sub(r"[^a-z]", "", word.lower()) for word in re.findall(r"[A-Za-z]+", label)] |
| 1535 | concepts = set() |
| 1536 | for word in words: |
| 1537 | if not word: |
| 1538 | continue |
| 1539 | if word.startswith("profit") or _edit_distance(word[:6], "profit") <= 2: |
| 1540 | concepts.add("PROFIT") |
| 1541 | if word.startswith("period") or _edit_distance(word[:6], "period") <= 2: |
| 1542 | concepts.add("PERIOD") |
| 1543 | if word.startswith("continu") or _edit_distance(word[:10], "continuing") <= 3: |
| 1544 | concepts.add("CONTINUING_OPERATIONS") |
| 1545 | if word.startswith("discontinu"): |
| 1546 | concepts.add("DISCONTINUED_OPERATIONS") |
| 1547 | if word.startswith("before") or _edit_distance(word[:6], "before") <= 2: |
| 1548 | concepts.add("BEFORE_TAX") |
| 1549 | if word.startswith("after") or _edit_distance(word[:5], "after") <= 2: |
| 1550 | concepts.add("AFTER_TAX") |
| 1551 | if word.startswith("tax"): |
| 1552 | concepts.add("TAX") |
| 1553 | if word.startswith("comprehens"): |
| 1554 | concepts.add("COMPREHENSIVE_INCOME") |
| 1555 | if word.startswith("revenue"): |
| 1556 | concepts.add("REVENUE") |
| 1557 | return frozenset(concepts) |
| 1558 | |
| 1559 | |
| 1560 | def _classify_financial_row(label: str) -> str | None: |
| 1561 | tokens = _row_semantic_tokens(label) |
| 1562 | if "COMPREHENSIVE_INCOME" in tokens or ("TAX" in tokens and "PROFIT" not in tokens): |
| 1563 | return None |
| 1564 | if "PROFIT" not in tokens: |
| 1565 | return None |
| 1566 | if "BEFORE_TAX" in tokens: |
| 1567 | return "PBT" |
| 1568 | if "AFTER_TAX" in tokens or "PERIOD" in tokens: |
| 1569 | return "PAT" |
| 1570 | return None |
| 1571 | |
| 1572 | |
| 1573 | def _classified_row_value(statement: FinancialStatement, classification: str) -> Decimal | None: |
| 1574 | cells = _classified_row_cells(statement, classification) |
| 1575 | return cells[statement.current_column.index] if cells is not None else None |
| 1576 | |
| 1577 | |
| 1578 | def _classified_row_cells(statement: FinancialStatement, classification: str) -> tuple[Decimal | None, ...] | None: |
| 1579 | text = _structural_row_text(statement.region) |
| 1580 | for match in re.finditer(r"\b(?:profit|proflt|ptoflt)\w*", text, re.I): |
| 1581 | tail = text[match.end():match.end() + 260] |
| 1582 | # Flattened NSE tables may retain a row/note reference immediately |
| 1583 | # after the semantic label, for example ``Net Profit ... (5)``. It |
| 1584 | # is not the first financial cell. This is deliberately local to |
| 1585 | # the semantic PAT/PBT extractor: an integer parenthesized reference |
| 1586 | # is only discarded before locating aligned table values. |
| 1587 | tail = _ROW_REFERENCE_FORMULA_RE.sub("", tail) |
| 1588 | tail = re.sub(r"\(\s*\d+(?:\s*[+\-*/•]\s*\d+)+\s*\)", "", tail) |
| 1589 | tail = re.sub(r"^\s*\(\s*\d{1,2}\s*\)\s*", "", tail) |
| 1590 | tokens = list(re.finditer(r"\S+", tail)) |
| 1591 | def is_row_reference(index: int, token: re.Match[str]) -> bool: |
| 1592 | # A standalone integer in parentheses is a common NSE row/note |
| 1593 | # marker. Treat it as such only when a complete aligned row |
| 1594 | # follows, so an incomplete value row still fails closed. |
| 1595 | if not re.fullmatch(r"\(\s*\d{1,2}\s*\)", token.group()): |
| 1596 | return False |
| 1597 | following = tokens[index + 1:index + 1 + len(statement.columns)] |
| 1598 | return len(following) == len(statement.columns) and all( |
| 1599 | _financial_number(value.group()) is not None for value in following |
| 1600 | ) |
| 1601 | first = next(( |
| 1602 | index for index, token in enumerate(tokens) |
| 1603 | if re.search(r"\d", token.group()) |
| 1604 | and not _ROW_REFERENCE_FORMULA_RE.fullmatch(token.group()) |
| 1605 | and not is_row_reference(index, token) |
| 1606 | ), None) |
| 1607 | if first is None: |
| 1608 | continue |
| 1609 | label = text[match.start():match.end() + tokens[first].start()] |
| 1610 | if _classify_financial_row(label) != classification: |
| 1611 | continue |
| 1612 | raw_cells = _coalesce_split_financial_cells( |
| 1613 | [token.group() for token in tokens[first:first + len(statement.columns) + 2]] |
| 1614 | )[:len(statement.columns)] |
| 1615 | if len(raw_cells) != len(statement.columns): |
| 1616 | continue |
| 1617 | cells = tuple(_financial_number(cell) for cell in raw_cells) |
| 1618 | if any(value is not None for value in cells): |
| 1619 | return cells |
| 1620 | return None |
| 1621 | |
| 1622 | |
| 1623 | def _quarter_nine_month_reporting_semantics(region: str) -> bool: |
| 1624 | """Recognize the filing-level period semantics before table OCR damage.""" |
| 1625 | return bool(re.search( |
| 1626 | r"\b(?:quarter|three\s+months?)\b.{0,80}\bnine\s+months?\b.{0,80}\bended\b", |
| 1627 | region[:700], |
| 1628 | re.I, |
| 1629 | )) |
| 1630 | |
| 1631 | |
| 1632 | def _statement_revenue(statement: FinancialStatement | None, factory): |
| 1633 | cells = _statement_revenue_cells(statement) |
| 1634 | if statement is None or cells is None: |
| 1635 | return None |
| 1636 | value = cells[statement.current_column.index] |
| 1637 | return factory(value, _reported_unit(statement.unit_context[:400], statement.unit_context[:800])) if value is not None else None |
| 1638 | |
| 1639 | |
| 1640 | def _statement_revenue_cells(statement: FinancialStatement | None) -> tuple[Decimal | None, ...] | None: |
| 1641 | if statement is None: |
| 1642 | return None |
| 1643 | for pattern in ( |
| 1644 | r"\btotal\s+revenue\s+from\s+operations\b", |
| 1645 | r"(?<!total )\brevenue\s+from\s+operations\b", |
| 1646 | r"\btotal\s+income\b", |
| 1647 | ): |
| 1648 | if (cells := _aligned_row_cells(statement, pattern)) is not None: |
| 1649 | return cells |
| 1650 | if re.search(pattern, _structural_row_text(statement.region), re.I): |
| 1651 | return None |
| 1652 | return None |
| 1653 | |
| 1654 | |
| 1655 | def _statement_eps(statement: FinancialStatement | None, factory): |
| 1656 | cells = _statement_eps_cells(statement) |
| 1657 | if statement is None or cells is None: |
| 1658 | return None |
| 1659 | value = cells[statement.current_column.index] |
| 1660 | return factory(value, "INR per share") if value is not None else None |
| 1661 | |
| 1662 | |
| 1663 | def _statement_eps_cells(statement: FinancialStatement | None) -> tuple[Decimal | None, ...] | None: |
| 1664 | return _statement_eps_candidate(statement)[0] |
| 1665 | |
| 1666 | |
| 1667 | def _statement_eps_candidate( |
| 1668 | statement: FinancialStatement | None, |
| 1669 | ) -> tuple[tuple[Decimal | None, ...] | None, tuple[bool, ...]]: |
| 1670 | """Return EPS cells and per-column direct-alignment provenance.""" |
| 1671 | if statement is None or not re.search( |
| 1672 | r"\b(?:earn(?:ing|lng)s?\s+per(?:\s+(?:equity|eciuity))?\s+(?:share|sr\)?are)|(?:basic|diluted)\s+eps)\b", |
| 1673 | statement.region, |
| 1674 | re.I, |
| 1675 | ): |
| 1676 | return None, () |
| 1677 | basic_cells = _aligned_row_cells(statement, r"\bbasic\s+(?:earnings?\s+per\s+share|eps)\b|\bbasic\b(?:\s*\(\s*rs\.?\s*\))?") |
| 1678 | diluted_cells = _aligned_row_cells(statement, r"\bdiluted\s+(?:earnings?\s+per\s+share|eps)\b|\bdiluted\b(?:\s*\(\s*rs\.?\s*\))?") |
| 1679 | # Use one complete aligned EPS row. Combining Basic and Diluted cells |
| 1680 | # column-by-column could shift values when one OCR row is malformed. |
| 1681 | direct_source = max( |
| 1682 | (cells for cells in (basic_cells, diluted_cells) if cells is not None), |
| 1683 | key=lambda cells: sum(value is not None for value in cells), |
| 1684 | default=None, |
| 1685 | ) |
| 1686 | minimum_aligned_cells = 1 if len(statement.columns) == 1 else 2 |
| 1687 | if direct_source is not None and sum(value is not None for value in direct_source) >= minimum_aligned_cells: |
| 1688 | direct_cells = direct_source |
| 1689 | direct_flags = tuple(value is not None for value in direct_cells) |
| 1690 | else: |
| 1691 | direct_cells = tuple(None for _ in statement.columns) |
| 1692 | direct_flags = tuple(False for _ in statement.columns) |
| 1693 | row_region = _structural_row_text(statement.region) |
| 1694 | label = re.search(r"earn(?:ing|lng)s?\s+per\s+(?:equity|eciuity)\s+(?:sh(?:are|are)|sr\)?are)\b", row_region, re.I) |
| 1695 | fallback_cells: tuple[Decimal | None, ...] | None = None |
| 1696 | if label: |
| 1697 | tokens = list(re.finditer(r"\S+", row_region[max(0, label.start() - 120):label.start()])) |
| 1698 | fallback_values = [_financial_number(token.group()) for token in tokens] |
| 1699 | valid = [cell for cell in fallback_values if cell is not None] |
| 1700 | if len(valid) >= len(statement.columns): |
| 1701 | fallback_cells = tuple(valid[-len(statement.columns):]) |
| 1702 | if fallback_cells is None: |
| 1703 | return (direct_cells if any(direct_flags) else basic_cells), direct_flags |
| 1704 | cells = tuple( |
| 1705 | direct_cells[index] if direct_flags[index] else fallback_cells[index] |
| 1706 | for index in range(len(statement.columns)) |
| 1707 | ) |
| 1708 | return cells, direct_flags |
| 1709 | |
| 1710 | |
| 1711 | def _aligned_row_value(statement: FinancialStatement, label_pattern: str) -> Decimal | None: |
| 1712 | cells = _aligned_row_cells(statement, label_pattern) |
| 1713 | return cells[statement.current_column.index] if cells is not None else None |
| 1714 | |
| 1715 | |
| 1716 | def _aligned_row_cells(statement: FinancialStatement, label_pattern: str) -> tuple[Decimal | None, ...] | None: |
| 1717 | row_region = _structural_row_text(statement.region) |
| 1718 | matches = re.finditer(label_pattern, row_region, re.I) |
| 1719 | match = next((candidate for candidate in matches |
| 1720 | if not re.match(r"\s*\(?before\b", row_region[candidate.end():], re.I)), None) |
| 1721 | if not match: |
| 1722 | return None |
| 1723 | tail = row_region[match.end():match.end() + 360] |
| 1724 | # Strip leading row annotations, but retain parenthesized numeric cells: |
| 1725 | # ``(A+B+C) (5,507.81)`` has an annotation followed by a legitimate |
| 1726 | # negative first-column value. |
| 1727 | tail = re.sub( |
| 1728 | r"^\s*(?:from\s+continuing\s+operations\s*)?(?:\((?![^)]*\d)[^)]+\)\s*)+", |
| 1729 | "", |
| 1730 | tail, |
| 1731 | flags=re.I, |
| 1732 | ) |
| 1733 | # Cash-flow rows commonly carry an alphabetic section marker such as |
| 1734 | # ``(A+B+C)`` between the row label and its aligned cells. |
| 1735 | tail = re.sub(r"^\s*(?:\([A-Za-z+]+\)\s*)+", "", tail) |
| 1736 | # Flattened statements retain row-reference formulae between a label and |
| 1737 | # its values, e.g. ``(1)+(2)`` or ``(10)-(11)``. They are not financial |
| 1738 | # cells: require two parenthesized row references joined by arithmetic |
| 1739 | # operators, preserving legitimate parenthesized negative values. |
| 1740 | tail = _ROW_REFERENCE_FORMULA_RE.sub("", tail) |
| 1741 | tail = re.sub(r"^\s*eps\b", "", tail, flags=re.I) |
| 1742 | tokens = list(re.finditer(r"\S+", tail)) |
| 1743 | def is_separator(token: str) -> bool: |
| 1744 | return token.strip().strip("()[]:;,." ) in {"I", "l", "|"} |
| 1745 | first_index = next((index for index, token in enumerate(tokens) |
| 1746 | if not is_separator(token.group(0)) and re.search(r"[0-9ZSOIl|\\]", token.group(0))), None) |
| 1747 | if first_index is None: |
| 1748 | return None |
| 1749 | prefix = re.sub(r"\(\s*rs\.?\s*\)", "", tail[:tokens[first_index].start()], flags=re.I) |
| 1750 | prefix = re.sub(r"\b[I|l]\b", "", prefix) |
| 1751 | prefix = re.sub(r"\bfrom\s+continuing\s+operations\b\s*(?:\([^)]+\))?", "", prefix, flags=re.I) |
| 1752 | if re.search(r"[A-Za-z]", prefix): |
| 1753 | return None |
| 1754 | raw_cells = _coalesce_split_financial_cells( |
| 1755 | [token.group(0) for token in tokens[first_index:] if not is_separator(token.group(0))] |
| 1756 | )[:len(statement.columns)] |
| 1757 | if len(raw_cells) != len(statement.columns): |
| 1758 | return None |
| 1759 | cells = tuple(_financial_number(token) for token in raw_cells) |
| 1760 | # A small whole number immediately followed by decimal financial cells is |
| 1761 | # a row/note marker in flattened statements, not a current-period value. |
| 1762 | if cells[0] is not None and cells[0] == cells[0].to_integral_value() and cells[0] <= 99 and any( |
| 1763 | value is not None and value != value.to_integral_value() for value in cells[1:]): |
| 1764 | return None |
| 1765 | return cells |
| 1766 | |
| 1767 | |
| 1768 | def _coalesce_split_financial_cells(tokens: list[str]) -> list[str]: |
| 1769 | """Rejoin a bounded OCR split such as ``551 .59`` into one table cell.""" |
| 1770 | result: list[str] = [] |
| 1771 | for token in tokens: |
| 1772 | if re.fullmatch(r"\.\d{1,2}", token) and result and re.fullmatch(r"[-+]?\d[\d,]*", result[-1]): |
| 1773 | result[-1] += token |
| 1774 | else: |
| 1775 | result.append(token) |
| 1776 | return result |
| 1777 | |
| 1778 | |
| 1779 | def _structural_row_text(text: str) -> str: |
| 1780 | """Repair bounded OCR row labels only; numeric cells are never altered.""" |
| 1781 | for corrupted, repaired in ( |
| 1782 | ("contjp±±i?gop,engtions", "continuing operations"), |
| 1783 | ("contjp▒▒i?gop,engtions", "continuing operations"), |
| 1784 | ("Total Reveiiue I:ron operations", "Total Revenue From operations"), |
| 1785 | ("Total Revenue I:ron operations", "Total Revenue From operations"), |
| 1786 | ("Profltforthe perfod From contliiLilng operatlons", "Profit for the period From continuing operations"), |
| 1787 | ("PTofltforthe perfod From contliiLilng operatlons", "Profit for the period From continuing operations"), |
| 1788 | ("Net Profk for the D€riod", "Net Profit for the period"), |
| 1789 | ("Re`/enue FiwTi oneratlan=", "Revenue From operations"), |
| 1790 | ("Re`/enue FiwTi oneratlan", "Revenue From operations"), |
| 1791 | ): |
| 1792 | text = text.replace(corrupted, repaired) |
| 1793 | return re.sub(r"contjp.{4}i\?gop,engtions", "continuing operations", text) |
| 1794 | |
| 1795 | |
| 1796 | def _financial_number(raw: str) -> Decimal | None: |
| 1797 | """Parse a whole OCR cell only when bounded substitutions yield one valid number.""" |
| 1798 | token = raw.strip().strip("()[]:;") |
| 1799 | if not re.fullmatch(r"[-+0-9,\.ZSOIl|]+", token, re.I): |
| 1800 | return None |
| 1801 | normalized = token.translate(str.maketrans({"Z": "2", "z": "2", "S": "5", "s": "5", "O": "0", "o": "0", "I": "1", "l": "1", "|": "1"})) |
| 1802 | if re.fullmatch(r"\d{1,3}\.\d{3}\.\d{2}", normalized): |
| 1803 | normalized = normalized.replace(".", ",", 1) |
| 1804 | if not re.fullmatch( |
| 1805 | r"[-+]?(?:\d{1,3}(?:,\d{3})+|\d{1,2}(?:,\d{2})*,\d{3}|\d+)(?:\.\d{1,2})?", |
| 1806 | normalized, |
| 1807 | ): |
| 1808 | return None |
| 1809 | return _decimal(normalized) |
| 1810 | |
| 1811 | |
| 1812 | def _nse_table_period_end(text: str) -> str | None: |
| 1813 | """NSE financial statements place the current reporting date first.""" |
| 1814 | heading = _NSE_QUARTER_HEADING.search(text) |
| 1815 | if not heading: |
| 1816 | return None |
| 1817 | header = text[heading.end():heading.end() + 260] |
| 1818 | fragment = _NSE_DAY_MONTH.search(header) |
| 1819 | year = re.search(r"\b([2Z]\d{3})\b", header) |
| 1820 | if fragment and year: |
| 1821 | parsed = _nse_date(fragment.group(1), fragment.group(2), year.group(1).replace("Z", "2")) |
| 1822 | if parsed: |
| 1823 | return parsed.isoformat() |
| 1824 | for match in _NSE_DATE.finditer(text, heading.end(), min(len(text), heading.end() + 220)): |
| 1825 | if parsed := _nse_date(*match.groups()): |
| 1826 | return parsed.isoformat() |
| 1827 | return None |
| 1828 | |
| 1829 | |
| 1830 | def _nse_date(day: str, month_text: str, year: str) -> date | None: |
| 1831 | compact = re.sub(r"[^a-z]", "", month_text.lower()) |
| 1832 | month = _MONTHS.get(compact) |
| 1833 | if month is None: |
| 1834 | matches = [number for name, number in _MONTHS.items() if _edit_distance(compact, name) <= 2] |
| 1835 | month = matches[0] if len(matches) == 1 else None |
| 1836 | try: |
| 1837 | return date(int(year.replace("Z", "2")), month or 0, int(day)) |
| 1838 | except ValueError: |
| 1839 | return None |
| 1840 | |
| 1841 | |
| 1842 | def _edit_distance(left: str, right: str) -> int: |
| 1843 | previous = list(range(len(right) + 1)) |
| 1844 | for row, char in enumerate(left, 1): |
| 1845 | current = [row] |
| 1846 | for column, other in enumerate(right, 1): |
| 1847 | current.append(min(current[-1] + 1, previous[column] + 1, previous[column - 1] + (char != other))) |
| 1848 | previous = current |
| 1849 | return previous[-1] |
| 1850 | |
| 1851 | |
| 1852 | def _nse_table_metric(text, labels, factory): |
| 1853 | """A table row's first numeric cell is the current-period column.""" |
| 1854 | for label in labels: |
| 1855 | match = re.search(rf"\b{re.escape(label)}\b[^\d+-]{{0,160}}{NUMBER}", text, re.I) |
| 1856 | if match and (value := _decimal(match.group(1))) is not None: |
| 1857 | return factory(value, _reported_unit(match.group(0), text[:match.end() + 120])) |
| 1858 | return None |
| 1859 | |
| 1860 | |
| 1861 | def _nse_revenue_metric(text, factory): |
| 1862 | # Prefer the explicitly named operations row; total revenue is a fallback |
| 1863 | # only when the statement does not expose that canonical row. |
| 1864 | for pattern in (r"(?<!total )\brevenue\s+from\s+operations\b", r"\btotal\s+revenue\s+from\s+operations\b", r"\btotal\s+income\b"): |
| 1865 | match = re.search(rf"{pattern}[^\d+-]{{0,160}}{NUMBER}", text, re.I) |
| 1866 | if match and (value := _decimal(match.group(1))) is not None: |
| 1867 | return factory(value, _reported_unit(match.group(0), text[:match.end() + 120])) |
| 1868 | return None |
| 1869 | |
| 1870 | |
| 1871 | def _nse_eps_metric(text, factory): |
| 1872 | """Read the Basic row, never the share face value in the EPS heading.""" |
| 1873 | heading = re.search(r"\bearning(?:s)?\s+per\s+share\b", text, re.I) |
| 1874 | if not heading: |
| 1875 | return None |
| 1876 | basic = re.search(r"\bbasic\b\s*(?:\(\s*rs\.?\s*\))?[^\d+-]{0,48}" + NUMBER, text[heading.end():heading.end() + 360], re.I) |
| 1877 | if not basic or (value := _decimal(basic.group(1))) is None: |
| 1878 | return None |
| 1879 | return factory(value, "INR per share") |
| 1880 | |
| 1881 | |
| 1882 | def _percent_metric(text, labels, qualifier, factory): |
| 1883 | for label in labels: |
| 1884 | suffix = rf"[^.\n]{{0,90}}?{re.escape(qualifier)}" if qualifier else r"[^.\n]{0,30}?" |
| 1885 | match = re.search(rf"\b{re.escape(label)}\b{suffix}[^\d+-]{{0,12}}{NUMBER}\s*%", text, re.I) |
| 1886 | if match: |
| 1887 | return factory(_decimal(match.group(1)), "PERCENT") |
| 1888 | return None |
| 1889 | |
| 1890 | |
| 1891 | def _label_number(text, labels): |
| 1892 | for label in labels: |
| 1893 | match = re.search(rf"\b{re.escape(label)}\b[^\d+-]{{0,24}}{NUMBER}", text, re.I) |
| 1894 | if match: |
| 1895 | return _decimal(match.group(1)) |
| 1896 | return None |
| 1897 | |
| 1898 | |
| 1899 | def _percentage_label_number(text, labels): |
| 1900 | """Prefer the resulting level after a basis-point delta, not the delta itself.""" |
| 1901 | for label in labels: |
| 1902 | match = re.search( |
| 1903 | rf"\b{re.escape(label)}\b[^\n.]{{0,100}}?[-+]?{NUMBER}\s*bps\b[^\n.]{{0,60}}?\bto\s+{NUMBER}\s*%", |
| 1904 | text, |
| 1905 | re.I, |
| 1906 | ) |
| 1907 | if match: |
| 1908 | return _decimal(match.group(2)) |
| 1909 | return _label_number(text, labels) |
| 1910 | |
| 1911 | |
| 1912 | def _two_percentages(text, labels): |
| 1913 | for label in labels: |
| 1914 | match = re.search(rf"\b{re.escape(label)}\b[^\n]{{0,160}}?{NUMBER}\s*%[^\n]{{0,80}}?{NUMBER}\s*%", text, re.I) |
| 1915 | if match: |
| 1916 | return _decimal(match.group(1)), _decimal(match.group(2)) |
| 1917 | return None |
| 1918 | |
| 1919 | |
| 1920 | def _decimal(raw): |
| 1921 | try: |
| 1922 | return Decimal(raw.replace(",", "")) |
| 1923 | except InvalidOperation: |
| 1924 | return None |
| 1925 | |
| 1926 | |
| 1927 | def _decimal_value(value): |
| 1928 | return Decimal(str(value.value)) if value is not None else None |
| 1929 | |
| 1930 | |
| 1931 | def _provenance(document, value, *, unit=None, period=None, confidence=None, calculation_basis=None): |
| 1932 | return ProvenancedValue(value=value, unit=unit, as_of_date=document.published_at, period=period, |
| 1933 | source_url=document.canonical_url, source_name=document.source_name, source_type=str(document.source_type), |
| 1934 | published_at=document.published_at, retrieved_at=document.retrieved_at, confidence=confidence, |
| 1935 | calculation_basis=calculation_basis) |
| 1936 | |
| 1937 | |
| 1938 | def _period_key(period: str) -> tuple[int, int]: |
| 1939 | if iso := re.fullmatch(r"(20\d{2})-(\d{2})-(\d{2})", period): |
| 1940 | return int(iso.group(1)), (int(iso.group(2)) - 1) // 3 + 1 |
| 1941 | quarter = re.search(r"Q([1-4])\s*(?:FY)?\s*(\d{2,4})", period, re.I) |
| 1942 | if quarter: |
| 1943 | year = int(quarter.group(2)) |
| 1944 | if year < 100: |
| 1945 | year += 2000 |
| 1946 | return year, int(quarter.group(1)) |
| 1947 | ended = re.search(r"ended\s+(?:(\d{1,2})\s+)?([A-Za-z]+)(?:\s+\d{1,2})?,?\s+(\d{4})", period, re.I) |
| 1948 | if ended: |
| 1949 | month = {name.lower(): index for index, name in enumerate( |
| 1950 | ("january", "february", "march", "april", "may", "june", "july", "august", "september", "october", "november", "december"), 1 |
| 1951 | )}.get(ended.group(2).lower(), 0) |
| 1952 | return int(ended.group(3)), (month - 1) // 3 + 1 if month else 0 |
| 1953 | return 0, 0 |
| 1954 | |
| 1955 | |
| 1956 | def _canonical_quarter_end(period: str) -> str | None: |
| 1957 | if re.fullmatch(r"20\d{2}-\d{2}-\d{2}", period): |
| 1958 | return period |
| 1959 | match = re.fullmatch(r"Q([1-4])\s*FY\s*(\d{2,4})", period.strip(), re.I) |
| 1960 | if not match: |
| 1961 | return None |
| 1962 | financial_year = int(match.group(2)) |
| 1963 | if financial_year < 100: |
| 1964 | financial_year += 2000 |
| 1965 | month, day = {"1": (6, 30), "2": (9, 30), "3": (12, 31), "4": (3, 31)}[match.group(1)] |
| 1966 | return date(financial_year - 1 if match.group(1) != "4" else financial_year, month, day).isoformat() |
| 1967 | |
| 1968 | |
| 1969 | def _reported_unit(value_text: str, context: str) -> str | None: |
| 1970 | """Preserve the reported scale; never silently compare lakh/crore/millions.""" |
| 1971 | evidence = f"{value_text} {context}".lower() |
| 1972 | currency = "INR" if "₹" in evidence or "inr" in evidence or re.search(r"\b(?:rs\.?|rupees?)\b", evidence) else "EUR" if "€" in evidence or "eur" in evidence else "USD" if "$" in evidence or "usd" in evidence else None |
| 1973 | scale = next((name for name in ("crore", "lakh", "billion", "million", "thousand") if re.search(rf"\b{name}s?\b", evidence)), None) |
| 1974 | if currency and scale: |
| 1975 | return f"{currency} {scale}" |
| 1976 | return currency or scale |
| 1977 | |
| 1978 | |
| 1979 | def _sentence_containing(text: str, labels: tuple[str, ...]) -> str | None: |
| 1980 | for sentence in re.split(r"(?<=[.!?])\s+|\n+", text): |
| 1981 | if any(label in sentence.lower() for label in labels): |
| 1982 | return sentence.strip()[:500] or None |
| 1983 | return None |
| 1984 | |
| 1985 | |
| 1986 | def _commentary_highlights(text: str) -> list[str]: |
| 1987 | highlights = [] |
| 1988 | for sentence in re.split(r"(?<=[.!?])\s+|\n+", text): |
| 1989 | lowered = sentence.lower() |
| 1990 | if any(term in lowered for term in ("management said", "management commentary", "outlook", "guidance")): |
| 1991 | cleaned = sentence.strip() |
| 1992 | if cleaned: |
| 1993 | highlights.append(cleaned[:500]) |
| 1994 | if len(highlights) == 3: |
| 1995 | break |
| 1996 | return highlights |