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% SA CCF under Basel 3.1, except where the otherwise-resolved
        CCF is 10% (Row 6 UCC) or 100% (Row 2) — the Row 4(b) carve-out.
        """
        # 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: under CRR, "other commitments" mapped
        # to MR (50%, >1yr) or MLR (20%, <=1yr). The sa_ccf_expression gives OC 50%
        # as the conservative default; override to 20% when remaining maturity <= 1yr.
        if not is_b31:
            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()
            if "maturity_date" in schema_names:
                is_short_maturity = pl.col("maturity_date").is_not_null() & (
                    (
                        pl.col("maturity_date").cast(pl.Date) - pl.lit(config.reporting_date)
                    ).dt.total_days()
                    <= _OC_SHORT_MATURITY_THRESHOLD_DAYS
                )
                exposures = exposures.with_columns(
                    pl.when(is_oc & is_short_maturity)
                    .then(pl.lit(_OC_SHORT_MATURITY_CCF))
                    .otherwise(pl.col("_sa_ccf_from_risk_type"))
                    .alias("_sa_ccf_from_risk_type"),
                )

        # 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%". When
        # the flag is set under Basel 3.1, override the otherwise-resolved SA CCF
        # to the MR / Row 4(b) rate (50%), unless that CCF is already 10% (Row 6
        # UCC) or 100% (Row 2), in which case the carve-out leaves it untouched.
        # No effect under CRR (Table A1 is Basel 3.1 only) — see the gate below.
        if is_b31:
            row_4b_ccf = _SA_CCF_B31_MAP["MR"]
            carve_out_ccfs = (_SA_CCF_B31_MAP["LR"], _SA_CCF_B31_MAP["FR"])
            is_resi_commitment = pl.col("is_uk_residential_mortgage_commitment").fill_null(False)
            sa_not_in_carve_out = ~pl.col("_sa_ccf_from_risk_type").is_in(carve_out_ccfs)
            exposures = exposures.with_columns(
                pl.when(is_resi_commitment & sa_not_in_carve_out)
                .then(pl.lit(row_4b_ccf))
                .otherwise(pl.col("_sa_ccf_from_risk_type"))
                .alias("_sa_ccf_from_risk_type"),
            )

            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:
            airb_ccf = ccf_modelled_expr.fill_null(pl.col("_sa_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")
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")
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 = _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, config)
    else:
        exposures = _apply_crr_risk_weight_overrides(exposures, uc, is_domestic_currency)

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

        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))
        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. 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")
def build_domestic_cgcb_guarantor_expr(
    country_col: str,
    currency_col: str | pl.Expr,
) -> 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.

    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.

    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.
    """
    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)
    return is_uk_domestic | is_eu_domestic

@cites("CRR Art. 114")
@cites("CRR Art. 115")
@cites("CRR Art. 116")
@cites("CRR Art. 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%
        MDB Table 2B (Art. 117(1))
        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) when True, CRR tables when False. PSE / RGLA / MDB /
            IO / CCP values are framework-identical.
        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 — Table 2B (Art. 117(1)).
        .when(gec == "mdb")
        .then(
            _cqs_table_lookup_expr(
                cqs_col,
                _MDB_RW,
                float(_MDB_UNRATED_RW),
            )
        )
        # 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. All other preview branches are
            framework-identical by construction (see corporate note above).
        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 (Art. 117(1)).
        .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))
    )

@cites("CRR Art. 115")
def _create_rgla_df() -> pl.DataFrame:
    """Create RGLA risk weight lookup DataFrame (Art. 115(1)(b), Table 1B, own-rating).

    Rated RGLAs join against this table via their own CQS.
    Unrated RGLAs use sovereign-derived treatment handled in the SA calculator.
    UK devolved govts (0%) and UK local authorities (20%) are overrides in the calculator.
    """
    return _build_cqs_rw_df(
        RGLA_RISK_WEIGHTS_OWN_RATING,
        "RGLA",
        order=_CQS_ORDER_RATED_ONLY,
    )

@cites("CRR Art. 114")
@cites("CRR Art. 115")
@cites("CRR Art. 116")
@cites("CRR Art. 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%
        MDB Table 2B (Art. 117(1))
        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) when True, CRR tables when False. PSE / RGLA / MDB /
            IO / CCP values are framework-identical.
        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 — Table 2B (Art. 117(1)).
        .when(gec == "mdb")
        .then(
            _cqs_table_lookup_expr(
                cqs_col,
                _MDB_RW,
                float(_MDB_UNRATED_RW),
            )
        )
        # 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. All other preview branches are
            framework-identical by construction (see corporate note above).
        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 (Art. 117(1)).
        .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))
    )

@cites("CRR Art. 116")
def _create_pse_df() -> pl.DataFrame:
    """Create PSE risk weight lookup DataFrame (Art. 116(2), Table 2A, own-rating).

    Rated PSEs join against this table via their own CQS.
    Unrated PSEs use sovereign-derived treatment handled in the SA calculator.
    """
    return _build_cqs_rw_df(
        PSE_RISK_WEIGHTS_OWN_RATING,
        "PSE",
        order=_CQS_ORDER_RATED_ONLY,
    )

@cites("CRR Art. 114")
@cites("CRR Art. 115")
@cites("CRR Art. 116")
@cites("CRR Art. 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%
        MDB Table 2B (Art. 117(1))
        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) when True, CRR tables when False. PSE / RGLA / MDB /
            IO / CCP values are framework-identical.
        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 — Table 2B (Art. 117(1)).
        .when(gec == "mdb")
        .then(
            _cqs_table_lookup_expr(
                cqs_col,
                _MDB_RW,
                float(_MDB_UNRATED_RW),
            )
        )
        # 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. All other preview branches are
            framework-identical by construction (see corporate note above).
        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 (Art. 117(1)).
        .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))
    )