Citation Coverage Matrix¶

@cites("CRR Art. 111")
def build_product_to_risk_type_expr(
    product_col: str = "obs_product",
) -> pl.Expr:
    """Build a Polars expression mapping a concrete OBS product to its risk_type.

    Resolves the abstract Annex I ``risk_type`` bucket (FR / MLR / ...) from a
    normalised concrete product key via the ``obs_product_to_risk_type`` rulepack
    CategoryMap (rebound to ``_ANNEX1_PRODUCT_RISK_TYPE`` at module load). The
    mapping is framework-invariant (CRR Annex I == PRA PS1/26 Table A1 for every
    product in scope). Unknown / unmapped products and nulls produce a null
    result, so the caller can leave the existing ``risk_type`` resolution
    untouched.

    Args:
        product_col: Name of the obs_product column on the frame.

    Returns:
        String Polars expression evaluating to the resolved risk_type (or null
        when the product is null / unmapped).
    """
    casted = pl.col(product_col).cast(pl.Utf8, strict=False).fill_null("")
    lowered = casted.str.to_lowercase()
    canonical = lowered.replace_strict(OBS_PRODUCT_SYNONYMS, default=casted.str.to_uppercase())
    return canonical.replace_strict(
        _ANNEX1_PRODUCT_RISK_TYPE,
        default=pl.lit(None, dtype=pl.Utf8),
    )

@cites("CRR Art. 111")
def sa_ccf_expression(
    risk_type_col: str = "risk_type",
    is_basel_3_1: bool = False,
) -> pl.Expr:
    """Polars expression mapping risk_type to SA CCFs.

    CRR Art. 111 (Annex I categories) when ``is_basel_3_1`` is False, PRA
    PS1/26 Table A1 when True. The CCF values come from the rulepack
    (``sa_ccf`` lookup); unrecognised risk_type falls back to the
    MR-equivalent ``sa_ccf_default`` (50%).
    """
    table = _SA_CCF_B31_MAP if is_basel_3_1 else _SA_CCF_CRR_MAP
    canonical = _normalize_risk_type(risk_type_col)
    return (
        pl.when(canonical == "FR")
        .then(pl.lit(table["FR"]))
        .when(canonical == "FRC")
        .then(pl.lit(table["FRC"]))
        .when(canonical == "MR")
        .then(pl.lit(table["MR"]))
        # CRR Annex I Row 3 issued medium-risk OBS items — explicit 50%
        # (mirrors MR / Row 4) so EAD is provably equal, not a default fallback.
        .when(canonical == "MR_ISSUED")
        .then(pl.lit(table["MR_ISSUED"]))
        .when(canonical == "OC")
        .then(pl.lit(table["OC"]))
        .when(canonical == "MLR")
        .then(pl.lit(table["MLR"]))
        .when(canonical == "LR")
        .then(pl.lit(table["LR"]))
        .otherwise(pl.lit(_SA_CCF_DEFAULT))
    )

    @cites("CRR Art. 111")
    @cites("CRR Art. 166")
    def apply_ccf(
        self,
        exposures: pl.LazyFrame,
        config: CalculationConfig,
        *,
        pack: ResolvedRulepack | None = None,
    ) -> pl.LazyFrame:
        """
        Apply CCF to calculate EAD for off-balance sheet exposures.

        CCF determination follows CRR Art. 111 categories based on risk_type:
        - SA: FR=100%, MR=50%, MLR=20%, LR=0%
        - F-IRB Art. 166(8)(d): MR/MLR/OC commitments (credit lines / NIFs / RUFs)
          when ``is_obs_commitment=True`` -> 75%
        - F-IRB Art. 166(10) fallback: issued OBS items (``is_obs_commitment=False``)
          -> 100% FR / 50% MR / 20% MLR / 0% LR
        - F-IRB Art. 166(8)(b): MLR with ``is_short_term_trade_lc=True`` -> 20%
        - A-IRB CRR: Uses ccf_modelled if provided, otherwise falls back to SA
        - A-IRB B31: Own CCF only for revolving (non-100% SA); else SA CCF (Art. 166D)
        - Art. 111(1)(c): When underlying_risk_type is specified, CCF is capped
          at the lower of the commitment's CCF and the underlying OBS item's CCF

        Args:
            exposures: Exposures with nominal_amount, risk_type, and approach columns
            config: Calculation configuration

        Returns:
            LazyFrame with ead_from_ccf and ccf columns added
        """
        schema = exposures.collect_schema()
        names = schema.names()
        original_has_risk_type = "risk_type" in names
        original_has_underlying = "underlying_risk_type" in names
        original_has_interest = "interest" in names
        has_provision_cols = "nominal_after_provision" in names and "provision_on_drawn" in names

        exposures, added_cols = self._ensure_columns(exposures, names, has_provision_cols)
        exposures = self._compute_ccf(exposures, config, pack=pack)
        exposures = self._compute_ead(exposures, has_provision_cols, config, pack=pack)
        exposures = self._build_audit_trail(
            exposures, original_has_risk_type, original_has_underlying, original_has_interest
        )

        # Clean up temp and default-populated columns
        return exposures.drop(
            "_sa_ccf_from_risk_type",
            "_firb_ccf_from_risk_type",
            "_nominal_is_zero",
            *added_cols,
        )

    @cites("CRR Art. 111")
    @cites("PS1/26, paragraph 111")
    def _compute_ccf(
        self,
        exposures: pl.LazyFrame,
        config: CalculationConfig,
        *,
        pack: ResolvedRulepack | None = None,
    ) -> pl.LazyFrame:
        """Compute CCF based on risk type and approach.

        Determines SA and F-IRB CCFs from risk_type, then selects the final CCF
        based on the exposure's approach (SA/F-IRB/A-IRB).

        CRR Annex I / Art. 111(1) obs_product fill: before resolving CCFs, any row
        whose ``risk_type`` is null/empty has its ``risk_type`` resolved from the
        concrete ``obs_product`` key via ANNEX1_PRODUCT_RISK_TYPE (framework-
        invariant). An explicit ``risk_type`` always wins — the fill is gated on
        the existing value being null/empty.

        Applies the PRA PS1/26 Art. 111(1) Table A1 Row 4(b) override: a UK
        residential-property commitment (``is_uk_residential_mortgage_commitment``)
        gets a 50% CCF under Basel 3.1 — on the SA and the F-IRB / Slotting
        carrier alike (Art. 166C(1)) — except where the otherwise-resolved CCF is
        10% (Row 7 UCC) or 100% (Row 1/2) — the Row 4(b) carve-out.

        References:
        - CRR Art. 111(1) / Annex I: SA CCF buckets.
        - CRR Art. 166(8)(a)-(d): F-IRB bespoke supervisory CCFs.
        - CRR Art. 166(9): own-estimate (modelled) A-IRB CCFs are admissible
          only within the Art. 166(8) product scope.
        - CRR Art. 166(10): residual supervisory fallback for issued OBS items
          outside the Art. 166(8) scope.
        """
        # CRR Annex I / Art. 111(1): resolve risk_type from the concrete OBS
        # product when (and only when) no explicit risk_type was supplied. Explicit
        # risk_type always wins; an unmapped/null product yields null and leaves
        # risk_type unchanged.
        risk_type_is_blank = (
            pl.col("risk_type").cast(pl.Utf8, strict=False).fill_null("").str.len_chars() == 0
        )
        product_risk_type = build_product_to_risk_type_expr("obs_product")
        exposures = exposures.with_columns(
            pl.when(risk_type_is_blank & product_risk_type.is_not_null())
            .then(product_risk_type)
            .otherwise(pl.col("risk_type"))
            .alias("risk_type"),
        )

        resolved_pack = pack if pack is not None else RulepackV0.from_config(config).pack
        # S9c: the F-IRB-uses-SA-CCF routing gate (Art. 166C) reads the cited pack
        # Feature; sa_ccf_expression / _firb_ccf_for_col keep their is_basel_3_1 bool
        # plumbing params (Option B). All CCF VALUES stay static data-layer tables.
        is_b31 = resolved_pack.feature("firb_uses_sa_ccf")

        if is_b31:
            # Basel 3.1 Art. 166C: F-IRB uses SA CCFs (PRA PS1/26 Art. 111 Table A1)
            # FR=100%, MR=50%, MLR=20%, LR(UCC)=10%
            firb_ccf = sa_ccf_expression(is_basel_3_1=True)
        else:
            # CRR F-IRB: Art. 166(8)(d) -> 75% for credit lines / NIFs / RUFs
            # (is_obs_commitment=True); Art. 166(10) -> 100/50/20/0% fallback for
            # issued OBS items not in scope of paragraphs 1-8.
            firb_ccf = _firb_ccf_for_col("risk_type")

        exposures = exposures.with_columns(
            sa_ccf_expression(is_basel_3_1=is_b31).alias("_sa_ccf_from_risk_type"),
            firb_ccf.alias("_firb_ccf_from_risk_type"),
            (pl.col("nominal_amount").cast(pl.Float64, strict=False).abs() < 1e-10).alias(
                "_nominal_is_zero"
            ),
        )

        # CRR maturity-dependent OC override (Art. 111(1) / Annex I items 2(b),
        # 3(b)): "other commitments" attract the MR 50% CCF when their ORIGINAL
        # maturity is > 1yr (item 2(b)) and the MLR 20% CCF when it is <= 1yr
        # (item 3(b)). The split keys on ORIGINAL maturity, not residual: the
        # explicit ``original_maturity_years`` when present, else the
        # (maturity_date - value_date) start-date fallback. With no origination
        # source the conservative MR 50% default (from sa_ccf_expression) stands.
        if not is_b31:
            exposures = self._apply_oc_original_maturity_ccf(exposures)

        # PRA PS1/26 Art. 111(1) Table A1 Row 4(b): commitments to extend credit
        # secured by residential property attract a 50% CCF — "to the extent
        # that they are not subject to a conversion factor of 10% or 100%". The
        # override lands on BOTH the SA and the F-IRB carrier because Art. 166C(1)
        # defines the F-IRB / Slotting CCF as the Art. 111 SA CCF (P1.251).
        # No effect under CRR (Table A1 is Basel 3.1 only) — see the gate below.
        if is_b31:
            exposures = self._apply_uk_residential_mortgage_ccf(exposures)
            exposures = self._apply_purchased_receivable_ccf(exposures)

        # Art. 111(1)(c): commitment-to-issue lower-of rule.
        # When underlying_risk_type is specified, cap CCFs at the underlying item's CCF.
        # "the lower of (i) the CCF applicable to the underlying OBS item and
        #  (ii) the CCF applicable to the commitment type"
        has_underlying = pl.col("underlying_risk_type").fill_null("").str.len_chars() > 0
        underlying_sa = sa_ccf_expression("underlying_risk_type", is_basel_3_1=is_b31)
        exposures = exposures.with_columns(
            pl.when(has_underlying)
            .then(pl.min_horizontal(pl.col("_sa_ccf_from_risk_type"), underlying_sa))
            .otherwise(pl.col("_sa_ccf_from_risk_type"))
            .alias("_sa_ccf_from_risk_type"),
            pl.when(has_underlying)
            .then(
                pl.min_horizontal(
                    pl.col("_firb_ccf_from_risk_type"),
                    sa_ccf_expression("underlying_risk_type", is_basel_3_1=True)
                    if is_b31
                    else _firb_ccf_for_col("underlying_risk_type"),
                )
            )
            .otherwise(pl.col("_firb_ccf_from_risk_type"))
            .alias("_firb_ccf_from_risk_type"),
        )

        # A-IRB CCF: use modelled value, with Basel 3.1 restrictions
        ccf_modelled_expr = pl.col("ccf_modelled").cast(pl.Float64, strict=False)
        if is_b31:
            # Basel 3.1 Art. 166D(1)(a): own-estimate CCFs only for revolving
            # facilities whose SA CCF is not 100% (Table A1 Row 2 carve-out).
            # Non-revolving A-IRB must use SA CCFs from Table A1.
            # Revolving with SA CCF < 100%: own CCF with 50% SA floor (CRE32.27).
            airb_revolving_ccf = pl.max_horizontal(
                ccf_modelled_expr.fill_null(pl.col("_sa_ccf_from_risk_type")),
                pl.col("_sa_ccf_from_risk_type")
                * scalar_value(resolved_pack.scalar_param("airb_revolving_ccf_floor_multiplier")),
            )
            is_eligible_for_own_ccf = pl.col("is_revolving").fill_null(False) & (
                pl.col("_sa_ccf_from_risk_type") < 1.0
            )
            airb_ccf = (
                pl.when(is_eligible_for_own_ccf)
                .then(airb_revolving_ccf)
                .otherwise(pl.col("_sa_ccf_from_risk_type"))
            )
        else:
            # CRR Art. 166(9): own-estimate (modelled) A-IRB CCFs are admissible
            # only within the Art. 166(8) product scope — undrawn commitments
            # (``is_obs_commitment``) and short-term trade LCs
            # (``is_short_term_trade_lc``) — and never for FR/FRC full-risk
            # substitutes. Out-of-scope rows (issued OBS items governed by
            # Art. 166(10), full-risk items) take the supervisory F-IRB CCF
            # unconditionally, so a spuriously low modelled value is ignored.
            # In scope, a null modelled CCF falls back to the Art. 166(8)/(10)
            # supervisory value (``_firb_ccf_from_risk_type``), NOT the SA
            # Art. 111 CCF.
            in_166_8_scope = (
                pl.col("is_obs_commitment").fill_null(True)
                | pl.col("is_short_term_trade_lc").fill_null(False)
            ) & ~_normalize_risk_type("risk_type").is_in(["FR", "FRC"])
            airb_ccf = (
                pl.when(in_166_8_scope)
                .then(ccf_modelled_expr.fill_null(pl.col("_firb_ccf_from_risk_type")))
                .otherwise(pl.col("_firb_ccf_from_risk_type"))
            )

        # Select final CCF based on approach
        return exposures.with_columns(
            pl.when(pl.col("_nominal_is_zero"))
            .then(pl.lit(0.0))
            .when(pl.col("approach") == ApproachType.AIRB.value)
            .then(airb_ccf)
            .when(pl.col("approach") == ApproachType.FIRB.value)
            .then(pl.col("_firb_ccf_from_risk_type"))
            # CRR Art. 147(8): specialised-lending slotting is a corporate IRB
            # exposure, so its OBS EAD is governed by Art. 166(8) — the F-IRB CCF
            # (e.g. MR -> 75%), not the SA 50%. Under Basel 3.1, Art. 166C makes
            # F-IRB CCFs equal SA CCFs, so slotting stays on the SA path below.
            .when((pl.col("approach") == ApproachType.SLOTTING.value) & (not is_b31))
            .then(pl.col("_firb_ccf_from_risk_type"))
            .otherwise(pl.col("_sa_ccf_from_risk_type"))
            .alias("ccf"),
        )

    @cites("CRR Art. 111(1)")
    def _apply_oc_original_maturity_ccf(
        self,
        exposures: pl.LazyFrame,
    ) -> pl.LazyFrame:
        """Remap the OC ("other commitments") SA CCF on ORIGINAL maturity (CRR).

        CRR Annex I items 2(b)/3(b) split "other commitments" on their ORIGINAL
        maturity: 50% (MR) when > 1yr, 20% (MLR) when <= 1yr. ``sa_ccf_expression``
        supplies the conservative 50% MR default; this override drops the CCF to
        the 20% MLR rate only where an original-maturity source exists and is at
        or below the ``oc_short_maturity_threshold_days`` (365-day) boundary.

        Original maturity is taken from ``original_maturity_years`` when present,
        else the ``(maturity_date - value_date)`` start-date fallback. With neither
        source the row keeps the conservative 50% MR default.
        """
        normalized_rt = pl.col("risk_type").fill_null("").str.to_lowercase()
        is_oc = normalized_rt.is_in(["oc", "other_commit"])
        schema_names = exposures.collect_schema().names()

        # Preferred explicit source: original_maturity_years (years -> days on a
        # 365-day year, matching the oc_short_maturity_threshold_days scalar).
        if "original_maturity_years" in schema_names:
            original_years_available = pl.col("original_maturity_years").is_not_null()
            original_years_days = (
                pl.col("original_maturity_years").cast(pl.Float64, strict=False) * 365.0
            )
        else:
            original_years_available = pl.lit(False)
            original_years_days = pl.lit(None, dtype=pl.Float64)

        # Fallback source: (maturity_date - value_date) in days.
        if "maturity_date" in schema_names and "value_date" in schema_names:
            start_available = (
                pl.col("maturity_date").is_not_null() & pl.col("value_date").is_not_null()
            )
            start_days = (
                (pl.col("maturity_date").cast(pl.Date) - pl.col("value_date").cast(pl.Date))
                .dt.total_days()
                .cast(pl.Float64)
            )
        else:
            start_available = pl.lit(False)
            start_days = pl.lit(None, dtype=pl.Float64)

        original_maturity_days = (
            pl.when(original_years_available).then(original_years_days).otherwise(start_days)
        )
        has_original_source = original_years_available | start_available
        is_short_original = has_original_source & (
            original_maturity_days <= _OC_SHORT_MATURITY_THRESHOLD_DAYS
        )

        return exposures.with_columns(
            pl.when(is_oc & is_short_original)
            .then(pl.lit(_OC_SHORT_MATURITY_CCF))
            .otherwise(pl.col("_sa_ccf_from_risk_type"))
            .alias("_sa_ccf_from_risk_type"),
        )

@cites("CRR Art. 111")
def resolve_provisions(
    exposures: pl.LazyFrame,
    provisions: pl.LazyFrame,
    config: CalculationConfig,
    *,
    pack: ResolvedRulepack | None = None,
) -> pl.LazyFrame:
    """
    Resolve provisions with multi-level beneficiary and drawn-first deduction.

    This is called *before* CCF so that nominal_after_provision feeds into
    the CCF calculation: ``ead_from_ccf = nominal_after_provision * ccf``.

    Resolution levels (based on beneficiary_type):
    1. Direct (loan/exposure/contingent): join on exposure_reference
    2. Facility: join on parent_facility_reference, pro-rata by exposure weight
    3. Counterparty: join on counterparty_reference, pro-rata by exposure weight

    SA drawn-first deduction (CRR Art. 111(2)):
    - ``floored_drawn = max(0, drawn_amount)``
    - ``provision_on_drawn = min(provision_allocated, floored_drawn)``
    - ``provision_on_nominal = min(remainder, nominal_amount)``
    - Interest is never reduced by provision.

    IRB/Slotting: provision_on_drawn=0, provision_on_nominal=0 (provisions
    feed into EL shortfall/excess instead). provision_allocated is tracked.

    Args:
        exposures: Exposures with drawn_amount, interest, nominal_amount, approach
        provisions: Provision data with beneficiary_reference, amount,
                    and optionally beneficiary_type
        config: Calculation configuration

    Returns:
        Exposures with provision_allocated, provision_on_drawn,
        provision_on_nominal, provision_deducted, nominal_after_provision
    """
    prov_schema = provisions.collect_schema()
    exp_schema = exposures.collect_schema()
    has_beneficiary_type = "beneficiary_type" in prov_schema.names()
    has_parent_facility = "parent_facility_reference" in exp_schema.names()
    has_risk_type = "risk_type" in exp_schema.names()

    if has_beneficiary_type:
        # S9e: the SA-CCF table used as the pro-rata provision-weighting basis is
        # regime-selected via the cited pack Feature; _resolve_provisions_multi_level
        # (and the sa_ccf_expression it calls) keep their is_basel_3_1 bool param
        # (Option B). The CCF table VALUES stay static data-layer constants.
        resolved_pack = pack if pack is not None else RulepackV0.from_config(config).pack
        exposures = _resolve_provisions_multi_level(
            exposures,
            provisions,
            has_parent_facility,
            has_risk_type,
            resolved_pack.feature("sa_revised_ccf_table"),
        )
    else:
        # Fallback: direct-only join (backward compat)
        provisions_agg = provisions.group_by("beneficiary_reference").agg(
            pl.col("amount").sum().alias("provision_allocated"),
        )
        exposures = exposures.join(
            provisions_agg,
            left_on="exposure_reference",
            right_on="beneficiary_reference",
            how="left",
        ).with_columns(
            pl.col("provision_allocated").fill_null(0.0),
        )

    # --- SA drawn-first deduction; IRB/Slotting: no deduction ---
    is_sa = pl.col("approach") == ApproachType.SA.value

    floored_drawn = pl.col("drawn_amount").clip(lower_bound=0.0)

    # provision_on_drawn: min(allocated, floored_drawn) for SA; 0 for IRB
    provision_on_drawn = (
        pl.when(is_sa)
        .then(pl.min_horizontal("provision_allocated", floored_drawn))
        .otherwise(pl.lit(0.0))
    )

    exposures = exposures.with_columns(
        provision_on_drawn.alias("provision_on_drawn"),
    )

    # provision_on_nominal: min(remaining, nominal) for SA; 0 for IRB
    remaining = (pl.col("provision_allocated") - pl.col("provision_on_drawn")).clip(lower_bound=0.0)
    provision_on_nominal = (
        pl.when(is_sa)
        .then(pl.min_horizontal(remaining, pl.col("nominal_amount")))
        .otherwise(pl.lit(0.0))
    )

    exposures = exposures.with_columns(
        provision_on_nominal.alias("provision_on_nominal"),
    )

    # provision_deducted = on_drawn + on_nominal
    exposures = exposures.with_columns(
        (pl.col("provision_on_drawn") + pl.col("provision_on_nominal")).alias("provision_deducted"),
    )

    # nominal_after_provision for CCF: nominal - provision_on_nominal
    exposures = exposures.with_columns(
        (pl.col("nominal_amount") - pl.col("provision_on_nominal")).alias(
            "nominal_after_provision"
        ),
    )

    return exposures

@cites("CRR Art. 112")
@cites("CRR Art. 123")
@cites("CRR Art. 126")
def _add_exposure_class_applied(lf: pl.LazyFrame) -> pl.LazyFrame:
    """Add ``exposure_class_applied`` — the approach-agnostic applied class.

    The routing ``exposure_class`` records origination + guarantee substitution
    but omits three SA-only applied-treatment movements, so the reconciliation and
    COREP class dimensions previously mis-bucketed those rows (the RWA is correct
    in every case — only the class label was wrong):

    - **SME managed as retail** (CRR Art. 123 / PS1/26 Art. 123A) — a
      corporate-SME row that took the 75% retail risk weight logically belongs
      to the retail class: Art. 122 corporate has no 75% band, so a 75%-weighted
      SME entails retail. The predicate mirrors the SA risk-weight branch exactly
      (``engine/sa/risk_weights.py``) so the reported class tracks the applied RW.
    - **Defaulted** (CRR Art. 112(1)(j) / Art. 127) — a defaulted SA exposure
      belongs to the "Exposures in default" class, which wins over origination
      (PS1/26 Table A2 priority 5). High-risk (Art. 128, Basel 3.1) still outranks
      default (priority 4), so a defaulted high-risk row keeps its class.
    - **Secured by a mortgage on commercial immovable property** (CRR
      Art. 112(1)(i) / Art. 126; PS1/26 Art. 112(1)(i) / Art. 124H-124I) — an
      exposure whose SA risk weight is set by the commercial real-estate branch
      belongs to Art. 112(1)(i), not to the counterparty's class. It reuses the
      dispatcher's own :func:`is_commercial_re_class` predicate, whose
      ``property_type == "commercial"`` limb is precisely what lets a
      ``corporate``-routed exposure take the Art. 126 50% (or Art. 124I
      income-producing) risk weight; sharing one expression is what stops the
      reported class and the applied risk weight from drifting apart again.

    Both frameworks rank the real-estate class the same way and both make the
    protection — not the counterparty — the classifying fact. PS1/26 Art. 112(2)
    Table A2 is explicit ("Where an exposure meets the criteria for more than one
    exposure class it shall be assigned to the exposure class that has the highest
    position in Table A2"): real estate is row (7), retail row (14), corporates
    row (15). COREP Annex II ¶62 gives the CRR twin ranking — "Exposures secured
    by mortgages on immovable property" is rank 6, corporates and retail rank 9 —
    and ¶58 notes class (i) is the one class where "a protection effect is
    intrinsically part of the definition of an exposure class". So the limb sits
    BELOW default/high-risk (which outrank real estate in both rankings) and ABOVE
    the retail limb. Rows already in a real-estate class are left alone: they are
    in Art. 112(1)(i) already, and re-labelling them would only shuffle the
    C 09.01 "of which" sub-rows without changing the class total.

    ¶60 is what makes this a reporting-side overlay rather than a classifier
    change: the prioritisation governs "the assignment of the Original exposure
    pre-conversion factor by exposure classes, without prejudice to the specific
    treatment (risk weight) that each specific exposure shall receive within the
    assigned exposure class". The routing ``exposure_class`` keeps driving
    approach selection, CRM and the risk-weight tables; only the reported class
    moves, so no RWA changes.

    Only SA rows (``approach_applied == "standardised"``) are re-mapped: IRB
    already reclassifies corporate→retail on ``exposure_class`` and reports
    default via a PD override (not a class), slotting keeps SPECIALISED_LENDING,
    and equity keeps EQUITY — so every non-SA approach keeps ``exposure_class``.

    This is a PRE-substitution (obligor-side) class and is applied to guaranteed
    exposures too. A guaranteed exposure is physically split into ``__G_`` /
    ``__REM`` legs (``engine/crm/guarantees.py``) that BOTH carry the obligor's
    origination ``exposure_class`` — the guarantor's class lives only in
    ``post_crm_exposure_class_guaranteed``, which drives the COREP C 07.00
    substitution inflow/outflow. So the guaranteed leg of a defaulted (or
    SME-managed-as-retail) obligor correctly takes the same applied class as its
    remainder: in C 07.00 the whole exposure originates in the obligor's sheet
    ("Exposures in default" / Retail) and the guaranteed portion leaves as an
    outflow. Gating the overlay on ``~is_guaranteed`` would wrongly drop the
    guaranteed portion out of that class and understate it.
    """
    is_sa = pl.col("approach_applied") == ApproachType.SA.value
    upper_class = pl.col("exposure_class").str.to_uppercase()
    is_high_risk = pl.col("exposure_class") == ExposureClass.HIGH_RISK.value
    sme_managed_as_retail = (
        upper_class.str.contains("SME", literal=True)
        & (pl.col("cp_is_managed_as_retail") == True)  # noqa: E712
        & (pl.col("qualifies_as_retail") == True)  # noqa: E712
    )
    # Classes that outrank real estate in BOTH rankings (PS1/26 Table A2 rows
    # (1)-(6); COREP Annex II para 62 ranks 1-5) and so keep their own class even
    # when property-secured. Defaulted is already handled by the limb above.
    outranks_real_estate = pl.col("exposure_class").is_in(
        [
            ExposureClass.HIGH_RISK.value,
            ExposureClass.EQUITY.value,
            ExposureClass.COVERED_BOND.value,
            ExposureClass.DEFAULTED.value,
        ]
    )
    already_real_estate = pl.col("exposure_class").is_in(
        [
            ExposureClass.RETAIL_MORTGAGE.value,
            ExposureClass.RESIDENTIAL_MORTGAGE.value,
            ExposureClass.COMMERCIAL_MORTGAGE.value,
        ]
    )
    commercial_real_estate = (
        is_commercial_re_class(upper_class) & ~outranks_real_estate & ~already_real_estate
    )
    return lf.with_columns(
        pl.when(~is_sa)
        .then(pl.col("exposure_class"))
        # A null is_defaulted falls through the when() (treated as not defaulted),
        # so no fill_null is needed — keep the applied class off origination.
        .when((pl.col("is_defaulted") == True) & ~is_high_risk)  # noqa: E712
        .then(pl.lit(ExposureClass.DEFAULTED.value))
        # Art. 112(1)(i) outranks corporates and retail in both frameworks, so
        # this limb must precede the retail one below.
        .when(commercial_real_estate)
        .then(pl.lit(ExposureClass.COMMERCIAL_MORTGAGE.value))
        .when(sme_managed_as_retail)
        .then(pl.lit(ExposureClass.RETAIL_OTHER.value))
        .otherwise(pl.col("exposure_class"))
        .alias("exposure_class_applied")
    )

@cites("CRR Art. 235")
@cites("CRR Art. 112")
def _add_reporting_projection(lf: pl.LazyFrame) -> pl.LazyFrame:
    """Add the canonical per-leg reporting projection (Phase 7 S2).

    The results frame IS the two-leg substitution ledger — CRM physically splits
    each guaranteed exposure into ``__G_<guarantor>`` guaranteed legs and
    ``__REM`` / ``__REM_FL`` / ``__REM_SEN`` retained legs
    (``engine/crm/guarantees.py``). This projection names that ledger once, on
    the sealed exit, so no downstream consumer re-derives class/approach/method
    or sniffs reference suffixes (COREP, Pillar 3, reconciliation, and the UI
    all read these columns instead of re-picking among the raw twins):

    - ``reporting_class`` — post-substitution class the RWA is bucketed under
      (Art. 235: guarantor class on guaranteed legs) = ``exposure_class_post_crm``.
    - ``reporting_class_origin`` — obligor applied class, uniform across a
      guaranteed exposure's legs (Art. 112/123) = ``exposure_class_applied``.
    - ``reporting_approach`` / ``reporting_approach_origin`` — the post- and
      pre-substitution approach twins (``approach_post_crm`` / ``approach_applied``).
    - ``reporting_country`` / ``reporting_country_origin`` — the post- and
      pre-substitution COUNTRY twins, the geographical mirror of the class pair.
      PS1/26 Annex II §3.4 ¶86 says outright that "CRM techniques with
      substitution effects can change the allocation of an exposure to a
      country", and ¶87 splits the geographical breakdown by column: "original
      exposure pre-conversion factors" reports at the country of residence of
      the IMMEDIATE obligor, "exposure value" and "risk-weighted exposure
      amounts" at the country of residence of the ULTIMATE obligor. So the
      origin twin is the obligor's own ``cp_country_code`` on every leg, and the
      post twin is the guarantor's country on a BENEFICIALLY guaranteed leg —
      gated identically to ``reporting_class`` (see
      :func:`_add_post_crm_reporting_class`), because a guarantee the engine
      DECLINES moves neither the class nor the country. Degrades to the
      obligor's country wherever the guarantor's is unknown, so a run with no
      CRM guarantee sub-step reports one country on both twins.
    - ``reporting_method`` — the STD/FIRB/AIRB/SLOTTING/EQUITY methodology label
      of the post-substitution approach (``method_label_expr`` materialised).
    - ``reporting_leg_role`` — ``guaranteed`` (the ``__G_`` leg,
      ``is_guaranteed=True``), ``retained`` (the ``__REM*`` remainder /
      Art. 234 tranche legs), or ``whole``. COREP C 07.00 substitution
      outflow/inflow reconstruct as two sums over the ``guaranteed`` legs
      grouped by origin vs post-substitution class.
    - ``reporting_on_balance_sheet`` — declared at source from
      ``exposure_type`` (loan -> on; facility/contingent -> off; anything else
      null = excluded from both on- and off-BS template cells). Mirrors the
      production rule in ``reporting/kernel/filters.py`` (``bs_type`` never
      reaches the aggregator, so the exposure-type rule IS today's behaviour).
    - ``reporting_subclass`` / ``reporting_ead`` / ``reporting_rw`` — aliases of
      ``exposure_subclass`` / ``ead_final`` / ``risk_weight``.
    - ``reporting_gross_drawn`` / ``_interest`` / ``_nominal`` / ``_undrawn`` —
      the raw gross carriers (``drawn_amount`` / ``interest`` /
      ``nominal_amount`` / ``undrawn_amount``) clipped at 0. A negative
      drawn/interest is the on-balance netting convention (a deposit under a
      ``netting_agreement_reference``); the raw carriers seal negative, so
      gross-exposure template cells sum these floored twins instead (CRR
      Art. 111 SA / Art. 166 IRB). Nulls stay null. Computed after the CRM
      guarantee split so the floored amounts are leg-consistent.
    - ``reporting_gross_on_bs`` / ``reporting_gross_off_bs`` — the per-side
      floored gross carriers a template's on/off-balance-sheet gross cells sum
      DIRECTLY, independent of ``reporting_on_balance_sheet`` (which stays a
      strict loan/facility/contingent ladder — the unified pipeline emits
      ``facility_undrawn`` for undrawn commitment headroom, a value that ladder
      leaves null, silently dropping the leg from both gross sides while its EAD
      stays in the EAD/RWEA cells). The side rule keys on ``exposure_type``:
      an on-balance credit type (loan/contingent/facility_undrawn) with an
      unknown drawn AND interest stays null (unknown stays unknown), else its
      on-side is the floored drawn + interest (a null component counts as 0);
      the off-side is a contingent's floored nominal, a facility_undrawn's
      floored undrawn (counted exactly ONCE — the two carriers alias the same
      headroom), a loan's true 0.0, else null. The legacy ``"facility"`` alias
      (never emitted by the pipeline but recognised by the on/off-BS
      discriminators and R11-era fixtures) joins the on-side credit types and
      takes the aliased-pair ``max_horizontal(nominal, undrawn)`` off-side, so a
      type the discriminators put on a side always has that side's carrier
      populated. CCR / settlement legs are outside the on/off-BS credit-risk
      gross scope, so both sides are null there (their EAD/RWEA still report).
      CRR Art. 111 SA / Art. 166 IRB.
    - ``reporting_crm_lgd_financial`` / ``_real_estate`` / ``_other_physical`` /
      ``_receivables`` (W5) — the four "CRM techniques taken into account in LGD
      estimates" amounts (COREP C 08.01/02 cols 0180/0190/0200/0210), with the
      METHOD-DEPENDENT basis resolved here, once, so no template re-derives it:
      an AIRB leg reports the ESTIMATED MARKET VALUE, every other leg the
      ADJUSTED value C_i (PS1/26 Annex II p.108 "where exposures are subject to
      the Foundation Collateral Method … the adjusted value of collateral Ci …
      where exposures are subject to the AIRB approach … the estimated market
      value"; CRR Annex II p.101 keys the same split on "where own estimates of
      LGD are (not) used" — CRR Art. 181(1)(e)-(f)). The discriminator is
      ``approach_applied``, NOT the post-substitution twin, because the C 08
      sheets themselves key on ``reporting_approach_origin``: resolving the
      basis on the post twin would report a market-value figure on a sheet
      selected by the origin approach. The financial carrier folds cash on
      deposit in (defect D4): ``collateral_category_expr`` routes cash/deposit
      to its own category ahead of financial, but Art. 197(1)(a) makes it
      eligible financial collateral and col 0180 is where it belongs — the
      Art. 231 waterfall already groups the two together. The fold mirrors
      ``reporting_gross_on_bs``'s two-component convention: a null component
      counts as 0, but BOTH null stays null.
    - ``reporting_ofcp_lgd_cash_deposit`` / ``_life_insurance`` /
      ``reporting_ofcp_substitution`` (RD-8) — plain aliases of the three
      Art. 200(1) "other funded credit protection" amounts the CRM stage has
      ALREADY routed. Whether a leg's protection reports as an Art. 232
      guarantee (col 0060) or under the AIRB LGD Modelling Collateral Method
      (cols 0171/0172) turns on the run-level ``AIRBCollateralMethod``
      election, which never reaches a template — so ``engine/crm/``, holding
      the config and the pack, decides once and the projection adds no logic
      here. The three are mutually exclusive by construction, which is what
      makes the ``{c0170} = {c0171}+{c0172}+{c0173}`` identity and the
      0060/0171-0172 exclusivity structural rather than conventional.
    - ``guarantee_rwa_benefit`` (Phase 7 decision F8, recorded) — the additive
      per-leg Art. 235/236 substitution relief:
      ``ead_final x guarantee_benefit_rw`` = leg EAD x (borrower-basis RW -
      substituted RW). PRE-supporting-factor and PRE-floor by definition (the
      branch snapshots the delta before Art. 501/501a and the portfolio
      floor), isolating the substitution effect; the applied delta already
      folds the double-default override (Art. 153(3)) and the Art. 160(4)
      no-better-than-direct floor, so the benefit ties exactly to the relief
      the engine granted. 0.0 on retained/whole/non-beneficial legs; NULL
      where the substitution machinery never ran (unguaranteed runs, where
      the branch delta column is absent). Slotting legs substitute via
      RWSM (Art. 235(1), fixed 2026-07-12) and carry real benefits on the
      slotting borrower basis.

    Called after the residual multiplier and the output floor so the aliases
    mirror the sealed final values. Per-row post-floor RWA is deliberately NOT
    projected here — the floor is a portfolio-level max and its per-row
    allocation is a recorded-decision slice of its own (Phase 7 plan S5).
    """
    is_retained_leg = pl.col("exposure_reference").str.contains(r"__REM(?:_FL|_SEN)?$")
    leg_role = (
        pl.when(pl.col("is_guaranteed") == True)  # noqa: E712
        .then(pl.lit("guaranteed"))
        .when(is_retained_leg)
        .then(pl.lit("retained"))
        .otherwise(pl.lit("whole"))
    )
    on_balance_sheet = (
        pl.when(pl.col("exposure_type") == "loan")
        .then(pl.lit(True))
        .when(pl.col("exposure_type").is_in(["facility", "contingent"]))
        .then(pl.lit(False))
        .otherwise(pl.lit(None, dtype=pl.Boolean))
    )
    if "guarantee_benefit_rw" in lf.collect_schema().names():
        # Every branch (SA/IRB/slotting) produces the delta on guaranteed
        # runs; non-beneficial legs are clamped to 0.0 at the branch.
        rwa_benefit = pl.col("ead_final") * pl.col("guarantee_benefit_rw")
    else:
        rwa_benefit = pl.lit(None, dtype=pl.Float64)
    # The ¶87 ULTIMATE-obligor country. Read through the same absence-tolerant
    # selector ``_beneficial_gate`` uses rather than a schema branch: an
    # unguaranteed run never joined a guarantor counterparty, so
    # ``guarantor_country_code`` is simply not there and the coalesce yields the
    # typed null the gate then routes to ``otherwise``. An empty string is
    # treated as unknown for the same reason the class twin does it — a joined
    # counterparty row with a blank country is not a country.
    guarantor_country = _optional_country(_GUARANTOR_COUNTRY_COL)
    # The obligor's own country is read through the same selector, not as a bare
    # column: it is a required aggregator-exit column, but the projection is also
    # exercised directly on minimal frames, and a hard read would make the
    # function's input contract wider than the two twins actually need.
    obligor_country = _optional_country("cp_country_code")
    country_post = (
        pl.when(
            (pl.col("is_guaranteed") == True)  # noqa: E712
            & _beneficial_gate()
            & guarantor_country.is_not_null()
            & (guarantor_country != "")
        )
        .then(guarantor_country)
        .otherwise(obligor_country)
    )
    # Per-side floored gross carriers (CRR Art. 111 SA / Art. 166 IRB). See the
    # docstring: on-side = floored drawn + interest for the on-balance credit
    # types (unknown drawn AND interest -> null); off-side = a contingent's
    # nominal, a facility_undrawn's undrawn (once), a loan's true 0.0. CCR /
    # settlement legs fall outside the credit-risk gross scope -> null both
    # sides. sum_horizontal treats a null component as 0 (never fill_null in
    # engine/), and the is_null guard keeps a wholly-unknown on-side null.
    # "facility" is a LEGACY OFF-BS ALIAS (Wave 3 amendment): the production
    # pipeline never emits it, but reporting_on_balance_sheet / filter_off_bs /
    # the c07_bs+c08_bs ladders all put it off-BS, and R11-era unit fixtures use
    # it (off-BS gross in undrawn_amount). A type the discriminators put on a
    # side MUST have that side's carrier populated, so "facility" joins the
    # credit-type list on-side and takes the aliased-pair off-side rule below.
    on_bs_carrier = (
        pl.when(
            pl.col("exposure_type").is_in(["loan", "contingent", "facility_undrawn", "facility"])
        )
        .then(
            pl.when(pl.col("drawn_amount").is_null() & pl.col("interest").is_null())
            .then(pl.lit(None, dtype=pl.Float64))
            .otherwise(
                pl.sum_horizontal(
                    pl.col("drawn_amount").clip(lower_bound=0.0),
                    pl.col("interest").clip(lower_bound=0.0),
                )
            )
        )
        .otherwise(pl.lit(None, dtype=pl.Float64))
    )
    # The method-dependent CRM-in-LGD basis (COREP C 08.01/02 cols 0180-0210).
    # See the docstring: AIRB legs report the estimated market value, every
    # other leg the adjusted value C_i, keyed on the ORIGIN approach because
    # that is the approach the C 08 sheets themselves are selected by.
    crm_lgd_financial, crm_lgd_re, crm_lgd_other_physical, crm_lgd_receivables = _crm_lgd_carriers()
    off_bs_carrier = (
        pl.when(pl.col("exposure_type") == "contingent")
        .then(pl.col("nominal_amount").clip(lower_bound=0.0))
        .when(pl.col("exposure_type") == "facility_undrawn")
        .then(pl.col("undrawn_amount").clip(lower_bound=0.0))
        .when(pl.col("exposure_type") == "loan")
        .then(pl.lit(0.0))
        # Legacy "facility" alias: its off-BS carrier home is ambiguous
        # (nominal or undrawn), which pipeline facility_undrawn rows alias, so
        # max_horizontal counts the pair exactly once. All-null -> null.
        .when(pl.col("exposure_type") == "facility")
        .then(
            pl.max_horizontal(
                pl.col("nominal_amount").clip(lower_bound=0.0),
                pl.col("undrawn_amount").clip(lower_bound=0.0),
            )
        )
        .otherwise(pl.lit(None, dtype=pl.Float64))
    )
    return lf.with_columns(
        pl.col("exposure_class_post_crm").alias("reporting_class"),
        pl.col("exposure_class_applied").alias("reporting_class_origin"),
        pl.col("approach_post_crm").alias("reporting_approach"),
        pl.col("approach_applied").alias("reporting_approach_origin"),
        country_post.alias("reporting_country"),
        obligor_country.alias("reporting_country_origin"),
        method_label_expr("approach_post_crm").alias("reporting_method"),
        leg_role.alias("reporting_leg_role"),
        on_balance_sheet.alias("reporting_on_balance_sheet"),
        pl.col("exposure_subclass").alias("reporting_subclass"),
        pl.col("ead_final").alias("reporting_ead"),
        pl.col("risk_weight").alias("reporting_rw"),
        rwa_benefit.alias("guarantee_rwa_benefit"),
        # Floored gross-exposure carriers (CRR Art. 111 SA / Art. 166 IRB).
        # A negative drawn/interest is the on-balance netting convention (a
        # deposit under a netting_agreement_reference); the EAD path already
        # floors it, but the RAW carriers seal negative and would make a
        # gross-exposure template cell (COREP C 07/C 08, Pillar 3 CR4/5/6/10)
        # report a negative figure. Clip at 0 so gross cells never go negative;
        # nulls stay null (never fill Float nulls to 0.0 — anti-conservative).
        # Computed here, after the CRM guarantee split, so they are leg-consistent.
        pl.col("drawn_amount").clip(lower_bound=0.0).alias("reporting_gross_drawn"),
        pl.col("interest").clip(lower_bound=0.0).alias("reporting_gross_interest"),
        pl.col("nominal_amount").clip(lower_bound=0.0).alias("reporting_gross_nominal"),
        pl.col("undrawn_amount").clip(lower_bound=0.0).alias("reporting_gross_undrawn"),
        on_bs_carrier.alias("reporting_gross_on_bs"),
        off_bs_carrier.alias("reporting_gross_off_bs"),
        # CRM techniques taken into account in LGD estimates, on the
        # method-resolved basis (COREP C 08.01/02 cols 0180/0190/0200/0210).
        crm_lgd_financial.alias("reporting_crm_lgd_financial"),
        crm_lgd_re.alias("reporting_crm_lgd_real_estate"),
        crm_lgd_other_physical.alias("reporting_crm_lgd_other_physical"),
        crm_lgd_receivables.alias("reporting_crm_lgd_receivables"),
        # RD-8: plain aliases of the three already-routed Art. 200(1) amounts.
        _optional_amount("ofcp_lgd_cash_deposit").alias("reporting_ofcp_lgd_cash_deposit"),
        _optional_amount("ofcp_lgd_life_insurance").alias("reporting_ofcp_lgd_life_insurance"),
        _optional_amount("ofcp_substitution_amount").alias("reporting_ofcp_substitution"),
    )

@cites("CRR Art. 112")
def apply_risk_weights(
    lf: pl.LazyFrame,
    config: CalculationConfig,
    *,
    pack: ResolvedRulepack | None = None,
) -> pl.LazyFrame:
    """Look up and apply risk weights based on exposure class.

    Orchestrates the three-phase SA risk weight assignment:
        1. Setup — ensure columns, derive maturity, classify, join CQS table
        2. Framework-specific when/then overrides (CRR vs Basel 3.1)
        3. Cleanup — sovereign floor, defaulted RW blending, drop temp cols

    Branches in the override chain are order-sensitive (first match wins);
    the framework override helpers apply them in the sequence prescribed
    by the regulation.
    """
    exposures, uc, is_domestic_currency, is_uk_domestic = _prepare_risk_weight_lookup(
        lf, config, pack=pack
    )

    resolved_pack = pack if pack is not None else RulepackV0.from_config(config).pack
    if resolved_pack.feature("sa_revised_risk_weight_overrides"):
        exposures = _apply_b31_risk_weight_overrides(
            exposures, uc, is_domestic_currency, is_uk_domestic, config
        )
    else:
        exposures = _apply_crr_risk_weight_overrides(
            exposures, uc, is_domestic_currency, is_uk_domestic
        )

    # Art. 140(2) obligor ST contamination (regime-invariant, post-ladder).
    # ORDER: must precede _apply_defaulted_risk_weight — its unconditional Art. 127
    # overwrite keeps provision-based defaulted RWs; reordering flips them to 150%.
    exposures = _apply_obligor_st_contamination_override(exposures)

    # Art. 121(6) (CRR) / CRE20.22 (Basel 3.1): Sovereign RW floor for
    # FX-denominated unrated institution exposures. Exception:
    # self-liquidating trade items with original maturity <= 1yr.
    exposures = _apply_sovereign_floor_for_institutions(exposures, is_domestic_currency)

    # Art. 127 defaulted risk weight (secured/unsecured split). Runs after
    # the base RW when-chain so defaulted exposures have their non-defaulted
    # base RW available for blending with collateral coverage.
    exposures = _apply_defaulted_risk_weight(exposures, config, pack=resolved_pack)

    # Drop temporary columns used only during risk-weight application.
    schema_names = exposures.collect_schema().names()
    temp_cols = [
        "_lookup_class",
        "_lookup_cqs",
        "_upper_class",
        "_cqs_risk_weight",
        "_sovereign_rw",
        "risk_weight_rw",
    ]
    return exposures.drop([c for c in temp_cols if c in schema_names])

    @cites("CRR Art. 112")
    @cites("CRR Art. 147")
    def classify(
        self,
        data: ResolvedHierarchyBundle,
        config: CalculationConfig,
        *,
        pack: ResolvedRulepack | None = None,
    ) -> ClassifiedExposuresBundle:
        """
        Classify exposures and split by approach.

        Args:
            data: Hierarchy-resolved data from HierarchyResolver
            config: Calculation configuration

        Returns:
            ClassifiedExposuresBundle with exposures split by approach
        """
        resolved_pack = pack if pack is not None else RulepackV0.from_config(config).pack

        # Reads top-to-bottom as a recipe; each helper owns one regulatory
        # concept. See the sibling sub-modules for per-step regulatory
        # references.
        exposures = add_counterparty_attributes(
            data.exposures,
            data.counterparty_lookup.counterparties,
        )
        exposures = join_specialised_lending(exposures, data.specialised_lending)

        # Single schema snapshot — used by the remaining schema-conditional
        # helpers (non-contract scratch columns and the EU-sovereign currency
        # probe) without re-scanning the LazyFrame. Contract columns
        # (hierarchy_exit / cp_lookup_* / raw_model_permissions) need no
        # presence gate — sealed inputs always carry them.
        schema_names = set(exposures.collect_schema().names())

        classification_errors = collect_input_warnings(data, config, pack=resolved_pack)

        classified = derive_independent_flags(exposures, config, schema_names, pack=resolved_pack)
        classified = classify_exposure_subtypes(classified, config, pack=resolved_pack)
        classified = reclassify_corporate_to_retail(
            classified, config, schema_names, pack=resolved_pack
        )
        classified = flag_property_reclassification_candidates(
            classified, config, schema_names, pack=resolved_pack
        )
        classified = sync_irb_exposure_class(classified, pack=resolved_pack)
        # CRR Art. 160(2)/(6): the top-down PD for purchased corporate receivables
        # must land BEFORE assign_approach — the IRB gate is internal_pd non-null,
        # so without it a pool with no obligor PD falls to SA.
        classified = derive_purchased_receivables_pd(classified, config, pack=resolved_pack)

        has_model_permissions = data.model_permissions is not None
        if data.model_permissions is not None:
            classified = resolve_model_permissions(classified, data.model_permissions)

        classified = assign_approach(
            classified,
            config,
            schema_names,
            has_model_permissions=has_model_permissions,
            pack=resolved_pack,
        )
        classified = derive_exposure_subclass(classified, config, pack=resolved_pack)

        # Stage-exit edge (producer-side): the diagnostic emits below run
        # against in-memory data instead of re-executing the upstream lazy
        # plan, and CRMProcessor receives an eager-backed frame. Laziness is
        # strictly intra-stage (migration Phase 1).
        classified = materialise_edge(classified, config, "classifier_exit")

        classification_errors.extend(collect_beel_on_non_defaulted_warnings(classified))
        classification_errors.extend(collect_qrre_gate_demotion_warnings(classified))
        if has_model_permissions:
            classification_errors.extend(emit_model_permission_diagnostics(classified))

        # Producer seal (Phase 3): validates the contract and strips
        # intra-stage scratch (including _model_permission_diagnostic) —
        # pure plan ops over the eager-backed frame. CCR runs carry the
        # SA-CCR provenance columns through, so the contract is selected
        # by the input frame's brand.
        exit_edge = (
            CLASSIFIER_EXIT_CCR_EDGE
            if sealed_edge_of(data.exposures) == "ccr_exit"
            else CLASSIFIER_EXIT_EDGE
        )
        classified = seal(classified, exit_edge)

        return self._build_bundle(classified, data, classification_errors)

@cites("CRR Art. 113")
def calculate_rwa(lf: pl.LazyFrame) -> pl.LazyFrame:
    """Compute pre-factor RWA = EAD x Risk Weight.

    Emits ``rwa_pre_factor`` for downstream supporting-factor scaling.

    References:
    - CRR Art. 113(1)-(5): general rule for SA risk-weighted exposure amounts.
    """
    return lf.with_columns(
        (pl.col("ead_final") * pl.col("risk_weight")).alias("rwa_pre_factor"),
    )

@cites("CRR Art. 113")
def apply_intragroup_zero_rw(
    lf: pl.LazyFrame,
    config: CalculationConfig,
    *,
    pack: ResolvedRulepack | None = None,
) -> pl.LazyFrame:
    """Apply the CRR Art. 113(6) core-UK-group 0% risk weight.

    On an individual-basis run the scope resolver flips
    ``intragroup_zero_rw_eligible`` True for exposures to a core-UK-group entity
    (both the reporting entity and the counterparty inside the same permission
    perimeter). Those rows take the pack's cited 0% final risk weight, applied
    as the LAST risk-weight step — after standard SA assignment and every CRM
    adjustment (FCSM, life-insurance, guarantee substitution, currency mismatch,
    due diligence). The permission is a hard override, not a benefit blended
    against the counterparty risk weight, so it deliberately wins over the
    due-diligence increase that precedes it.

    Keyed on the row's OWN eligibility carrier: a guarantee-split leg of an
    eligible intragroup loan inherits the flag (splits copy row columns), while
    an external loan merely guaranteed BY a group member is untouched —
    Art. 113(6) covers direct exposures to members, not protection from them.

    SA-routed rows only. IRB rows are unaffected: the IRB route to a 0%
    intragroup exposure is Art. 150(1)(e) permanent partial use, which
    reclassifies the exposure to SA upstream, where this override then fires.
    The Feature (enabled under both CRR and Basel 3.1 — PS1/26 retains the
    permission) carries the regime story, so there is no is_crr branch.

    The override fires ONLY on a scoped **individual**-basis run — the one path
    on which the resolver is authoritative over the carrier. This closes the
    user-loadable bypass: ``intragroup_zero_rw_eligible`` is a declared (optional)
    schema column, so an input file could ship it True, but on an unscoped or
    consolidated run the resolver never overwrites it; gating here means a stray
    input True can never win the 0% without the PRA permission. Reading
    ``reporting_entity`` / ``reporting_basis`` is scope gating, not regime
    branching (arch_check check 17 is untouched).

    No-op when the Feature is disabled, the run is not scoped-individual, or the
    carrier column is absent (a direct calculator invocation that never went
    through the scope resolver).
    """
    resolved_pack = pack if pack is not None else RulepackV0.from_config(config).pack
    if not resolved_pack.feature("intragroup_zero_rw"):
        return lf

    # Scope gate (bypass closure): the 0% can only apply where the resolver has
    # authoritatively set the carrier — a scoped individual-basis run.
    if config.reporting_entity is None or config.reporting_basis is not ReportingBasis.INDIVIDUAL:
        return lf

    cols = lf.collect_schema().names()
    if "intragroup_zero_rw_eligible" not in cols:
        return lf

    zero_rw = scalar_value(resolved_pack.scalar_param("intragroup_zero_rw_pct"))
    # A null carrier routes to the ``otherwise`` branch below (unchanged RW),
    # identical to False — no fill needed (and the edge seal already null-fills).
    eligible = pl.col("intragroup_zero_rw_eligible")
    # SA-routed rows only (Art. 150(1)(e) PPU is the IRB route to SA treatment).
    # On the output-floor path the SA pipe runs over the full unified frame, so
    # gate on the approach column to leave IRB / slotting SA-equivalent RWs alone.
    if "approach" in cols:
        eligible = eligible & (pl.col("approach") == ApproachType.SA.value)

    return lf.with_columns(
        pl.when(eligible)
        .then(pl.lit(zero_rw))
        .otherwise(pl.col("risk_weight"))
        .alias("risk_weight"),
    )

@cites("CRR Art. 114")
@cites("CRR Art. 141")
def build_eu_domestic_currency_expr(
    country_col: str,
    currency_col: str | pl.Expr = "currency",
) -> pl.Expr:
    """
    Build a Polars expression that checks if an exposure is to an EU member
    state's central government/central bank denominated in that state's
    domestic currency.

    Uses replace_strict to map country code → domestic currency, then compares
    with the exposure denomination currency.

    Args:
        country_col: Column name containing the ISO country code
        currency_col: Column name (str) or Polars expression for the
            exposure's denomination currency. A string is wrapped in
            ``pl.col(...)``. Callers operating on a post-FX-conversion
            LazyFrame should pass ``denomination_currency_expr(...)`` so the
            original (pre-conversion) currency is compared — not the reporting
            currency.

    Returns:
        Boolean Polars expression: True when country is EU and currency matches
        that country's domestic currency.
    """
    currency_expr = pl.col(currency_col) if isinstance(currency_col, str) else currency_col
    return (
        pl.col(country_col)
        .fill_null("")
        .replace_strict(_EU_COUNTRY_DOMESTIC_CURRENCY, default=None)
        .eq(currency_expr)
    )

@cites("CRR Art. 114")
@cites("CRR Art. 235")
@cites("PS1/26, paragraph 235")
def build_domestic_cgcb_guarantor_expr(
    country_col: str,
    currency_col: str | pl.Expr,
    funding_currency_col: str | pl.Expr | None = None,
) -> pl.Expr:
    """
    Build a Polars expression that identifies a domestic-currency CGCB guarantor
    under CRR Art. 114(4) and Art. 114(7) (Basel 3.1 preservation).

    Combines the UK (GB/GBP) and EU (member state / member-state-domestic-currency)
    branches into a single boolean expression.

    Callers pass the guarantor's country code column and the currency column to
    test against. For guarantee substitution (Art. 215-217) the currency column
    should be the **guarantee** currency — the Art. 233(3) 8% FX haircut handles
    any mismatch between the guarantee and the underlying exposure separately.

    Art. 235(3) funding limb: the Art. 114(4)/(7) 0% extension to a centrally-
    guaranteed exposure requires the exposure to be BOTH denominated in the
    guarantor's domestic currency (the ``currency_col`` limb) AND *funded* in
    that same currency. When ``funding_currency_col`` is supplied, the limb
    ``funding == currency`` is ANDed in — because ``currency`` has already passed
    the domestic-currency test, equality with it is equivalent to "funded in the
    domestic currency", and holds uniformly across the UK/GBP and EU branches.
    When it is None (the frame carries no funding source) the funding limb is
    omitted, preserving the pure-denomination behaviour. Callers should pass a
    null-PERMISSIVE funding expression (see :func:`funding_currency_expr`) so an
    unreported funding currency reuses the denomination and keeps the exposure's
    existing 0% treatment.

    Args:
        country_col: Column name containing the guarantor's ISO country code.
        currency_col: Column name (str) or Polars expression for the currency
            to test against the guarantor's domestic currency.
        funding_currency_col: Column name (str) or Polars expression for the
            exposure's funding currency. When None, the Art. 235(3) funding limb
            is not applied.

    Returns:
        Boolean Polars expression: True when the guarantor is UK CGCB in GBP or
        an EU-member CGCB in that member state's domestic currency, and — when a
        funding currency is supplied — the exposure is funded in that currency.
    """
    currency_expr = pl.col(currency_col) if isinstance(currency_col, str) else currency_col
    is_uk_domestic = (pl.col(country_col).fill_null("") == "GB") & (currency_expr == "GBP")
    is_eu_domestic = build_eu_domestic_currency_expr(country_col, currency_expr)
    denominated_domestic = is_uk_domestic | is_eu_domestic
    if funding_currency_col is None:
        return denominated_domestic
    funding_expr = (
        pl.col(funding_currency_col)
        if isinstance(funding_currency_col, str)
        else funding_currency_col
    )
    return denominated_domestic & funding_expr.eq(currency_expr)

@cites("CRR Art. 114")
@cites("CRR Art. 235")
def funding_currency_expr(schema_names: list[str] | set[str]) -> pl.Expr | None:
    """
    Return the exposure's funding-currency expression for the Art. 235(3) limb.

    The Art. 114(4)/(7) 0% risk weight — and its Art. 235(3) extension to
    centrally-guaranteed exposures — requires the exposure to be BOTH
    denominated AND *funded* in the relevant domestic currency. This helper
    yields the "funded in" currency: an explicit ``funding_currency`` column when
    present, otherwise the exposure's denomination currency as the proxy the
    audit endorses.

    Null-PERMISSIVE: a null ``funding_currency`` falls back to the denomination
    (``denomination_currency_expr``), so a dataset that does not report a
    separate funding currency keeps the treatment it had before this limb existed
    (mirrors the Art. 237(2)(a) original-maturity null fallback). Returns None
    when the frame carries no currency column at all, signalling the caller to
    omit the funding limb entirely.

    Args:
        schema_names: Column names from ``lf.collect_schema().names()``.

    Returns:
        Polars expression yielding the funding currency per row, or None when no
        currency source is available on the frame.
    """
    names = set(schema_names)
    has_denomination = "original_currency" in names or "currency" in names
    if "funding_currency" in names:
        if has_denomination:
            return pl.col("funding_currency").fill_null(denomination_currency_expr(names))
        return pl.col("funding_currency")
    if has_denomination:
        return denomination_currency_expr(names)
    return None

@cites("CRR Art. 114(3)")
@cites("PS1/26, paragraph 114")
def ecb_rw_expr() -> pl.Expr:
    """Art. 114(3): the ECB 0% risk weight, read from the common pack.

    Exposed as an expression builder rather than a module-scope constant so the
    regulatory value stays in the rulepack (arch_check check 5) — the pack-binding
    shim ``crr_risk_weight_tables`` is its only engine-side home.
    """
    return pl.lit(float(ECB_ZERO_RW))

@cites("CRR Art. 114(3)")
@cites("PS1/26, paragraph 114")
def is_ecb_expr() -> pl.Expr:
    """Art. 114(3): identify exposures to the ECB (0% RW, unconditionally).

    ``eq_missing`` returns False rather than null for a null ``cp_entity_type``,
    so no ``fill_null`` is needed and a missing entity type can never be read as
    the ECB.
    """
    return pl.col("cp_entity_type").eq_missing(_ECB_ENTITY_TYPE)

@cites("CRR Art. 114")
@cites("CRR Art. 115")
@cites("CRR Art. 116")
@cites("CRR Art. 117")
@cites("PS1/26, paragraph 117")
@cites("CRR Art. 118")
@cites("CRR Art. 235")
def build_guarantor_rw_expr(
    *,
    exposure_class_col: str,
    entity_type_col: str,
    cqs_col: str,
    country_code_col: str,
    ccp_client_cleared_col: str,
    scra_grade_col: str,
    is_basel_3_1: bool,
    domestic_cgcb_expr: pl.Expr | None = None,
    short_term_flag_col: str | None = None,
    no_guarantee_expr: pl.Expr | None = None,
) -> pl.Expr:
    """Build the full when/then chain that maps a guarantor to its SA RW.

    Dispatches on the guarantor's SA exposure class (derived from
    ``ENTITY_TYPE_TO_SA_CLASS`` by the caller — e.g. the CRM processor)
    rather than regex on entity_type, ensuring all valid entity types are
    covered. Reproduces the SA-side reference chain
    (``engine/sa/rw_adjustments.py::_build_guarantor_rw_expr``) branch-for-branch.

    Branch order (first match wins):
        no guarantee -> null (only when ``no_guarantee_expr`` is supplied)
        domestic CGCB sovereign (Art. 114(4)/(7)) -> 0%
        CGCB CQS table (Art. 114 Table 1)
        CCP (CRR Art. 306, CRE54.14-15)
        International Organisation (Art. 118) -> 0%
        Named MDB (Art. 117(2)) -> 0%
        Non-named MDB (Art. 117(1)) — PS1/26 Table 2B under Basel 3.1;
            institution treatment (Art. 120 Table 3 / Art. 121 unrated,
            no short-term preferential) under CRR
        Institution (ECRA / SCRA via build_institution_guarantor_rw_expr —
            short-term Art. 120(2) Table 4 when ``short_term_flag_col``
            evaluates True, otherwise long-term Table 3)
        PSE (Art. 116(2) Table 2A, sovereign-derived for unrated)
        RGLA (Art. 115(1)(b) Table 1B, sovereign-derived for unrated)
        Corporate (Art. 122 corporate CQS table)
        else -> null (no substitution)

    The unrated PSE / RGLA fallback is the documented SA-side approximation
    (no guarantor sovereign-CQS join exists in the CRM column production):
    a GB guarantor receives the 20% RGLA / PSE domestic-currency treatment,
    any other country the conservative 100% unrated default — NOT the full
    Art. 116(1) Table 2 / Art. 115(1)(a) Table 1A sovereign-derived lookup.

    Args:
        exposure_class_col: Name of the guarantor SA exposure-class column
            (e.g. ``guarantor_exposure_class``).
        entity_type_col: Name of the guarantor entity-type column — used for
            the CCP override and the named-MDB (``mdb_named``) carve-out.
        cqs_col: Name of the integer guarantor CQS column.
        country_code_col: Name of the guarantor country-code column — drives
            the unrated PSE / RGLA GB-vs-other approximation.
        ccp_client_cleared_col: Name of the Boolean client-cleared flag
            column (null -> proprietary 2%).
        scra_grade_col: Name of the guarantor SCRA-grade column threaded
            into ``build_institution_guarantor_rw_expr`` for the B31 unrated
            institution dispatch.
        is_basel_3_1: Select PS1/26 tables (institution ECRA / corporate
            Table 6 / MDB Table 2B) when True, CRR tables when False — for
            MDBs that means the Art. 117(1) institution treatment, since CRR
            has no MDB table. PSE / RGLA / IO / CCP values are
            framework-identical. Threaded by both live call sites from the
            cited ``sa_revised_risk_weight_tables`` pack Feature.
        domestic_cgcb_expr: Caller-supplied Art. 114(4)/(7) domestic-currency
            test (SA and IRB derive domesticity differently). ``None``
            disables the domestic 0% branch (treated as never-domestic).
        short_term_flag_col: Optional Boolean column routing institution
            guarantors to the Art. 120(2) Table 4 short-term dicts. The IRB
            chain passes ``None`` today.
        no_guarantee_expr: Caller-owned leading guard — rows where it
            evaluates True yield null (no substitution priced). ``None``
            omits the guard (the chain prices every row).

    Returns:
        Float64 Polars expression evaluating to the guarantor's SA RW, or
        null where no substitution treatment exists.
    """
    gec = pl.col(exposure_class_col).fill_null("")
    # PSE/RGLA Art. 116(2)/115(1)(b) unrated fallback: domestic-GB guarantors
    # get the RGLA/PSE 20% domestic-currency treatment; otherwise the
    # conservative 100% PSE/RGLA unrated default applies.
    sovereign_derived_unrated = _pse_rgla_unrated_fallback_expr(country_code_col)

    cgcb_unrated = float(_CGCB_RW[CQS.UNRATED])

    is_domestic_guarantor = domestic_cgcb_expr if domestic_cgcb_expr is not None else pl.lit(False)
    skip_substitution = no_guarantee_expr if no_guarantee_expr is not None else pl.lit(False)

    return (
        pl.when(skip_substitution)
        .then(pl.lit(None).cast(pl.Float64))
        # Art. 114(4)/(7): Domestic sovereign -> 0% regardless of CQS.
        .when((gec == "central_govt_central_bank") & is_domestic_guarantor)
        .then(pl.lit(0.0))
        # CGCB guarantors via CQS (Table 1 — sovereign weights).
        .when(gec == "central_govt_central_bank")
        .then(
            _cqs_table_lookup_expr(
                cqs_col,
                _CGCB_RW,
                cgcb_unrated,
            )
        )
        # CCP guarantors: 2% proprietary / 4% client-cleared
        # (CRR Art. 306, CRE54.14-15) — overrides institution CQS weights.
        .when(pl.col(entity_type_col) == "ccp")
        .then(
            pl.when(pl.col(ccp_client_cleared_col).fill_null(False))
            .then(pl.lit(_QCCP_CLIENT_CLEARED_RW))
            .otherwise(pl.lit(_QCCP_PROPRIETARY_RW))
        )
        # International Organisation (Art. 118): 0% unconditional.
        .when(gec == "international_organisation")
        .then(pl.lit(float(_IO_ZERO_RW)))
        # Named MDB (Art. 117(2)): 0% unconditional.
        .when((gec == "mdb") & (pl.col(entity_type_col).fill_null("") == "mdb_named"))
        .then(pl.lit(float(_MDB_NAMED_ZERO_RW)))
        # Rated / unrated non-named MDB (Art. 117(1)) — framework-divergent:
        #   PS1/26 Art. 117(1)(a)/(b): the dedicated Basel 3.1 MDB Table 2B
        #     (CQS2 30%, unrated 50%).
        #   CRR Art. 117(1): non-named MDBs "shall be treated in the same manner
        #     as exposures to institutions" — Art. 120 Table 3 when rated, the
        #     Art. 121 unrated institution fallback (100%) otherwise. There is no
        #     MDB table in CRR. The Art. 119(2)/120(2)/121(3) short-term
        #     preferential "shall not be applied", so ``short_term_flag_col`` is
        #     deliberately NOT threaded into this branch (unlike the institution
        #     branch below). Mirrors the direct, non-guarantor CRR MDB path in
        #     ``sa/risk_weights.py::_apply_crr_risk_weight_overrides`` (P1.253).
        .when(gec == "mdb")
        .then(
            _cqs_table_lookup_expr(
                cqs_col,
                _MDB_RW,
                float(_MDB_UNRATED_RW),
            )
            if is_basel_3_1
            else build_institution_guarantor_rw_expr(cqs_col, is_basel_3_1=False)
        )
        # Institution guarantors — RW driven from institution_rw_crr /
        # institution_rw_b31_ecra so the pack remains the single source of
        # truth. When the short-term flag evaluates True (CRR/PS1/26
        # Art. 120(2)), the short-term Table 4 dicts apply instead.
        .when(gec == "institution")
        .then(
            build_institution_guarantor_rw_expr(
                cqs_col,
                is_basel_3_1,
                short_term_flag_col=short_term_flag_col,
                scra_grade_col=scra_grade_col,
            )
        )
        # PSE guarantors — Art. 116(2) Table 2A for rated, sovereign-derived for unrated.
        .when(gec == "pse")
        .then(
            _cqs_table_lookup_expr(
                cqs_col,
                _PSE_OWN_RW,
                sovereign_derived_unrated,
            )
        )
        # RGLA guarantors — Art. 115(1)(b) Table 1B for rated, sovereign-derived for unrated.
        .when(gec == "rgla")
        .then(
            _cqs_table_lookup_expr(
                cqs_col,
                _RGLA_OWN_RW,
                sovereign_derived_unrated,
            )
        )
        # Corporate guarantors — Art. 122 corporate CQS table.
        # Basel 3.1 (PRA PS1/26 Art. 122(2) Table 6): CQS3 = 75% (CRR: 100%);
        # PRA retains CQS5 = 150%. Gated on framework so CRR runs are
        # unchanged.
        .when(gec.is_in(["corporate", "corporate_sme"]))
        .then(build_corporate_guarantor_rw_expr(cqs_col, is_basel_3_1))
        .otherwise(pl.lit(None).cast(pl.Float64))
    )

@cites("CRR Art. 114")
@cites("CRR Art. 115")
@cites("CRR Art. 116")
@cites("CRR Art. 117")
@cites("CRR Art. 118")
@cites("CRR Art. 122")
@cites("CRR Art. 123")
def build_entity_rw_expr(
    *,
    entity_type_col: str,
    cqs_col: str,
    is_basel_3_1: bool,
    country_code_col: str | None = None,
) -> pl.Expr:
    """Build the entity-level SA risk-weight preview expression.

    Compiled by the hierarchy facility-share selection
    (``engine/stages/hierarchy/facility_undrawn.py::
    _derive_facility_share_counterparty``) to rank candidate counterparties
    by SA-equivalent risk weight. The preview is non-binding: the chosen
    counterparty still flows through the full classifier and SA/IRB pipeline
    downstream. Keeping the preview SA-only avoids a circular dependency with
    the classifier's IRB approach gating.

    Routes the lowercased ``entity_type`` through the SA exposure-class
    buckets (``ENTITY_TYPES_BY_SA_CLASS``) and maps CQS -> RW via the same
    table branches as :func:`build_guarantor_rw_expr`:

        sovereign (CGCB CQS Table 1, Art. 114)
        International Organisation (Art. 118) -> 0%
        Named MDB (Art. 117(2)) -> 0%
        MDB Table 2B (Art. 117(1))
        Institution (Art. 120 Table 3 / PS1/26 ECRA via
            ``build_institution_guarantor_rw_expr``)
        PSE (Art. 116(2) Table 2A, GB/other approximation for unrated)
        RGLA (Art. 115(1)(b) Table 1B, GB/other approximation for unrated)
        Corporate + covered bond (Art. 122 CRR Table 5 — see note below)
        Retail (Art. 123 flat 75%)
        High risk (Art. 128 flat 150%)
        else -> 1.0 (conservative preview default for unmatched entity
            types, e.g. equity / other items)

    Branch-parity notes (the pre-existing preview branches are preserved
    value-for-value):

    - The corporate branch always prices from ``corporate_risk_weights``
      (CRR Art. 122 Table 5), NOT the Basel 3.1 Table 6 — matching the
      historical preview. Covered bonds use the corporate-equivalent CQS RWs
      in the preview; the precise covered-bond table only applies in real
      SA pricing.
    - The unrated PSE / RGLA fallback is the documented SA-side
      approximation (see :func:`build_guarantor_rw_expr`): GB -> 20%
      domestic-currency treatment, other / unknown country -> 100% unrated
      default. When ``country_code_col`` is ``None`` the 100% default
      applies unconditionally.

    Args:
        entity_type_col: Name of the entity-type column. Null-filled to ""
            and lowercased before bucket routing.
        cqs_col: Name of the integer CQS column; null / out-of-range values
            fall to each table's unrated default.
        is_basel_3_1: Select the PS1/26 institution ECRA table when True,
            CRR Art. 120 Table 3 when False. The remaining preview branches are
            framework-identical: the corporate branch by name (see note above), and
            the MDB / CGCB / IO / PSE / RGLA branches because their pack tables carry
            identical values in both regimes.
        country_code_col: Optional name of the country-code column driving
            the unrated PSE / RGLA GB-vs-other approximation. ``None`` falls
            back to the conservative 100% unrated default.

    Returns:
        Float64 Polars expression evaluating to the entity's SA-equivalent
        preview risk weight (never null — unmatched entity types yield 1.0).
    """
    et = pl.col(entity_type_col).fill_null("").str.to_lowercase()

    sovereign_types = list(ENTITY_TYPES_BY_SA_CLASS[ExposureClass.CENTRAL_GOVT_CENTRAL_BANK.value])
    io_types = list(ENTITY_TYPES_BY_SA_CLASS[ExposureClass.INTERNATIONAL_ORGANISATION.value])
    mdb_types = list(ENTITY_TYPES_BY_SA_CLASS[ExposureClass.MDB.value])
    institution_types = list(ENTITY_TYPES_BY_SA_CLASS[ExposureClass.INSTITUTION.value])
    pse_types = list(ENTITY_TYPES_BY_SA_CLASS[ExposureClass.PSE.value])
    rgla_types = list(ENTITY_TYPES_BY_SA_CLASS[ExposureClass.RGLA.value])
    corporate_types = list(ENTITY_TYPES_BY_SA_CLASS[ExposureClass.CORPORATE.value])
    covered_bond_types = list(ENTITY_TYPES_BY_SA_CLASS[ExposureClass.COVERED_BOND.value])
    retail_types = list(ENTITY_TYPES_BY_SA_CLASS[ExposureClass.RETAIL_OTHER.value])
    high_risk_types = list(ENTITY_TYPES_BY_SA_CLASS[ExposureClass.HIGH_RISK.value])

    unrated_pse_rgla = _pse_rgla_unrated_fallback_expr(country_code_col)

    return (
        # CGCB (Art. 114 Table 1 — sovereign weights).
        pl.when(et.is_in(sovereign_types))
        .then(
            _cqs_table_lookup_expr(
                cqs_col,
                _CGCB_RW,
                float(_CGCB_RW[CQS.UNRATED]),
            )
        )
        # International Organisation (Art. 118): 0% unconditional.
        .when(et.is_in(io_types))
        .then(pl.lit(float(_IO_ZERO_RW)))
        # Named MDB (Art. 117(2)): 0% unconditional — carved out ahead of Table 2B.
        .when(et == "mdb_named")
        .then(pl.lit(float(_MDB_NAMED_ZERO_RW)))
        # Rated / unrated non-named MDB — Table 2B (CRR Art. 117(1) / PS1/26 Art. 117(1)(a)-(b)).
        .when(et.is_in(mdb_types))
        .then(_cqs_table_lookup_expr(cqs_col, _MDB_RW, float(_MDB_UNRATED_RW)))
        # Institution — Art. 120 Table 3 / PS1/26 ECRA via the shared builder
        # so the pack remains the single source of truth.
        .when(et.is_in(institution_types))
        .then(build_institution_guarantor_rw_expr(cqs_col, is_basel_3_1))
        # PSE — Art. 116(2) Table 2A for rated, GB/other approximation for unrated.
        .when(et.is_in(pse_types))
        .then(_cqs_table_lookup_expr(cqs_col, _PSE_OWN_RW, unrated_pse_rgla))
        # RGLA — Art. 115(1)(b) Table 1B for rated, GB/other approximation for unrated.
        .when(et.is_in(rgla_types))
        .then(_cqs_table_lookup_expr(cqs_col, _RGLA_OWN_RW, unrated_pse_rgla))
        # Corporate + covered bond — CRR Art. 122 Table 5 (preview parity:
        # not framework-switched; see docstring).
        .when(et.is_in(corporate_types + covered_bond_types))
        .then(
            _cqs_table_lookup_expr(
                cqs_col,
                _CORPORATE_RW,
                float(_CORPORATE_RW[CQS.UNRATED]),
            )
        )
        # Retail (Art. 123): flat 75%.
        .when(et.is_in(retail_types))
        .then(pl.lit(_RETAIL_RISK_WEIGHT))
        # High-risk items (Art. 128): flat 150%.
        .when(et.is_in(high_risk_types))
        .then(pl.lit(_HIGH_RISK_RW))
        # Conservative preview default for unmatched entity types.
        .otherwise(pl.lit(1.0))
    )