cursus.processing.numerical.numerical_imputation_processor

Numerical Imputation Processor - Single Column Architecture

This processor performs imputation on a SINGLE numerical column. Follows the single-column architecture pattern for real-time inference pipelines.

For batch processing of multiple columns, use one processor per column.

class NumericalVariableImputationProcessor(column_name, imputation_value=None, strategy=None)

Bases: Processor

A processor that performs imputation on a SINGLE numerical variable/column.

Designed for real-time inference pipelines where each processor handles one column and processors can be chained with >> operator.

For batch processing of multiple columns, use one processor per column.

Examples

>>> # Create processor for single column
>>> proc = NumericalImputationProcessor(
...     column_name='age',
...     strategy='mean'
... )
>>>
>>> # Fit on training data
>>> proc.fit(train_df['age'])
>>>
>>> # Process single value (real-time inference)
>>> imputed_value = proc.process(None)  # Returns mean value
>>>
>>> # Transform Series or DataFrame
>>> imputed_series = proc.transform(test_df['age'])

get_name()

Return processor name for base class compatibility.

fit(X, y=None)

Fit imputation value on a Series (single column).

Parameters:

• X (Series | DataFrame) – Series (preferred) or DataFrame with column_name

• y (Series | None) – Ignored (for sklearn compatibility)

Returns:

self (for method chaining)

Raises:

• ValueError – If column_name not found in DataFrame

• ValueError – If strategy is unknown

Return type:

NumericalVariableImputationProcessor

process(input_value)

Process a SINGLE numerical value for this column.

This method is called by __call__ (inherited from base Processor). It handles single-value processing for real-time inference.

Parameters:

input_value (int | float | Any) – Single value to impute if missing

Returns:

Imputed value (or original if not missing)

Raises:

RuntimeError – If processor not fitted

Return type:

int | float

transform(X)

Transform data using the fitted imputation value.

Parameters:

X (Series | DataFrame | Any) – Series, DataFrame, or single value

Returns:

Imputed data in same format as input

Raises:

• RuntimeError – If processor not fitted

• ValueError – If column_name not found in DataFrame

Return type:

Series | DataFrame | float

Performance optimized: Uses fast path for single-value Series.

get_imputation_value()

Get the fitted imputation value.

Returns:

Imputation value for this column

Raises:

RuntimeError – If processor not fitted

Return type:

int | float

set_imputation_value(value)

Set imputation value (for pre-fitted processor).

Parameters:

value (int | float) – Imputation value to use

Raises:

ValueError – If value is not numeric

get_params()

Get processor parameters (DEPRECATED).

Use get_imputation_value() instead for the fitted value.

Returns:

Dictionary with all parameters

Return type:

dict

save_imputation_value(output_dir)

Save imputation value to disk.

Creates two files: 1. {column_name}_impute_value.pkl (for loading) 2. {column_name}_impute_value.json (for human readability)

Parameters:

output_dir (Path | str) – Directory to save artifacts to

Raises:

RuntimeError – If processor not fitted

Examples

>>> proc = NumericalImputationProcessor('age', strategy='mean')
>>> proc.fit(train_df['age'])
>>> proc.save_imputation_value('model_artifacts/')

load_imputation_value(filepath)

Load imputation value from disk.

Parameters:

filepath (Path | str) – Path to pickle file or directory containing it

Raises:

• FileNotFoundError – If file not found

• ValueError – If value is invalid

Examples

>>> proc = NumericalImputationProcessor('age', strategy='mean')
>>> proc.load_imputation_value('model_artifacts/')
>>> # Or specify exact file
>>> proc.load_imputation_value('model_artifacts/age_impute_value.pkl')

classmethod from_imputation_dict(imputation_dict)

Create processors from script output (impute_dict.pkl).

This factory method simplifies creating multiple processors from the dictionary format used by missing_value_imputation.py script.

Parameters:

imputation_dict (dict) – Dictionary mapping column names to imputation values Format: {column_name: imputation_value}

Returns:

Dictionary mapping column names to fitted processors

Raises:

• TypeError – If imputation_dict is not a dictionary

• ValueError – If column name is not a string

Return type:

dict

Examples

>>> with open("impute_dict.pkl", "rb") as f:
...     impute_dict = pkl.load(f)
>>> processors = NumericalImputationProcessor.from_imputation_dict(impute_dict)
>>> # Use processors in pipeline
>>> for col, proc in processors.items():
...     dataset.add_pipeline(col, proc)

classmethod from_script_artifacts(artifacts_dir, filename='impute_dict.pkl')

Load processors from script output directory.

Looks for impute_dict.pkl in the specified directory and creates processors from it.

Parameters:

• artifacts_dir (Path | str) – Directory containing impute_dict.pkl

• filename (str) – Name of the imputation dict file (default: impute_dict.pkl)

Returns:

Dictionary mapping column names to fitted processors

Raises:

FileNotFoundError – If impute_dict.pkl not found

Return type:

dict

Examples

>>> processors = NumericalImputationProcessor.from_script_artifacts(
...     "model_artifacts/"
... )
>>> for col, proc in processors.items():
...     dataset.add_pipeline(col, proc)