nirs4all.operators.data.merge module

Merge operator configuration for branch and source merging.

This module provides configuration dataclasses and enums for the MergeController, which handles combining branch outputs (features and/or predictions) and exiting branch mode.

The merge operator is the core primitive for all branch combination operations. It provides: - Feature merging from branches (horizontal concatenation) - Prediction merging with OOF reconstruction (data leakage prevention) - Per-branch model selection and aggregation strategies - Mixed merging (features from some branches, predictions from others)

Example

>>> # Simple feature merge
>>> {"merge": "features"}
>>>
>>> # Prediction merge with OOF safety
>>> {"merge": "predictions"}
>>>
>>> # Mixed merge with per-branch control
>>> {"merge": {
...     "predictions": [
...         {"branch": 0, "select": "best", "metric": "rmse"},
...         {"branch": 1, "aggregate": "mean"}
...     ],
...     "features": [2]
... }}

class nirs4all.operators.data.merge.BranchType(value)

Bases: Enum

Type of branch based on sample handling.

Variables:

• COPY – All branches see all samples (default branching behavior).

• METADATA_PARTITIONER – Branches partition samples by metadata column.

• SAMPLE_PARTITIONER – Branches partition samples by filter (e.g., outlier).

COPY = 'copy'

METADATA_PARTITIONER = 'metadata_partitioner'

SAMPLE_PARTITIONER = 'sample_partitioner'

class nirs4all.operators.data.merge.DisjointSelectionCriterion(value)

Bases: Enum

Criterion for selecting top-N models in disjoint branch merge.

When branches have different model counts, we select top-N models from each branch based on this criterion.

Variables:

• MSE – Select by lowest Mean Squared Error (default for regression).

• RMSE – Select by lowest Root Mean Squared Error.

• MAE – Select by lowest Mean Absolute Error.

• R2 – Select by highest R² score.

• ORDER – Select first N in definition order (no ranking).

MAE = 'mae'

MSE = 'mse'

ORDER = 'order'

R2 = 'r2'

RMSE = 'rmse'

class nirs4all.operators.data.merge.MergeConfig(collect_features: bool = False, feature_branches: str | list[int] = 'all', collect_predictions: bool = False, prediction_branches: str | list[int] = 'all', prediction_configs: list[BranchPredictionConfig] | None = None, model_filter: list[str] | None = None, use_proba: bool = False, include_original: bool = False, on_missing: str = 'error', on_shape_mismatch: str = 'error', unsafe: bool = False, output_as: str = 'features', source_names: list[str] | None = None, n_columns: int | None = None, select_by: str = 'mse', is_separation_merge: bool = False, source_merge: SourceMergeConfig | None = None, meta_feature_plan: MetaFeaturePlan | None = None, stacking_fit_contract: StackingFitContract | None = None)

Bases: object

Configuration for branch merging operations.

This dataclass provides complete configuration for the MergeController, controlling what data is collected from branches and how it is combined.

Variables:

• collect_features (bool) – Whether to collect features from branches.

• feature_branches (str | list[int]) – Which branches to collect features from. - “all” (default): All branches - List of branch indices: [0, 2] for specific branches

• collect_predictions (bool) – Whether to collect predictions from branches.

• prediction_branches (str | list[int]) – Legacy simple mode: which branches for predictions. Use prediction_configs for advanced per-branch control.

• prediction_configs (list[nirs4all.operators.data.merge.BranchPredictionConfig] | None) – Advanced per-branch prediction configuration. Takes precedence over prediction_branches when set.

• model_filter (list[str] | None) – Legacy: global model filter (simple mode). List of model names to include.

• use_proba (bool) – Legacy: global proba setting for classification.

• include_original (bool) – Include pre-branch features in merged output. When True, original features are prepended to merged features.

• on_missing (str) – How to handle missing branches or predictions. - “error” (default): Raise an error - “warn”: Log warning and skip - “skip”: Silent skip

• on_shape_mismatch (str) – Reserved for 3D layout feature merging. In 2D layout (default), features are flattened and concatenated horizontally, so different feature dimensions is normal and this parameter has no effect. For future 3D layout support: - “error”: Raise error if processings differ - “allow”: Flatten to 2D and concatenate - “pad”: Pad shorter processings with zeros - “truncate”: Truncate longer to match shortest

• unsafe (bool) – If True, DISABLE OOF reconstruction for predictions. ⚠️ CAUSES DATA LEAKAGE - only for rapid prototyping.

• output_as (str) – Where to put merged output. - “features” (default): Single concatenated feature matrix - “sources”: Each branch becomes a separate source - “dict”: Keep as structured dict for multi-head models

• source_names (list[str] | None) – Custom names for output sources (when output_as=”sources”). If not provided, uses “branch_0”, “branch_1”, etc.

Example

>>> # Simple feature merge
>>> MergeConfig(collect_features=True)
>>>
>>> # Prediction merge with OOF
>>> MergeConfig(collect_predictions=True)
>>>
>>> # Mixed merge with per-branch control
>>> MergeConfig(
...     collect_predictions=True,
...     prediction_configs=[
...         BranchPredictionConfig(branch=0, select="best"),
...         BranchPredictionConfig(branch=1, aggregate="mean")
...     ],
...     collect_features=True,
...     feature_branches=[2]
... )
>>>
>>> # Unsafe mode (with warning)
>>> MergeConfig(collect_predictions=True, unsafe=True)
>>>
>>> # Disjoint branch merge with n_columns override
>>> MergeConfig(
...     collect_predictions=True,
...     n_columns=2,
...     select_by="mse"
... )

__post_init__()

Validate configuration after initialization.

collect_features: bool = False

collect_predictions: bool = False

feature_branches: str | list[int] = 'all'

classmethod from_dict(data: dict[str, Any]) → MergeConfig

Create MergeConfig from a dictionary.

Used for loading merge configuration from manifest in prediction mode.

Parameters:

data – Dictionary representation of merge configuration.

Returns:

MergeConfig instance.

get_feature_branches(n_branches: int) → list[int]

Get list of branch indices to collect features from.

Parameters:

n_branches – Total number of branches available.

Returns:

List of branch indices.

get_merge_mode() → MergeMode

Determine the merge mode based on configuration.

Returns:

MergeMode enum value.

get_prediction_configs(n_branches: int) → list[BranchPredictionConfig]

Get prediction configurations, normalizing legacy format if needed.

Converts legacy simple mode (prediction_branches + model_filter + use_proba) to per-branch configurations for uniform processing.

Parameters:

n_branches – Total number of branches available.

Returns:

List of BranchPredictionConfig for each branch to collect from.

get_selection_criterion() → DisjointSelectionCriterion

Get the selection criterion enum for disjoint branch merging.

Returns:

DisjointSelectionCriterion enum value.

get_shape_mismatch_strategy() → ShapeMismatchStrategy

Get the shape mismatch strategy enum.

Returns:

ShapeMismatchStrategy enum value.

has_per_branch_config() → bool

Check if using advanced per-branch prediction configuration.

Returns:

True if prediction_configs is set and non-empty.

include_original: bool = False

is_separation_merge: bool = False

meta_feature_plan: MetaFeaturePlan | None = None

model_filter: list[str] | None = None

n_columns: int | None = None

on_missing: str = 'error'

on_shape_mismatch: str = 'error'

output_as: str = 'features'

prediction_branches: str | list[int] = 'all'

prediction_configs: list[BranchPredictionConfig] | None = None

select_by: str = 'mse'

source_merge: SourceMergeConfig | None = None

source_names: list[str] | None = None

stacking_fit_contract: StackingFitContract | None = None

to_dict() → dict[str, Any]

Serialize merge configuration to a dictionary.

Used for saving merge configuration to manifest for reproducibility in prediction mode and bundle export.

Returns:

Dictionary representation suitable for YAML/JSON serialization.

unsafe: bool = False

use_proba: bool = False

validate_relation_safety() → None

Refuse relation-aware prediction merges that disable OOF reconstruction.

class nirs4all.operators.data.merge.MergeMode(value)

Bases: Enum

What to merge from branches.

Variables:

• FEATURES – Merge feature matrices from branches.

• PREDICTIONS – Merge model predictions from branches (with OOF reconstruction).

• ALL – Merge both features and predictions from all branches.

ALL = 'all'

FEATURES = 'features'

PREDICTIONS = 'predictions'

class nirs4all.operators.data.merge.MetaFeatureAdapter(value)

Bases: Enum

How branch predictions are adapted to the meta-feature row domain.

BROADCAST = 'broadcast'

DIRECT = 'direct'

REDUCE = 'reduce'

SOURCE_TO_COMBO = 'source_to_combo'

class nirs4all.operators.data.merge.MetaFeaturePlan(meta_row_domain: str = 'sample', adapters: list[str] = <factory>, alignment_key: str = 'physical_sample_id', missing_prediction_policy: str = 'strict', allow_fallbacks: list[str] = <factory>, allow_experimental_combo_meta: bool = False, combo_reduction_plan: dict[str, ~typing.Any] | None=None, fit_influence_policy: dict[str, ~typing.Any] | None=None)

Bases: object

Replayable contract for late-fusion/meta-feature alignment.

adapters: list[str]

alignment_key: str = 'physical_sample_id'

allow_experimental_combo_meta: bool = False

allow_fallbacks: list[str]

combo_reduction_plan: dict[str, Any] | None = None

fit_influence_policy: dict[str, Any] | None = None

classmethod from_dict(data: dict[str, Any]) → MetaFeaturePlan

Create a plan from a manifest dictionary.

meta_row_domain: str = 'sample'

missing_prediction_policy: str = 'strict'

to_dict() → dict[str, Any]

Serialize the plan for workspace/bundle manifests.

class nirs4all.operators.data.merge.MetaRowDomain(value)

Bases: Enum

Unit domain for rows of late-fusion/meta-model features.

COMBO = 'combo'

SAMPLE = 'sample'

class nirs4all.operators.data.merge.MissingPredictionPolicy(value)

Bases: Enum

How late-fusion/stacking handles missing branch predictions.

DROP_BRANCH = 'drop_branch'

DROP_INCOMPLETE = 'drop_incomplete'

IMPUTE_DECLARED = 'impute_declared'

MASK = 'mask'

PAD = 'pad'

PARTIAL_MODEL = 'partial_model'

STRICT = 'strict'

class nirs4all.operators.data.merge.SelectionProtocol(value)

Bases: Enum

How branch/base models are selected for a meta-model.

BRANCH_LEVEL = 'branch_level'

GLOBAL = 'global'

META_AWARE = 'meta_aware'

class nirs4all.operators.data.merge.ShapeMismatchStrategy(value)

Bases: Enum

How to handle shape mismatches during 3D feature merging.

This strategy only applies when using 3D layout for features, where the number of processings must be aligned across branches. In 2D layout (the default), features are simply flattened and concatenated horizontally, so different feature dimensions across branches is expected and normal.

Example

• Branch 0: (200 samples, 500 features) from MinMaxScaler

• Branch 1: (200 samples, 4 processings, 20 features) from multi-processing

In 2D layout: concatenates to (200, 500 + 4*20 = 580) - no error In 3D layout: needs alignment strategy since processings differ

Variables:

• ERROR – Raise an error on shape mismatch (default, strictest).

• ALLOW – Flatten to 2D and concatenate regardless of differences.

• PAD – Pad shorter branches with zeros to match longest processings.

• TRUNCATE – Truncate longer branches to match shortest processings.

ALLOW = 'allow'

ERROR = 'error'

PAD = 'pad'

TRUNCATE = 'truncate'

class nirs4all.operators.data.merge.SourceBranchConfig(source_pipelines: str | dict[str | int, list[~typing.Any]]=<factory>, default_pipeline: list[Any] | None = None, merge_after: bool = True, merge_strategy: str = 'concat')

Bases: object

Configuration for source branching operations.

This dataclass provides configuration for the source_branch keyword, which creates per-source pipeline execution paths. Each source in a multi-source dataset gets its own independent processing pipeline.

Unlike regular branching (branch), which creates parallel paths that all process the same data, source branching assigns each source to a specific processing pipeline based on its name or index.

Variables:

• source_pipelines (str | dict[str | int, list[Any]]) – Mapping of source names/indices to their pipeline steps. - Dict[str, List]: Named sources to steps mapping - Dict[int, List]: Source indices to steps mapping - “auto”: Apply same steps to all sources independently

• default_pipeline (list[Any] | None) – Default pipeline for sources not explicitly specified. Applied when a source is not listed in source_pipelines. If None, unspecified sources are passed through unchanged.

• merge_after (bool) – Whether to automatically merge sources after branching. - True (default): Automatically call merge_sources after - False: Keep sources separate (user must merge manually)

• merge_strategy (str) – Strategy for auto-merge (when merge_after=True). - “concat” (default): Horizontal concatenation - “stack”: Stack along source axis - “dict”: Keep as dictionary

Example

>>> # Different preprocessing per source
>>> {"source_branch": {
...     "NIR": [SNV(), SavitzkyGolay()],
...     "markers": [VarianceThreshold(), MinMaxScaler()],
...     "Raman": [BaselineCorrection(), StandardScaler()]
... }}
>>>
>>> # Source branching with default fallback
>>> {"source_branch": {
...     "NIR": [SNV()],
...     "_default_": [MinMaxScaler()]  # Applied to other sources
... }}
>>>
>>> # Automatic same-preprocessing per source (isolates sources)
>>> {"source_branch": "auto"}
>>>
>>> # Source branching without auto-merge
>>> {"source_branch": {
...     "NIR": [SNV()],
...     "markers": [StandardScaler()],
...     "_merge_after_": False  # Disable auto-merge
... }}

__post_init__()

Validate configuration after initialization.

default_pipeline: list[Any] | None = None

classmethod from_dict(data: dict[str, Any]) → SourceBranchConfig

Create config from dictionary.

Note: This is primarily for metadata reconstruction. The actual pipeline steps must be restored from the manifest/artifacts.

Parameters:

data – Dictionary representation.

Returns:

SourceBranchConfig instance (with placeholder pipelines).

get_all_source_mappings(available_sources: list[str]) → dict[str, list[Any]]

Get pipeline mapping for all available sources.

Parameters:

available_sources – List of available source names.

Returns:

Dict mapping source names to their pipeline steps.

get_pipeline_for_source(source_name: str, source_index: int) → list[Any] | None

Get pipeline steps for a specific source.

Parameters:

• source_name – Name of the source.

• source_index – Index of the source.

Returns:

List of pipeline steps for this source, or None if passthrough.

is_auto_mode() → bool

Check if using automatic source branching.

Returns:

True if source_pipelines is “auto”.

merge_after: bool = True

merge_strategy: str = 'concat'

source_pipelines: str | dict[str | int, list[Any]]

to_dict() → dict[str, Any]

Serialize configuration to dictionary.

Returns:

Dictionary representation for manifest storage.

class nirs4all.operators.data.merge.SourceIncompatibleStrategy(value)

Bases: Enum

How to handle incompatible source shapes during stacking.

When using stack strategy with sources that have different feature dimensions or processing counts, this controls the resolution.

Variables:

• ERROR – Raise an error on incompatible shapes (default, strictest).

• FLATTEN – Force 2D concatenation instead of stacking.

• PAD – Pad shorter sources with zeros to match longest.

• TRUNCATE – Truncate longer sources to match shortest.

ERROR = 'error'

FLATTEN = 'flatten'

PAD = 'pad'

TRUNCATE = 'truncate'

class nirs4all.operators.data.merge.SourceMergeStrategy(value)

Bases: Enum

How to combine features from multiple data sources.

Used by the merge_sources keyword to control how multi-source datasets are unified into a single feature space.

Variables:

• CONCAT – Horizontal concatenation of all source features (default). Results in 2D array: (samples, sum_of_all_source_features). Different feature dimensions per source is expected.

• STACK – Stack sources along a new axis to create 3D tensor. Results in 3D array: (samples, n_sources, n_features). Requires all sources to have the same feature dimension.

• DICT – Keep sources as a structured dictionary. Results in Dict[str, ndarray] for multi-input models. Each source is accessible by name.

CONCAT = 'concat'

DICT = 'dict'

STACK = 'stack'

class nirs4all.operators.data.merge.StackingFitContract(meta_training_features: str = 'oof', inference_features: str = 'refit_base_predictions', selection_protocol: str = 'nested', base_prediction_calibration: str = 'none', model_selection_protocol: str = 'branch_level', final_validation_profile: bool = True)

Bases: object

Contract preventing in-sample late-fusion/meta-model leakage.

base_prediction_calibration: str = 'none'

final_validation_profile: bool = True

classmethod from_dict(data: dict[str, Any]) → StackingFitContract

Create a contract from a manifest dictionary.

inference_features: str = 'refit_base_predictions'

meta_training_features: str = 'oof'

model_selection_protocol: str = 'branch_level'

selection_protocol: str = 'nested'

to_dict() → dict[str, Any]

Serialize the contract for workspace/bundle manifests.

class nirs4all.operators.data.merge.StackingSelectionProtocol(value)

Bases: Enum

How final validation / selection data is obtained for stacking.

HOLDOUT = 'holdout'

NESTED = 'nested'

REUSE_OOF = 'reuse_oof'