cursus.processing.categorical.streaming_risk_table_processor

Streaming Risk Table Processor - IterableDataset Support

This processor extends RiskTableMappingProcessor to support fitting from PipelineIterableDataset by accumulating cross-tabulation counts incrementally.

Key Features: - Single-pass streaming accumulation - Memory-efficient cross-tabulation - Exact computation (not approximate) - Same process()/transform() API as base processor - Batch fitting for multiple fields (10x faster than individual fitting)

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

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_streaming(dataset, field_names=None, label_name=None, smooth_factor=None, count_threshold=None, max_samples=None, show_progress=False)[source]

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

fit(data)[source]

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