nirs4all.controllers.models.stacking.crossbranch module

Cross-Branch Stacking Support (Phase 7).

This module provides support for stacking across multiple branches, allowing meta-models to use predictions from models in different preprocessing branches.

Key Features: 1. BranchScope.ALL_BRANCHES support - stack across all branches 2. Feature alignment validation - ensure samples match across branches 3. Branch compatibility detection - identify compatible vs incompatible branches 4. Cross-branch prediction aggregation - combine predictions from multiple branches

Compatibility Matrix: | Branch Type | Cross-Branch Stacking | Notes | |---------------------|———————-|----------------------------------| | Preprocessing | ✅ Supported | Same samples, different features | | Generator (_or_) | ✅ Supported | Same samples, model variants | | Outlier Excluder | ✅ Supported | Same samples, different training | | Sample Partitioner | ❌ Not Supported | Different samples per partition |

Example

>>> validator = CrossBranchValidator(prediction_store)
>>> result = validator.validate_cross_branch_stacking(
...     source_candidates=candidates,
...     context=context
... )
>>> if result.is_compatible:
...     aligned_features = validator.align_branch_features(...)

class nirs4all.controllers.models.stacking.crossbranch.BranchPredictionInfo(branch_id: int, branch_name: str | None, model_names: List[str], sample_indices: Set[int], n_samples: int, n_folds: int, branch_type: BranchType = BranchType.UNKNOWN)[source]

Bases: object

Information about predictions from a specific branch.

branch_id

Unique branch identifier.

Type:: int

branch_name

Human-readable branch name.

Type:: str | None

model_names

List of model names in this branch.

Type:: List[str]

sample_indices

Set of sample indices with predictions.

Type:: Set[int]

n_samples

Number of samples.

Type:: int

n_folds

Number of folds.

Type:: int

branch_type

Type of branching.

Type:: nirs4all.controllers.models.stacking.branch_validator.BranchType

branch_id: int

branch_name: str | None

branch_type: BranchType = 'unknown'

model_names: List[str]

n_folds: int

n_samples: int

sample_indices: Set[int]

class nirs4all.controllers.models.stacking.crossbranch.CrossBranchCompatibility(value)[source]

Bases: Enum

Compatibility level for cross-branch stacking.

COMPATIBLE = 'compatible'

COMPATIBLE_WITH_ALIGNMENT = 'compatible_with_alignment'

INCOMPATIBLE_PARTITIONS = 'incompatible_partitions'

INCOMPATIBLE_SAMPLES = 'incompatible_samples'

NOT_APPLICABLE = 'not_applicable'

class nirs4all.controllers.models.stacking.crossbranch.CrossBranchValidationResult(is_compatible: bool = True, compatibility: ~nirs4all.controllers.models.stacking.crossbranch.CrossBranchCompatibility = CrossBranchCompatibility.NOT_APPLICABLE, branches: ~typing.Dict[int, ~nirs4all.controllers.models.stacking.crossbranch.BranchPredictionInfo] = <factory>, common_samples: ~typing.Set[int] = <factory>, alignment_issues: ~typing.List[str] = <factory>, warnings: ~typing.List[str] = <factory>, errors: ~typing.List[str] = <factory>)[source]

Bases: object

Result of cross-branch stacking validation.

is_compatible

Whether cross-branch stacking is possible.

Type:: bool

compatibility

Detailed compatibility level.

Type:: nirs4all.controllers.models.stacking.crossbranch.CrossBranchCompatibility

branches

Dict of BranchPredictionInfo by branch_id.

Type:: Dict[int, nirs4all.controllers.models.stacking.crossbranch.BranchPredictionInfo]

common_samples

Set of samples present in all branches.

Type:: Set[int]

alignment_issues

List of alignment problems found.

Type:: List[str]

warnings

List of warning messages.

Type:: List[str]

errors

List of error messages.

Type:: List[str]

add_error(message: str) → None[source]: Add an error and mark as incompatible.

add_warning(message: str) → None[source]: Add a warning message.

alignment_issues: List[str]

branches: Dict[int, BranchPredictionInfo]

common_samples: Set[int]

compatibility: CrossBranchCompatibility = 'not_applicable'

errors: List[str]

is_compatible: bool = True

property total_models: int: Total number of models across all branches.

warnings: List[str]

class nirs4all.controllers.models.stacking.crossbranch.CrossBranchValidator(prediction_store: Predictions, log_warnings: bool = True)[source]

Bases: object

Validates and supports cross-branch stacking.

This validator checks that stacking across multiple branches is feasible and provides utilities for aligning predictions from different branches.

prediction_store: Predictions storage.

log_warnings: Whether to emit Python warnings.

align_branch_features(branch_features: Dict[int, ndarray], branch_sample_indices: Dict[int, List[int]], target_sample_indices: List[int]) → Tuple[ndarray, ndarray][source]

Align features from multiple branches to common sample order.

Combines features from different branches into a single feature matrix, aligning samples to a common order.

Parameters:

branch_features – Dict mapping branch_id to feature matrix.
branch_sample_indices – Dict mapping branch_id to sample indices.
target_sample_indices – Target sample order for output.

Returns:

Tuple of (aligned_features, valid_mask).

Raises:

BranchFeatureAlignmentError – If alignment fails.

get_cross_branch_sources(source_candidates: List[ModelCandidate], context: ExecutionContext) → List[ModelCandidate][source]

Get source models from all branches for cross-branch stacking.

Filters and orders candidates for cross-branch stacking, ensuring proper handling of branch-specific models.

Parameters:

source_candidates – All candidate source models.
context – Execution context.

Returns:

Filtered and ordered list of candidates for cross-branch stacking.

validate_cross_branch_stacking(source_candidates: List[ModelCandidate], context: ExecutionContext, dataset: SpectroDataset | None = None) → CrossBranchValidationResult[source]

Validate cross-branch stacking feasibility.

Checks that all branches have compatible samples and predictions can be properly aligned for stacking.

Parameters:

source_candidates – List of candidate source models.
context – Execution context.
dataset – Optional dataset for sample validation.

Returns:

CrossBranchValidationResult with compatibility info.

nirs4all.controllers.models.stacking.crossbranch.validate_all_branches_scope(prediction_store: Predictions, source_candidates: List[ModelCandidate], context: ExecutionContext) → CrossBranchValidationResult[source]

Convenience function for validating BranchScope.ALL_BRANCHES.

Parameters:

prediction_store – Predictions storage.
source_candidates – List of candidate source models.
context – Execution context.

Returns:

CrossBranchValidationResult with compatibility info.