cursus.processing.numerical

Numerical Processing Module

This module provides atomic processors for numerical data processing, including scaling, normalization, imputation, and binning.

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'])

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

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)

get_imputation_value()

Get the fitted imputation value.

Returns:

Imputation value for this column

Raises:

RuntimeError – If processor not fitted

Return type:

int | float

get_name()

Return processor name for base class compatibility.

get_params()

Get processor parameters (DEPRECATED).

Use get_imputation_value() instead for the fitted value.

Returns:

Dictionary with all parameters

Return type:

dict

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')

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

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/')

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

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.

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

Bases: NumericalVariableImputationProcessor

Streaming-aware numerical imputation processor.

Extends NumericalVariableImputationProcessor to support fitting from PipelineIterableDataset by accumulating statistics incrementally.

Uses online algorithms for efficient single-pass statistics computation: - Mean: Running sum and count - Median: Approximate using sample collection (memory-limited) - Mode: Approximate using sample collection (memory-limited)

Examples

>>> # Create streaming processor
>>> proc = StreamingNumericalImputationProcessor(
...     column_name='age',
...     strategy='mean'
... )
>>>
>>> # Fit from streaming dataset
>>> proc.fit_streaming(train_iterable_dataset)
>>>
>>> # Use in pipeline (same API as base processor)
>>> dataset.add_pipeline('age', proc)

fit(X, y=None)

Fit imputation value from data.

Automatically detects input type and delegates to appropriate method: - IterableDataset: uses fit_streaming() - Series/DataFrame: uses parent class fit()

Parameters:

• X (Series | DataFrame | torch.utils.data.IterableDataset) – Series, DataFrame, or IterableDataset

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

Returns:

self (for method chaining)

Return type:

StreamingNumericalImputationProcessor

fit_streaming(dataset, field_names=None, strategy=None, max_samples=None, median_sample_limit=100000, show_progress=False)

Fit imputation values from streaming dataset.

Supports both single-field and multi-field (batch) fitting: - Single-field: Uses self.column_name, returns self for chaining - Multi-field: Processes all fields in ONE pass, returns Dict[field_name -> imputation_value]

Parameters:

• dataset (torch.utils.data.IterableDataset) – PipelineIterableDataset to stream from

• field_names (List[str] | None) – Optional list of fields for batch fitting. If None, uses self.column_name

• strategy (str | None) – Optional strategy override (‘mean’, ‘median’, ‘mode’). If None, uses self.strategy

• max_samples (int | None) – Optional limit on samples processed (for early stopping)

• median_sample_limit (int) – Max samples to keep for median/mode computation

• show_progress (bool) – Whether to show progress bar (for batch mode)

Returns:

self (for method chaining) - Multi-field mode: Dict[field_name -> imputation_value]

Return type:

• Single-field mode

Raises:

ValueError – If strategy is unknown

class NumericalBinningProcessor(column_name, n_bins=5, strategy='quantile', bin_labels=None, output_column_name=None, handle_missing_value='as_is', handle_out_of_range='boundary_bins')

Bases: Processor

A processor that performs numerical binning on a specified column using either equal-width or quantile strategies, outputting categorical bin labels.

fit(data)

get_params()

classmethod load_params(source)

process(input_value)

save_params(output_dir)

transform(data)

class MinMaxScalingProcessor(feature_range=(0, 1), learned_params=None, columns=None, clip_values=True)

Bases: Processor

Min-max scaling with learned parameters.

Extracted from TSA preprocessing: - seq_num_mtx[:, :-2] = seq_num_mtx[:, :-2] * np.array(seq_num_scale_) + np.array(seq_num_min_)

Parameters:

• feature_range (Tuple[float, float]) – Target range for scaling

• learned_params (Dict[str, Dict[str, float]] | None) – Pre-computed scaling parameters

• columns (List[str] | None) – Specific columns to scale

• clip_values (bool) – Whether to clip to feature_range

fit(data)

Learn scaling parameters from data.

If params were pre-supplied at construction (learned_params), fit is intentionally a no-op that REUSES them (load-a-prior-fit pattern) — it logs that it is skipping recomputation so the no-op is not mistaken for a silent failure. Pass learned_params=None to force learning from data.

get_config()

Return processor configuration

get_data_range(column=None)

Get the original data range(s)

get_scaling_info()

Get detailed scaling information

inverse_transform(scaled_data)

Inverse transform scaled data back to original scale

process(input_data)

Apply min-max scaling

class FeatureNormalizationProcessor(method='l2', axis=1, columns=None, epsilon=1e-08)

Bases: Processor

Normalizes features using L1, L2, or max normalization.

Extracted from TSA feature normalization requirements.

Parameters:

• method (str) – ‘l1’, ‘l2’, ‘max’

• axis (int) – Axis along which to normalize (0 for columns, 1 for rows)

• columns (List[str] | None) – Specific columns to normalize

• epsilon (float) – Small value to avoid division by zero

check_normalization(data, tolerance=1e-06)

Check if data is already normalized according to the specified method

fit(data)

No fitting required for normalization

get_config()

Return processor configuration

get_normalization_info(data)

Get information about the normalization that would be applied

process(input_data)

Apply feature normalization

Modules

feature_normalization_processor	Feature Normalization Processor for Numerical Features
minmax_scaling_processor	MinMax Scaling Processor for Numerical Features
numerical_binning_processor
numerical_imputation_processor	Numerical Imputation Processor - Single Column Architecture
streaming_numerical_imputation_processor	Streaming Numerical Imputation Processor - IterableDataset Support