Component Overview¶

class LoaderProtocol(Protocol):
    def load(self, path: Path) -> RawDataBundle:
        """Load raw data from the specified path."""
        ...

class ParquetLoader:
    """Load data from Parquet files."""

    def load(self, path: Path) -> RawDataBundle:
        return RawDataBundle(
            counterparties=pl.scan_parquet(path / "counterparties.parquet"),
            facilities=pl.scan_parquet(path / "facilities.parquet"),
            loans=pl.scan_parquet(path / "loans.parquet"),
            contingents=self._load_optional(path / "contingents.parquet"),
            collateral=self._load_optional(path / "collateral.parquet"),
            guarantees=self._load_optional(path / "guarantees.parquet"),
            provisions=self._load_optional(path / "provisions.parquet"),
            ratings=self._load_optional(path / "ratings.parquet"),
            org_mappings=self._load_optional(path / "org_mapping.parquet"),
            lending_mappings=self._load_optional(path / "lending_mapping.parquet"),
            fx_rates=self._load_optional(path / "fx_rates.parquet"),
            facility_mappings=self._load_optional(path / "facility_mapping.parquet"),
            model_permissions=self._load_optional(path / "model_permissions.parquet"),
        )

    def _load_optional(self, path: Path) -> pl.LazyFrame | None:
        return pl.scan_parquet(path) if path.exists() else None

class HierarchyResolverProtocol(Protocol):
    def resolve(
        self,
        raw_data: RawDataBundle,
        config: CalculationConfig
    ) -> ResolvedHierarchyBundle:
        """Resolve hierarchies and inherit attributes."""
        ...

class HierarchyResolver:
    """Resolve counterparty, facility, and lending group hierarchies."""

    def resolve(self, data: RawDataBundle, config: CalculationConfig) -> ResolvedHierarchyBundle:
        # Step 1: Build counterparty hierarchy lookup
        #   → _build_ultimate_parent_lazy() - traverse org_mappings (up to 10 levels)
        #   → _build_rating_inheritance_lazy() - inherit ratings from parent if missing
        #   → Returns CounterpartyLookup (counterparties, parent_mappings,
        #                                  ultimate_parent_mappings, rating_inheritance)

        # Step 2: Unify exposures (loans + contingents + facility undrawn)
        #   → _build_facility_root_lookup() - traverse facility hierarchies
        #   → _calculate_facility_undrawn() - limit minus aggregated drawn amounts
        #   → Combines all exposure types into single LazyFrame

        # Step 2a: Apply FX conversion (exposures + CRM data)

        # Step 2b: Add collateral LTV to exposures

        # Step 3: Calculate residential property coverage

        # Step 4: Calculate lending group totals (retail threshold)

        # Step 5: Add lending group totals to exposures

        return ResolvedHierarchyBundle(...)

class ClassifierProtocol(Protocol):
    def classify(
        self,
        resolved: ResolvedHierarchyBundle,
        config: CalculationConfig
    ) -> ClassifiedExposuresBundle:
        """Classify exposures into regulatory classes."""
        ...

class CRMProcessorProtocol(Protocol):
    def get_crm_unified_bundle(
        self,
        data: ClassifiedExposuresBundle,
        config: CalculationConfig,
    ) -> CRMAdjustedBundle:
        """Apply CRM and return the unified bundle (no approach split)."""
        ...

class CRMProcessor:
    """Process credit risk mitigation (Art. 111(1)(a)-(b) compliant)."""

    def get_crm_unified_bundle(
        self,
        data: ClassifiedExposuresBundle,
        config: CalculationConfig,
    ) -> CRMAdjustedBundle:
        # Step 0: Funded-only two-layer protection look-through (Art. 191A)

        # Step 1: Resolve provisions (before CCF)
        #   SA: drawn-first deduction, remainder reduces nominal
        #   IRB/Slotting: tracked but not deducted
        # Step 2: Apply CCFs (uses nominal_after_provision)
        # Step 3: Initialize EAD waterfall + crm_post_ead checkpoint
        exposures = self._run_ead_pipeline(data, config)

        # Step 4: Apply collateral (3 lookup collects: direct/facility/counterparty)
        #   + misdirected-AIRB diagnostics (CRM006)
        exposures, applied = self._apply_collateral_unified_step(
            exposures, collateral, config, errors
        )

        # Step 5: Apply guarantees (cross-approach CCF substitution),
        #   behind the crm_pre_guarantee_unified checkpoint
        exposures = self._apply_guarantees_step(
            exposures, guarantees, data, config, errors
        )

        # Step 6: Finalize EAD (no provision subtraction — already in ead_pre_crm)
        # Step 7: Audit columns, then the crm_exit stage edge
        return CRMAdjustedBundle(exposures=exposures, crm_errors=errors, ...)

class SACalculatorProtocol(Protocol):
    def calculate_branch(
        self,
        exposures: pl.LazyFrame,
        config: CalculationConfig,
        *,
        errors: list[CalculationError] | None = None,
    ) -> pl.LazyFrame:
        """Calculate SA RWA on pre-filtered SA-only rows."""
        ...

    def calculate_unified(
        self,
        exposures: pl.LazyFrame,
        config: CalculationConfig,
        *,
        errors: list[CalculationError] | None = None,
    ) -> pl.LazyFrame:
        """SA risk weights on the unified frame (B3.1 output floor path)."""
        ...

class SACalculator:
    """Calculate Standardised Approach RWA (thin orchestrator over the plain typed SA transforms)."""

    def calculate_branch(self, exposures, config, *, errors=None):
        if errors is not None:
            self._warn_equity_in_main_table(exposures, errors)  # SA005

        return (
            exposures.pipe(apply_risk_weights, config, pack=pack)
            .pipe(apply_fcsm_rw_substitution, config)
            .pipe(apply_life_insurance_rw_mapping)
            .pipe(apply_guarantee_substitution, config, pack=pack)
            .pipe(apply_currency_mismatch_multiplier, config, pack=pack)
            .pipe(apply_due_diligence_override, config, errors=errors, pack=pack)  # SA004
            .pipe(calculate_rwa)
            .pipe(apply_supporting_factors, config, errors=errors, pack=pack)      # SF001
        )  # + approach_applied / rwa_final standardisation for the aggregator

class IRBCalculatorProtocol(Protocol):
    def calculate_branch(
        self,
        exposures: pl.LazyFrame,
        config: CalculationConfig,
        *,
        errors: list[CalculationError] | None = None,
    ) -> pl.LazyFrame:
        """Calculate IRB RWA on pre-filtered IRB-only rows."""
        ...

class IRBCalculator:
    """Calculate IRB RWA using K formula."""

    def calculate_branch(self, exposures, config, *, errors=None):
        exposures = (
            exposures.pipe(classify_approach, config)             # F-IRB vs A-IRB
            .pipe(apply_firb_lgd, config, pack=resolved_pack)     # Supervisory LGD for F-IRB
            .pipe(prepare_columns, config, pack=resolved_pack)    # Ensure required columns
            .pipe(apply_all_formulas, config, pack=resolved_pack) # Full IRB calculation
            .pipe(apply_post_model_adjustments, config, pack=resolved_pack)
            .pipe(compute_el_shortfall_excess, errors=errors)
            .pipe(apply_guarantee_substitution, config, pack=resolved_pack)
        )
        # Supporting factors (CRR only — Art. 501), then aggregator columns
        exposures = self._apply_supporting_factors(exposures, config, errors=errors)
        return exposures  # + approach_applied / rwa_final / irb_maturity_m

class SlottingCalculatorProtocol(Protocol):
    def calculate_branch(
        self,
        exposures: pl.LazyFrame,
        config: CalculationConfig,
        *,
        errors: list[CalculationError] | None = None,
    ) -> pl.LazyFrame:
        """Calculate Slotting RWA on pre-filtered slotting-only rows."""
        ...

class SlottingCalculator:
    """Calculate Slotting RWA for specialised lending."""

    def calculate_branch(self, exposures, config, *, errors=None):
        exposures = (
            exposures.pipe(prepare_columns, config)
            .pipe(apply_slotting_weights, config, pack=pack)   # Art. 153(5) tables
            .pipe(calculate_rwa)
        )
        # Supporting factors (CRR Art. 501/501a), EL rates + shortfall/excess
        exposures = self._apply_supporting_factors(exposures, config, errors=errors)
        exposures = exposures.pipe(apply_el_rates, config, pack=pack).pipe(
            compute_el_shortfall_excess, errors=errors
        )
        return exposures  # + approach_applied / rwa_final

class EquityCalculatorProtocol(Protocol):
    def get_equity_result_bundle(
        self,
        data: CRMAdjustedBundle,
        config: CalculationConfig,
    ) -> EquityResultBundle:
        """Calculate equity RWA and return as bundle."""
        ...

class EquityCalculator:
    """Calculate equity exposure RWA."""

    def get_equity_result_bundle(
        self,
        data: CRMAdjustedBundle,
        config: CalculationConfig
    ) -> EquityResultBundle:
        approach = self._determine_approach(config)
        exposures = self._prepare_columns(data.equity_exposures, config)

        if approach == "sa":
            exposures = self._apply_equity_weights_sa(exposures, config)
        else:
            exposures = self._apply_equity_weights_irb_simple(exposures, config)

        exposures = self._calculate_rwa(exposures)
        audit = self._build_audit(exposures, approach)

        return EquityResultBundle(
            results=exposures,
            calculation_audit=audit,
            approach=approach,
            errors=[],
        )

class OutputAggregatorProtocol(Protocol):
    def aggregate(
        self,
        sa_results: pl.LazyFrame,
        irb_results: pl.LazyFrame,
        slotting_results: pl.LazyFrame,
        equity_bundle: EquityResultBundle | None,
        config: CalculationConfig,
        securitisation_audit: pl.LazyFrame | None = None,
    ) -> AggregatedResultBundle:
        """Aggregate results and apply final adjustments."""
        ...

class OutputAggregator:
    """Aggregate calculation results."""

    def aggregate(
        self,
        sa_results: pl.LazyFrame,
        irb_results: pl.LazyFrame,
        slotting_results: pl.LazyFrame,
        equity_bundle: EquityResultBundle | None,
        config: CalculationConfig,
        securitisation_audit: pl.LazyFrame | None = None,
    ) -> AggregatedResultBundle:
        # Combine the collected branch frames (+ equity results)
        combined = pl.concat([
            sa_results,
            irb_results,
            slotting_results,
            *( [equity_bundle.results] if equity_bundle else [] ),
        ], how="diagonal_relaxed")

        # Apply output floor — applicability resolves from the rulepack
        # (a cited pack Feature read from the resolved pack), not an inline
        # config.framework branch. The floor logic itself lives in
        # engine/aggregator/_floor.py (apply_floor_with_impact).
        if resolved_pack.feature("output_floor"):
            combined = apply_floor_with_impact(combined, resolved_pack)

        # Calculate totals
        totals = self._calculate_totals(combined)

        return AggregatedResultBundle(
            data=combined,
            total_rwa=totals.rwa,
            sa_rwa=totals.sa_rwa,
            irb_rwa=totals.irb_rwa,
            slotting_rwa=totals.slotting_rwa,
            total_expected_loss=totals.expected_loss,
        )