Retail Exposures¶

def reclassify_corporate_to_retail(
    exposures: pl.LazyFrame,
    config: CalculationConfig,
    schema_names: set[str],
    *,
    pack: ResolvedRulepack | None = None,
) -> pl.LazyFrame:
    """
    Reclassify qualifying corporates to retail.

    Retail outranks corporate in the exposure class waterfall per
    CRR Art. 147(5) / Basel CRE30.16-17. Corporate exposures are
    reclassified to retail when all of:
    1. Managed as part of a retail pool (is_managed_as_retail=True)
    2. Aggregated exposure < EUR 1m (qualifies_as_retail=True)
    3. Has internally modelled LGD (lgd IS NOT NULL)
    4. Counterparty is SME-sized (CRR Art. 4(1)(128D) — turnover <
       EUR 50m OR balance-sheet total < EUR 43m when turnover null)

    Reclassification is an exposure-class decision, independent of
    approach permissions. The approach (AIRB/FIRB/SA) is determined
    later by _assign_approach using model_permissions.
    """
    resolved_pack = pack if pack is not None else RulepackV0.from_config(config).pack
    is_sme_by_size = is_sme_by_size_expr(config, pack=resolved_pack)

    # Reclassification eligibility expression (inlined — not a column ref)
    reclassification_expr = (
        (
            pl.col("exposure_class").is_in(
                [
                    ExposureClass.CORPORATE.value,
                    ExposureClass.CORPORATE_SME.value,
                ]
            )
        )
        & (pl.col("cp_is_managed_as_retail") == True)  # noqa: E712
        & (pl.col("qualifies_as_retail") == True)  # noqa: E712
        & (pl.col("lgd").is_not_null())
        & is_sme_by_size
    )

    # Has property collateral expression (inlined)
    has_property_expr = _build_has_property_expr(schema_names)

    # Single .with_columns: reclassified_to_retail, has_property_collateral,
    # exposure_class update — all using inlined expressions (not column refs)
    return exposures.with_columns(
        [
            reclassification_expr.alias("reclassified_to_retail"),
            has_property_expr.alias("has_property_collateral"),
            pl.when(reclassification_expr & has_property_expr)
            .then(pl.lit(ExposureClass.RETAIL_MORTGAGE.value))
            .when(reclassification_expr)
            .then(pl.lit(ExposureClass.RETAIL_OTHER.value))
            .otherwise(pl.col("exposure_class"))
            .alias("exposure_class"),
        ]
    )

def enrich_with_property_coverage(
    exposures: pl.LazyFrame,
    collateral: pl.LazyFrame | None,
) -> pl.LazyFrame:
    """
    Add property collateral coverage columns to exposures inline.

    Calculates two separate values per exposure:
    1. residential_collateral_value: For CRR Art. 123(c) retail threshold exclusion
       (only residential property counts for threshold exclusion)
    2. property_collateral_value: For retail_mortgage classification
       (both residential AND commercial property qualify for mortgage treatment)

    Uses .over() window functions for facility/counterparty allocation weights
    instead of group_by + join-back, avoiding plan tree branching.

    Args:
        exposures: Unified exposures with hierarchy metadata
        collateral: Collateral data with property_type and market_value

    Returns:
        Exposures with columns added: residential_collateral_value,
        property_collateral_value, has_facility_property_collateral,
        exposure_for_retail_threshold
    """
    # Per CRR Article 147, "total amount owed" = drawn amount only (not undrawn)
    exposures = exposures.with_columns(
        pl.col("drawn_amount").clip(lower_bound=0.0).alias("total_exposure_amount"),
    )

    required_cols = {
        "beneficiary_reference",
        "collateral_type",
        "market_value",
        "property_type",
    }
    if not has_required_columns(collateral, required_cols):
        return exposures.with_columns(
            [
                pl.lit(0.0).alias("residential_collateral_value"),
                pl.lit(0.0).alias("property_collateral_value"),
                pl.lit(0.0).alias("residential_collateral_value_uncapped"),
                pl.lit(0.0).alias("commercial_collateral_value_uncapped"),
                pl.lit(False).alias("has_facility_property_collateral"),
                pl.lit(False).alias("re_collateral_non_qualifying"),
                pl.col("total_exposure_amount").alias("exposure_for_retail_threshold"),
            ]
        )

    # Single filter for all property collateral; split residential inline
    all_property_collateral = collateral.filter(
        pl.col("collateral_type").str.to_lowercase() == "real_estate"
    )
    # PRA PS1/26 Art. 124(4): a single non-qualifying RE component (Art. 124A
    # failure, e.g. valuation-independence breach) forces the WHOLE mixed-RE
    # exposure to Art. 124J. Track per-beneficiary whether any RE collateral
    # row fails the qualifying test so the classifier can fire the gate.
    # ``is_qualifying_re`` has no schema default — null means "unreported"
    # and is treated as qualifying here (fill_null(True) is a value fill).
    is_non_qualifying_re = pl.col("is_qualifying_re").fill_null(True) == False  # noqa: E712

    # Multi-level linking with .over() allocation weights
    exposures = _join_property_collateral_multi_level(
        exposures,
        all_property_collateral,
        is_non_qualifying_re=is_non_qualifying_re,
    )

    # Fill nulls, then cap at exposure amount and derive threshold.
    # Preserve the UNCAPPED residential / commercial RE collateral values
    # for the loan-splitter: the PRA PS1/26 Art. 124(4) pro-rata split is by
    # raw collateral value (and the 0.55xV cap is also on raw property value),
    # so the per-exposure cap applied below — which exists for the CRR retail
    # threshold — must not distort the split shares.
    exposures = (
        exposures.with_columns(
            [
                pl.col("residential_collateral_value").fill_null(0.0),
                pl.col("property_collateral_value").fill_null(0.0),
                pl.col("re_collateral_non_qualifying").fill_null(False),
            ]
        )
        .with_columns(
            [
                pl.col("residential_collateral_value").alias(
                    "residential_collateral_value_uncapped"
                ),
                (pl.col("property_collateral_value") - pl.col("residential_collateral_value"))
                .clip(lower_bound=0.0)
                .alias("commercial_collateral_value_uncapped"),
            ]
        )
        .with_columns(
            [
                pl.min_horizontal("residential_collateral_value", "total_exposure_amount").alias(
                    "residential_collateral_value"
                ),
                pl.min_horizontal("property_collateral_value", "total_exposure_amount").alias(
                    "property_collateral_value"
                ),
                (
                    pl.col("total_exposure_amount")
                    - pl.min_horizontal("residential_collateral_value", "total_exposure_amount")
                ).alias("exposure_for_retail_threshold"),
            ]
        )
    )

    return exposures