feat(cdsl): parse NPS holdings (holdings-only)

README.md

@@ -270,6 +270,17 @@ Demat statements return holdings (no transactions), grouped per demat account:
     }[],
   }[],
   parse_warnings: string[],              // non-fatal demat data-quality warnings
+  nps: {                                 // National Pension System, if present (CDSL)
+    pran: string | null,                 // PRAN (as printed; often masked)
+    nps_sp: string | null,               // NPS service provider / CRA, e.g. "PROT"
+    value: decimal,                      // reported NPS portfolio value
+    schemes: {
+      scheme: string, fund_manager: string | null,
+      tier: string | null,               // "I" | "II"
+      asset_class: string | null,        // "E" | "C" | "G" | "A"
+      units: decimal, nav: decimal, value: decimal,
+    }[],
+  } | null,
 }
 ```
 

casparser/cli.py

@@ -157,6 +157,35 @@ def print_nsdl(parsed_data: NSDLCASData):
 
     console.print(table)
 
+    # NPS holdings (CDSL, holdings-only)
+    if parsed_data.nps and parsed_data.nps.schemes:
+        nps = parsed_data.nps
+        nps_table = Table(title="NPS Holdings", show_lines=True)
+        nps_table.add_column("Scheme")
+        nps_table.add_column("Class")
+        nps_table.add_column("Tier")
+        nps_table.add_column("Units", justify="right")
+        nps_table.add_column("NAV", justify="right")
+        nps_table.add_column("Value", justify="right")
+        for s in nps.schemes:
+            nps_table.add_row(
+                s.scheme,
+                s.asset_class or "",
+                s.tier or "",
+                format_number(s.units),
+                format_number(s.nav),
+                formatINR(s.value),
+            )
+        console.print(nps_table)
+        meta = []
+        if nps.nps_sp:
+            meta.append(f"NPS-SP: [bold]{nps.nps_sp}[/]")
+        if nps.pran:
+            meta.append(f"PRAN: [bold]{nps.pran}[/]")
+        meta.append(f"Value: [bold green]{formatINR(nps.value)}[/]")
+        console.print("   ".join(meta))
+        console.print("")
+
     # Asset class breakdown
     equities_total = Decimal(0)
     mf_demat_total = Decimal(0)
@@ -186,7 +215,10 @@ def print_nsdl(parsed_data: NSDLCASData):
         asset_table.add_row("  Mutual Funds (demat)", formatINR(mf_demat_total))
     if mf_folio_total > 0:
         asset_table.add_row("  Mutual Fund Folios", formatINR(mf_folio_total))
-    total_all = debts_total + equities_total + mf_demat_total + mf_folio_total
+    nps_total = parsed_data.nps.value if parsed_data.nps else Decimal(0)
+    if nps_total > 0:
+        asset_table.add_row("  National Pension System", formatINR(nps_total))
+    total_all = debts_total + equities_total + mf_demat_total + mf_folio_total + nps_total
     asset_table.add_row("  " + "─" * 20, "")
     asset_table.add_row(
         f"  Total Portfolio Value [As of {data['statement_period']['to']}]",

casparser/parsers/cdsl.py

@@ -42,6 +42,8 @@
     DematOwner,
     Equity,
     MutualFund,
+    NPSAccount,
+    NPSScheme,
     NSDLCASData,
     StatementPeriod,
 )
@@ -97,6 +99,30 @@
     re.I | re.S,
 )
 
+# --- NPS section patterns ---
+# An NPS holding row's scheme name always begins with this marker.
+NPS_SCHEME_MARKER = "nps trust"
+# `NPS-SP : PROT   PRAN ID : <masked>` — service provider (CRA) + PRAN.
+NPS_SP_RE = re.compile(r"NPS[-\s]*SP\s*:\s*([A-Za-z0-9]+)", re.I)
+NPS_PRAN_RE = re.compile(r"PRAN\s*ID\s*:\s*([A-Z0-9]+)", re.I)
+# `Portfolio Value ` 56,05,171.60 as on 31-03-2026` — reported NPS total.
+NPS_VALUE_RE = re.compile(r"Portfolio\s+Value[^\d]*([\d,]+\.\d+)", re.I)
+# Tier ("TIER I"/"TIER II") and asset class ("SCHEME E/C/G/A") in the name.
+NPS_TIER_RE = re.compile(r"\bTIER\s*[-\s]*(I{1,2}|1|2)\b", re.I)
+NPS_ASSET_RE = re.compile(r"\bSCHEME\s+([A-Za-z])\b")
+# Text that marks a block as genuine NPS holding content (scheme name /
+# fund manager lines). Used to skip page furniture (bank header, tab bar,
+# investor name, column headers, "Page N of M" footer) that gets interleaved
+# when a scheme row straddles a page break.
+NPS_HOLDING_TEXT_RE = re.compile(
+    r"NPS\s+TRUST|PENSION\s+FUND|MANAGEMENT\s+LIMITED|SCHEME\s+[A-Z0-9]\s*-\s*TIER",
+    re.I,
+)
+# Minimum x-gap (points) that separates the two text columns (scheme name |
+# fund manager). Wrapped lines within a column share an x, so any gap this
+# large is a real column boundary — layout-independent, unlike a fixed x.
+_NPS_COL_GAP_MIN = 150.0
+
 
 # --- decimal helpers ---
 
@@ -220,6 +246,146 @@ def _split_bo_id(bo_id: str) -> Tuple[str, str, str]:
 # --- parser entry point ---
 
 
+def _norm_tier(t: str) -> str:
+    return {"1": "I", "2": "II"}.get(t.upper(), t.upper())
+
+
+def _is_nps_holding_block(block: Block) -> bool:
+    """True when a block is part of an NPS holding row — it carries a numeric
+    (units/nav) cell or scheme/fund-manager text. Everything else in the
+    holding region (bank header, tab bar, investor name, column headers,
+    "Page N of M" footer) is page furniture and is skipped."""
+    for c in block.cells:
+        if _looks_numeric(c.text):
+            return True
+        if NPS_HOLDING_TEXT_RE.search(c.text):
+            return True
+    return False
+
+
+def _build_nps_scheme(cells: list) -> Optional[NPSScheme]:
+    """Build one NPSScheme from an accumulated holding-row cell buffer.
+
+    A scheme row spans several extractor lines: two text columns (scheme
+    name, fund manager) and two numeric cells (units, then nav). Text is
+    grouped into the two columns by a coarse x split; the numeric cells are
+    assigned by order (left = units, right = nav). value = units * nav.
+    Rows without both numerics (e.g. redacted) are skipped.
+    """
+    text_cells = [c for c in cells if c.text.strip() and not _looks_numeric(c.text)]
+    num_cells = sorted((c for c in cells if _looks_numeric(c.text)), key=lambda c: c.x_left)
+    # Split the two text columns (scheme name | fund manager) by the largest
+    # x-gap rather than a fixed threshold — column x-positions drift between
+    # CAS versions. Left cluster = scheme name, right = fund manager.
+    by_x = sorted(text_cells, key=lambda c: c.x_left)
+    split = len(by_x)
+    if len(by_x) >= 2:
+        max_gap, at = max(
+            (by_x[i + 1].x_left - by_x[i].x_left, i + 1) for i in range(len(by_x) - 1)
+        )
+        if max_gap >= _NPS_COL_GAP_MIN:
+            split = at
+    name_cells = sorted(by_x[:split], key=lambda c: -c.y_top)
+    fm_cells = sorted(by_x[split:], key=lambda c: -c.y_top)
+    scheme = " ".join(c.text.replace("\n", " ").strip() for c in name_cells).strip()
+    scheme = re.sub(r"\s+", " ", scheme)
+    if NPS_SCHEME_MARKER not in scheme.lower():
+        return None
+    if len(num_cells) < 2:
+        return None  # units / nav not present (redacted) — cannot form a holding
+    fund_manager = " ".join(c.text.replace("\n", " ").strip() for c in fm_cells).strip()
+    fund_manager = re.sub(r"\s+", " ", fund_manager) or None
+    units = _to_decimal(num_cells[0].text)
+    nav = _to_decimal(num_cells[1].text)
+    tier = None
+    if m := NPS_TIER_RE.search(scheme):
+        tier = _norm_tier(m.group(1))
+    asset_class = None
+    if m := NPS_ASSET_RE.search(scheme):
+        asset_class = m.group(1).upper()
+    return NPSScheme(
+        scheme=scheme,
+        fund_manager=fund_manager,
+        tier=tier,
+        asset_class=asset_class,
+        units=units,
+        nav=nav,
+        value=(units * nav).quantize(Decimal("0.01")),
+    )
+
+
+def _parse_nps(blocks: List[Block]) -> Optional[NPSAccount]:
+    """Extract the NPS holdings section (if present) from a CDSL CAS.
+
+    Holdings only — the NPS transaction statement is intentionally not
+    parsed. Scans for the NPS `HOLDING STATEMENT` region (identified by its
+    `NPS TRUST-` scheme rows) and reads scheme name / fund manager /
+    units / nav; PRAN + NPS-SP + reported portfolio value from the section
+    text. Returns None when the CAS has no NPS section.
+    """
+    nps_sp: Optional[str] = None
+    pran: Optional[str] = None
+    value: Optional[Decimal] = None
+    schemes: List[NPSScheme] = []
+    mode: Optional[str] = None  # None | "txn" | "holding"
+    buf: list = []
+
+    def flush() -> None:
+        nonlocal buf
+        if buf:
+            sc = _build_nps_scheme(buf)
+            if sc is not None:
+                schemes.append(sc)
+            buf = []
+
+    for b in blocks:
+        txt = b.text()
+        low = txt.lower()
+        if nps_sp is None and (m := NPS_SP_RE.search(txt)):
+            nps_sp = m.group(1)
+        if pran is None and (m := NPS_PRAN_RE.search(txt)):
+            pran = m.group(1)
+        if "statement of transactions" in low:
+            flush()
+            mode = "txn"
+            continue
+        if "holding statement" in low and "as on" in low:
+            flush()
+            mode = "holding"
+            continue
+        if "portfolio value" in low:
+            # Closes the NPS holding region; capture the reported total only
+            # when we are actually closing NPS holdings (schemes seen).
+            if (schemes or buf) and (m := NPS_VALUE_RE.search(txt)):
+                value = _to_decimal(m.group(1))
+            flush()
+            mode = None
+            continue
+        if "nps investment summary" in low:
+            flush()
+            mode = None
+            continue
+        if mode != "holding":
+            continue
+        starts_scheme = any(c.text.strip().lower().startswith(NPS_SCHEME_MARKER) for c in b.cells)
+        if starts_scheme:
+            flush()
+            buf = list(b.cells)
+        elif buf and _is_nps_holding_block(b):
+            # Only holding content extends the row buffer; page furniture
+            # (bank header, tab bar, investor name, column headers, footer)
+            # interleaved across a page break is skipped so it doesn't leak
+            # into the scheme name of a row that straddles two pages.
+            buf.extend(b.cells)
+    flush()
+
+    if not schemes and value is None and nps_sp is None and pran is None:
+        return None
+    if value is None:
+        value = sum((s.value for s in schemes), Decimal(0))
+    return NPSAccount(pran=pran, nps_sp=nps_sp, value=value, schemes=schemes)
+
+
 def parse_cdsl(
     pdf_path: str,
     password: str,
@@ -411,6 +577,7 @@ def parse_cdsl(
             _atoms=atoms,
         ),
         file_type=file_type,
+        nps=_parse_nps(blocks),
     )
 
 

casparser/types.py

@@ -269,11 +269,51 @@ def fix_float(cls, data: dict):
         return data
 
 
+class NPSScheme(BaseModel):
+    """A single NPS scheme holding (one asset class within a tier).
+
+    NPS schemes are not securities — they carry no ISIN/AMFI code. The
+    depository CAS prints the pension-fund scheme name, the fund manager,
+    and the unit balance + NAV; `value` is `units * nav`.
+    """
+
+    scheme: str
+    fund_manager: Optional[str] = None
+    tier: Optional[str] = None  # "I" / "II"
+    asset_class: Optional[str] = None  # "E" | "C" | "G" | "A"
+    units: Decimal
+    nav: Decimal
+    value: Decimal
+
+    @model_validator(mode="before")
+    @classmethod
+    def fix_float(cls, data: dict):
+        for k, v in data.items():
+            try:
+                if issubclass(Decimal, cls.__annotations__[k]) and isinstance(v, str):
+                    data[k] = v.replace(",", "")
+            except TypeError:
+                pass
+        return data
+
+
+class NPSAccount(BaseModel):
+    """NPS holdings for a single PRAN, as printed in the depository CAS."""
+
+    pran: Optional[str] = None  # PRAN as printed (often masked)
+    nps_sp: Optional[str] = None  # NPS Service Provider / CRA (e.g. "PROT" = Protean)
+    value: Decimal  # reported "Portfolio Value" total for the PRAN
+    schemes: List[NPSScheme] = []
+
+
 class NSDLCASData(BaseModel):
     accounts: List[DematAccount]
     statement_period: StatementPeriod
     investor_info: InvestorInfo
     file_type: FileType
+    # National Pension System holdings, when the CAS includes an NPS
+    # section (holdings only — the transaction ledger is not parsed).
+    nps: Optional[NPSAccount] = None
     # Non-fatal data-quality warnings from NSDL demat parsing (e.g. a
     # holdings row whose nav/value could not be confirmed arithmetically).
     parse_warnings: List[str] = []