data_preprocessing.data_preprocessor

Generic class for building a preprocessing pipeline.

class DataPreprocessor(steps=None)

Bases: Pipeline, SaveLoadMixin

A configurable data preprocessing pipeline for tabular data.

A data preprocessing pipeline that allows for configurable steps based on the extended pipeline.

When no steps are provided, a default pipeline is created which removes features that are constant or binary and contain more 5% missing values. Afterward, remaining missing values are imputed with the mean and the features are scaled with the StandardScaler.

Parameters:

steps (Optional[List[Dict[str, Any]]]) –

Optional list of step specifications. Each item is a dict with:

• name: registered step name (see STEP_REGISTRY).

• enabled: optional bool (default True).

• params: dict of constructor arguments for the step.

• step_name: optional explicit pipeline name (defaults to name).

Notes

Enforced ordering in steps mode:

1. NaN introducing steps first (DuplicateValuesToNan, CounterDiffTransformer),

2. ColumnSelector (if present),

3. Other steps

4. Imputer placed before scaler (always present; mean strategy by default),

5. CategoricalEncoder (if present),

6. Scaler always last (StandardScaler by default).

7. TimestampTransformer (if present).

Configuration example:

train:
  data_preprocessor:
    steps:
    - name: column_selector
      params:
        max_nan_frac_per_col: 0.05
        features_to_exclude: ['exclude_this_feature']
    - name: counter_diff_transformer
      step_name: counter_flow
      params:
        counters: ['flow_total_m3']
        compute_rate: True
        fill_first: 'zero'
    - name: counter_diff_transformer
      step_name: counter_energy
      params:
        counters: ['energy_total_kwh']
        compute_rate: False
        fill_first: 'zero'
        reset_strategy: 'rollover'
        rollover_values:
          'energy_total_kwh': 100000.0

NAME_ALIASES: Dict[str, str] = {'angle_transform': 'angle_transformer', 'categorical_encoder': 'categorical_encoder', 'counter_diff': 'counter_diff_transformer', 'counter_diff_transform': 'counter_diff_transformer', 'duplicate_value_to_nan': 'duplicate_to_nan', 'duplicate_values_to_nan': 'duplicate_to_nan', 'ffill_imputer': 'ffill_imputer', 'imputer': 'simple_imputer', 'scaler': 'scaler', 'timestamp_features': 'timestamp_transformer', 'timestamp_transform': 'timestamp_transformer'}

STEP_REGISTRY = {'angle_transformer': <class 'energy_fault_detector.data_preprocessing.angle_transformer.AngleTransformer'>, 'categorical_encoder': <class 'energy_fault_detector.data_preprocessing.categorical_encoder.CategoricalEncoder'>, 'column_selector': <class 'energy_fault_detector.data_preprocessing.column_selector.ColumnSelector'>, 'counter_diff_transformer': <class 'energy_fault_detector.data_preprocessing.counter_diff_transformer.CounterDiffTransformer'>, 'duplicate_to_nan': <class 'energy_fault_detector.data_preprocessing.duplicate_value_to_nan.DuplicateValuesToNan'>, 'ffill_imputer': <class 'energy_fault_detector.data_preprocessing.ffill_imputer.ForwardFillImputer'>, 'low_unique_value_filter': <class 'energy_fault_detector.data_preprocessing.low_unique_value_filter.LowUniqueValueFilter'>, 'scaler': <class 'energy_fault_detector.data_preprocessing.scaler.Scaler'>, 'simple_imputer': <class 'energy_fault_detector.data_preprocessing.imputer.Imputer'>, 'timestamp_transformer': <class 'energy_fault_detector.data_preprocessing.timestamp_transformer.TimestampTransformer'>}

fit(X, y=None, **fit_params)

Fit all transformers in the pipeline.

Parameters:

• X (DataFrame) – Input DataFrame.

• y – Target variable (optional).

• **fit_params – Parameters to pass to the fit method of each step. Use step_name__parameter format (e.g., column_selector__protected_features=[‘feature1’]). Special parameter ‘protected_features’ can be used to validate that these features are present in the output after fitting.

Returns:

self

Raises:

ValueError – If protected_features are specified but not all are present in output.

fit_transform(x, **kwargs)

Fit and transform in one step.

Parameters:

• x (DataFrame) – Input DataFrame.

• **kwargs (Any) – Parameters to pass to the fit method of each step. Use step_name__parameter format (e.g., column_selector__protected_features=[‘feature1’]).

Return type:

DataFrame

Returns:

Transformed DataFrame with the same index as input.

inverse_transform(X, **kwargs)

Inverse-transform scaler, timestamp transformer, categorical encoder and angle transformer (other transforms are not reversed).

Parameters:

X (DataFrame) – The transformed data.

Return type:

DataFrame

Returns:

DataFrame with inverse scaling and angle back-transformation.

transform(x, **kwargs)

Apply pipeline steps to the input DataFrame.

Parameters:

x (DataFrame) – Input DataFrame.

Return type:

DataFrame

Returns:

DataFrame with the same index as input.