cursus.processing.categorical

Categorical Processing Module

This module provides atomic processors for categorical data processing, including encoding, imputation, validation, and numerical categorization.

class CategoricalLabelProcessor(initial_categories=None, update_on_new=True, unknown_label=-1)

Bases: Processor

Map category strings to integer labels.

Warning

With update_on_new=True (the default), process() MUTATES the category->label mapping on first sight of a new category. This is not safe across DataLoader workers: under num_workers > 0 the processor is forked per worker, so each worker assigns its own (divergent) ids to the same category and labels become inconsistent. For multi-worker / distributed use, pass a complete initial_categories list and set update_on_new=False so the mapping is fixed and read-only during processing.

process(input_text)

class MultiClassLabelProcessor(label_list=None, one_hot=False, strict=False)

Bases: Processor

Processes multi-class labels into a format suitable for machine learning models.

This processor handles encoding categorical labels into numerical arrays and optionally provides one-hot encoding using numpy.

Parameters:

• label_list (Optional[List[str]]) – A list of unique label strings. If provided, the processor will learn this mapping; otherwise, it will learn the mapping from the data it processes.

• one_hot (bool) – If True, output one-hot encoded labels.

• strict (bool) – If True, raise error for unknown labels when label_list is provided.

process(labels)

Encodes the input labels.

Parameters:

labels (Union[str, List[str]]) – A single label or a list of labels.

Returns:

Encoded labels as a numpy array.

Return type:

np.ndarray

class DictionaryEncodingProcessor(categorical_map=None, unknown_strategy='default', default_value=0, columns=None)

Bases: Processor

Pure dictionary-based categorical encoding.

Extracted from TSA CategoricalTransformer functionality.

Parameters:

• categorical_map (Dict[str, Dict[str, int]] | None) – Pre-defined category mappings

• unknown_strategy (str) – ‘error’, ‘default’, ‘ignore’

• default_value (int) – Value for unknown categories

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

fit(data)

Learn categorical mappings from data if not provided

get_all_vocab_sizes()

Get vocabulary sizes for all categorical columns

get_config()

Return processor configuration

get_vocab_size(column_name)

Get vocabulary size for a specific column

inverse_transform(encoded_data)

Inverse transform encoded data back to original categorical values

process(input_data)

Apply dictionary encoding

class CategoricalImputationProcessor(default_values=None, missing_indicators=None, strategy='default', constant_value='UNKNOWN')

Bases: Processor

Handles missing categorical values with configurable defaults.

Extracted from TSA default value handling logic.

Parameters:

• default_values (Dict[str, Any] | None) – Dictionary of field -> default value mappings

• missing_indicators (List[Any] | None) – Values that indicate missing data

• strategy (str) – ‘default’, ‘mode’, ‘constant’

• constant_value (str) – Value to use for constant strategy

add_missing_indicator(indicator)

Add a new missing indicator

fit(data)

Learn default values from data if needed

get_config()

Return processor configuration

get_missing_statistics(data)

Get statistics about missing values in the data

process(input_data)

Apply categorical imputation

remove_missing_indicator(indicator)

Remove a missing indicator

class NumericalCategoricalProcessor(binning_strategy='equal_width', n_bins=5, bin_edges=None, thresholds=None, labels=None, columns=None)

Bases: Processor

Converts numerical values to categorical labels using binning or thresholds.

Extracted from TSA str(int(float(cur_var))) conversion patterns.

Parameters:

• binning_strategy (str) – ‘equal_width’, ‘equal_frequency’, ‘custom’, ‘threshold’

• n_bins (int) – Number of bins for equal_width/equal_frequency

• bin_edges (List[float] | None) – Custom bin edges for ‘custom’ strategy

• thresholds (List[float] | None) – Threshold values for ‘threshold’ strategy

• labels (List[str] | None) – Custom labels for categories

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

fit(data)

Learn binning parameters from data.

For equal_width / equal_frequency the bin edges are LEARNED here. For custom / threshold the bins come from the constructor params (bin_edges / thresholds) and there is nothing to learn — but we still validate the required param is present so is_fitted is never set True for a strategy that has no usable bins (which previously caused a silent mis-binning at transform time).

get_bin_info(column=None)

Get binning information

get_config()

Return processor configuration

process(input_data)

Apply numerical to categorical conversion

class CategoricalValidationProcessor(allowed_values=None, validation_rules=None, validation_strategy='warn', report_violations=True, max_violations=None)

Bases: Processor

Validates categorical data quality and consistency.

Extracted from TSA data validation requirements.

Parameters:

• allowed_values (Dict[str, Set[Any]] | None) – Dictionary of field -> allowed values mappings

• validation_rules (Dict[str, callable] | None) – Custom validation rules

• validation_strategy (str) – ‘strict’, ‘warn’, ‘filter’

• report_violations (bool) – Whether to report validation violations

• max_violations (int | None) – Maximum allowed violations before error

add_allowed_values(field, values)

Add allowed values for a field

add_validation_rule(field, rule_func, rule_name=None)

Add custom validation rule for a field

fit(data)

Learn allowed values from training data if not provided

get_config()

Return processor configuration

get_validation_report()

Get detailed validation report

process(input_data)

Apply categorical validation

remove_allowed_values(field, values)

Remove allowed values for a field

class RiskTableMappingProcessor(column_name, label_name, smooth_factor=0.0, count_threshold=0, risk_tables=None)

Bases: Processor

A processor that performs risk-table-based mapping on a specified categorical variable. The ‘process’ method (called via __call__) handles single values. The ‘transform’ method handles pandas Series or DataFrames.

fit(data)

get_name()

get_risk_tables()

load_risk_tables(filepath)

process(input_value)

Process a single input value (for the configured ‘column_name’), mapping it to its binned risk value. This method is called when the processor instance is called as a function.

save_risk_tables(output_dir)

set_risk_tables(risk_tables)

transform(data)

Transform data using the computed risk tables. - If data is a DataFrame, transforms the ‘column_name’ Series within it. - If data is a Series, transforms the Series (assumed to be the target column). - If data is a single value, uses the ‘process’ method.

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

class StreamingRiskTableProcessor(column_name, label_name, smooth_factor=0.0, count_threshold=0, risk_tables=None)

Bases: RiskTableMappingProcessor

Streaming-aware risk table processor.

Extends RiskTableMappingProcessor to support fitting from PipelineIterableDataset by accumulating cross-tabulation counts incrementally.

Uses online cross-tabulation for exact computation: - Maintains category counts for each label value (0, 1) - Computes risk ratios with smoothing - Memory proportional to number of unique categories (not dataset size)

Examples

>>> # Create streaming processor
>>> proc = StreamingRiskTableProcessor(
...     column_name='customer_type',
...     label_name='label',
...     smooth_factor=0.1,
...     count_threshold=5
... )
>>>
>>> # Fit from streaming dataset
>>> proc.fit_streaming(train_iterable_dataset)
>>>
>>> # Use in pipeline (same API as base processor)
>>> dataset.add_pipeline('customer_type', proc)

fit(data)

Fit risk tables from data.

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

Parameters:

data (DataFrame | torch.utils.data.IterableDataset) – DataFrame or IterableDataset

Returns:

self (for method chaining)

Return type:

StreamingRiskTableProcessor

fit_streaming(dataset, field_names=None, label_name=None, smooth_factor=None, count_threshold=None, max_samples=None, show_progress=False)

Fit risk tables 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 -> risk_tables]

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

• label_name (str | None) – Optional label name override. If None, uses self.label_name

• smooth_factor (float | None) – Optional smoothing factor override. If None, uses self.smooth_factor

• count_threshold (int | None) – Optional count threshold override. If None, uses self.count_threshold

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

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

Returns:

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

Return type:

• Single-field mode

Raises:

RuntimeError – If no valid samples found

Modules

categorical_imputation_processor	Categorical Imputation Processor for Missing Values
categorical_label_processor
categorical_validation_processor	Categorical Validation Processor for Data Quality Checks
dictionary_encoding_processor	Dictionary Encoding Processor for Categorical Features
multiclass_label_processor
numerical_categorical_processor	Numerical Categorical Processor for Converting Numbers to Categories
risk_table_processor
streaming_risk_table_processor	Streaming Risk Table Processor - IterableDataset Support