Retail Exposures¶

        Compute all flags that depend only on raw input columns.

        Uses two .with_columns() batches: the first pre-computes shared
        intermediates (uppercase strings, entity-type mapping) that the
        second batch references, avoiding redundant str.to_uppercase()
        and replace_strict() calls.

        Sets: exposure_class_sa, exposure_class_irb, exposure_class, is_mortgage,
              is_defaulted, exposure_class_for_sa, is_infrastructure,
              qualifies_as_retail, retail_threshold_exclusion_applied
        """
        max_retail_exposure = float(config.retail_thresholds.max_exposure_threshold)

        # SL override: exposures with sl_type (from specialised_lending join) get
        # SPECIALISED_LENDING class regardless of counterparty entity_type.
        sl_override = pl.col("sl_type").is_not_null()

        # Batch 1: Pre-compute shared intermediates to avoid redundant work.
        # - _sa_class: entity type → SA class mapping (used 3× below)
        # - _irb_class: entity type → IRB class mapping
        # - _pt_upper: product_type uppercased (used in is_mortgage, infrastructure)
        exposures = exposures.with_columns(
            [
                pl.col("cp_entity_type")
                .replace_strict(ENTITY_TYPE_TO_SA_CLASS, default=ExposureClass.OTHER.value)
                .alias("_sa_class"),
                pl.col("cp_entity_type")
                .replace_strict(ENTITY_TYPE_TO_IRB_CLASS, default=ExposureClass.OTHER.value)
                .alias("_irb_class"),
                pl.col("product_type").str.to_uppercase().alias("_pt_upper"),
            ]
        )

        sl_class = pl.lit(ExposureClass.SPECIALISED_LENDING.value)

        # Batch 2: Derive all flags from pre-computed intermediates.
        return exposures.with_columns(
            [
                # --- Exposure class mappings (SL table overrides entity_type) ---
                pl.when(sl_override)
                .then(sl_class)
                .otherwise(pl.col("_sa_class"))
                .alias("exposure_class_sa"),
                pl.when(sl_override)
                .then(sl_class)
                .otherwise(pl.col("_irb_class"))
                .alias("exposure_class_irb"),
                pl.when(sl_override)
                .then(sl_class)
                .otherwise(pl.col("_sa_class"))
                .alias("exposure_class"),
                # --- Mortgage flag ---
                self._build_is_mortgage_expr(schema_names),
                # --- Default flags ---
                (pl.col("cp_default_status") == True)  # noqa: E712
                .alias("is_defaulted"),
                pl.when(pl.col("cp_default_status") == True)  # noqa: E712
                .then(pl.lit(ExposureClass.DEFAULTED.value))
                .when(sl_override)

                contingent_ref_col = "contingent_reference"

        if contingent_ref_col is None:
            contingent_totals = pl.LazyFrame(
                schema={
                    "aggregation_facility": pl.String,
                    "total_contingent": pl.Float64,
                }
            )
        else:
            # Filter mappings to only contingent children
            if type_col is not None:
                contingent_mappings = facility_mappings.filter(
                    pl.col(type_col).fill_null("").str.to_lowercase() == "contingent"
                ).unique(subset=["child_reference", "parent_facility_reference"])
            else:
                contingent_mappings = facility_mappings.unique(
                    subset=["child_reference", "parent_facility_reference"]
                )

            # Join contingents with their parent facility
            contingent_with_parent = contingents.join(
                contingent_mappings,
                left_on="contingent_reference",
                right_on="child_reference",
                how="inner",
            )

            contingent_with_parent = _resolve_to_root_facility(contingent_with_parent, root_lookup)

            contingent_totals = contingent_with_parent.group_by("aggregation_facility").agg(
                [
                    pl.col("nominal_amount").clip(lower_bound=0.0).sum().alias("total_contingent"),
                ]
            )

        # Identify sub-facilities to exclude from output
        # Sub-facilities appear as child_reference with child_type="facility"
        # Anti-join with empty frame naturally returns all rows
        sub_facility_refs = root_lookup.select(
            pl.col("child_facility_reference").alias("_sub_ref"),
        )

        # Join with facilities to calculate undrawn
        # Combine loan drawn + contingent utilisation
        facility_with_drawn = (
            facilities.join(
                loan_drawn_totals,
                left_on="facility_reference",
                right_on="aggregation_facility",
                how="left",
            )
            .join(
                contingent_totals,
                left_on="facility_reference",
                right_on="aggregation_facility",
                how="left",
            )
            .with_columns(
                [
                    pl.col("total_drawn").fill_null(0.0),
                    pl.col("total_contingent").fill_null(0.0),
                ]
            )
            .with_columns(
                [
                    (pl.col("total_drawn") + pl.col("total_contingent")).alias("total_utilised"),
                    (pl.col("limit") - (pl.col("total_drawn") + pl.col("total_contingent")))
                    .clip(lower_bound=0.0)
                    .alias("undrawn_amount"),
                ]
            )
        )

        # Exclude sub-facilities: only root/standalone facilities produce undrawn exposures
        facility_with_drawn = facility_with_drawn.join(
            sub_facility_refs,
            left_on="facility_reference",
            right_on="_sub_ref",
            how="anti",
        )

        # Get facility schema to check for optional columns
        facility_schema = facilities.collect_schema()
        facility_cols = set(facility_schema.names())

        # Build select expressions with defaults for missing columns
        # Note: parent_facility_reference is set to the source facility to enable
        # facility-level collateral allocation to undrawn amounts
        select_exprs = [
            (pl.col("facility_reference") + "_UNDRAWN").alias("exposure_reference"),
            pl.lit("facility_undrawn").alias("exposure_type"),
            pl.col("product_type")
            if "product_type" in facility_cols
            else pl.lit(None).cast(pl.String).alias("product_type"),
            pl.col("book_code").cast(pl.String, strict=False)
            if "book_code" in facility_cols
            else pl.lit(None).cast(pl.String).alias("book_code"),