Transform

Fitted transformation pipelines for numerical, categorical, and datetime features. The pipeline is the primary entry point — transformers are registered by type and applied in order. Pipelines serialise their fitted state with save() / load().

Pipeline

bitbullet.transform.core.pipeline.TransformPipeline

Composable transformation pipeline with state management.

This class implements the Chain of Responsibility pattern, where each transformer is a handler in the chain. It provides: - Declarative API for building pipelines - Automatic state management and serialization - Lazy evaluation for performance - Parallel execution support (coming in next version)

Design Features: - Immutable after fit: Prevents accidental state corruption - Type-safe: Full validation at each step - Production-ready: Battle-tested serialization - Observable: Rich metadata and logging

Example

pipeline = TransformPipeline() pipeline.add('numerical', 'log', columns=['revenue']) pipeline.add('categorical', 'label_encode', columns=['region']) df_transformed = pipeline.fit_transform(df) pipeline.save('production.pkl')

is_fitted property

Whether pipeline has been fitted.

metadata property

Pipeline metadata.

name property

Pipeline name.

transformers property

List of transformers (read-only).

__getitem__(index)

Get transformer by index.

__init__(registry=None, name='pipeline', validate=True)

Initialize transformation pipeline.

Parameters:

Name	Type	Description	Default
registry	Optional[TransformerRegistry]	Transformer registry (uses global if None)	None
name	str	Pipeline name for identification	'pipeline'
validate	bool	Whether to validate data at each step	True

__len__()

Number of transformers in pipeline.

__repr__()

String representation.

add(transformer_type, method, columns=None, name=None, enabled=True, params=None, **kwargs)

Add a transformation step to the pipeline.

This uses the Builder pattern for fluent API composition.

Parameters:

Name	Type	Description	Default
transformer_type	str	Category of transformer (e.g., 'numerical')	required
method	str	Specific method (e.g., 'log', 'standard_scale')	required
columns	Optional[List[str]]	Columns to transform (None = all applicable columns)	None
name	Optional[str]	Optional custom name for this step	None
enabled	bool	Whether this step is active	True
params	Optional[Dict[str, Any]]	Method-specific parameters (dict)	None
**kwargs	Any	Alternative way to pass method-specific parameters	{}

Returns:

Type	Description
TransformPipeline	Self for method chaining

Raises:

Type	Description
RuntimeError	If pipeline is already fitted
KeyError	If transformer not registered

Example

pipeline.add('numerical', 'log', columns=['revenue'], params={'base': 10})

Or use kwargs directly:

pipeline.add('numerical', 'log', columns=['revenue'], base=10)

add_transformer(transformer)

Add a pre-configured transformer instance.

This provides flexibility to add custom transformers.

Parameters:

Name	Type	Description	Default
transformer	BaseTransformer	Pre-configured transformer instance	required

Returns:

Type	Description
TransformPipeline	Self for method chaining

Raises:

Type	Description
RuntimeError	If pipeline is already fitted

clone()

Create a deep copy of the pipeline.

Returns:

Type	Description
TransformPipeline	New pipeline instance with same configuration

disable_step(name)

Disable a transformation step by name.

Parameters:

Name	Type	Description	Default
name	str	Name of the step to disable	required

Raises:

Type	Description
KeyError	If step not found

enable_step(name)

Enable a transformation step by name.

Parameters:

Name	Type	Description	Default
name	str	Name of the step to enable	required

Raises:

Type	Description
KeyError	If step not found

fit(data, y=None, verbose=False)

Fit all transformers in the pipeline.

This sequentially fits each transformer on the output of the previous transformer, learning parameters from the data.

Parameters:

Name	Type	Description	Default
data	DataFrame	Input data to fit on	required
y	Optional[Series]	Optional target values for target-aware transformers.	None
verbose	bool	Whether to print progress	False

Returns:

Type	Description
TransformPipeline	Self for method chaining

Raises:

Type	Description
ValueError	If data is invalid
RuntimeError	If pipeline is empty

fit_transform(data, y=None, verbose=False)

Fit pipeline and transform data in one step.

Each transformer is invoked through its own fit_transform so target-aware transformers (e.g. the OOF target encoder) can return training-time values that differ from fit(); transform() without the pipeline having to know anything about folds.

For every transformer that doesn't override fit_transform, BaseTransformer.fit_transform defaults to fit(); transform(), so the chained behaviour is identical to the previous self.fit(); self.transform(). Only the target encoder cares.

Parameters:

Name	Type	Description	Default
data	DataFrame	Input data to fit and transform.	required
y	Optional[Series]	Optional target values for target-aware transformers.	None
verbose	bool	Whether to print progress.	False

Returns:

Type	Description
DataFrame	Transformed data after every step has run.

get_step(name)

Get a transformer by name.

Parameters:

Name	Type	Description	Default
name	str	Name of the step	required

Returns:

Type	Description
BaseTransformer	Transformer instance

Raises:

Type	Description
KeyError	If step not found

list_steps()

List all steps in the pipeline.

Returns:

Type	Description
List[Dict[str, Any]]	List of step information dictionaries

load(path, registry=None) classmethod

Load fitted pipeline from disk.

Parameters:

Name	Type	Description	Default
path	Union[str, Path]	Path to saved pipeline	required
registry	Optional[TransformerRegistry]	Transformer registry (uses global if None)	None

Returns:

Type	Description
TransformPipeline	Loaded pipeline instance

Raises:

Type	Description
FileNotFoundError	If path doesn't exist

reset()

Reset pipeline to unfitted state.

This allows adding new transformers or re-fitting.

save(path, compress=3)

Save fitted pipeline to disk.

Uses joblib for efficient serialization with compression.

Parameters:

Name	Type	Description	Default
path	Union[str, Path]	Output file path	required
compress	int	Compression level (0-9, higher = more compression)	3

Raises:

Type	Description
RuntimeError	If pipeline not fitted

transform(data, verbose=False)

Transform data using fitted pipeline.

Parameters:

Name	Type	Description	Default
data	DataFrame	Input data to transform	required
verbose	bool	Whether to print progress	False

Returns:

Type	Description
DataFrame	Transformed data

Raises:

Type	Description
RuntimeError	If pipeline not fitted
ValueError	If data is invalid

Base Classes

bitbullet.transform.core.base.BaseTransformer

Bases: ABC

Abstract base class for all transformers.

This implements the Strategy pattern, allowing transformers to be swapped at runtime while maintaining a consistent interface.

Design Principles: - Immutability: Config is frozen after creation - Explicit state: Fitted parameters are clearly separated - Type safety: Full type hints with runtime validation - Performance: Designed for vectorization and parallel execution

config property

Immutable configuration.

is_fitted property

Whether this transformer has been fitted.

state property

Current fitted state.

__init__(config)

Initialize transformer with configuration.

Parameters:

Name	Type	Description	Default
config	TransformConfig	Immutable configuration for this transformer	required

__repr__()

String representation.

fit(data, y=None)

Fit transformer to data.

This is the public API that handles pre/post processing around the internal _fit implementation.

Parameters:

Name	Type	Description	Default
data	DataFrame	Input data to fit on	required
y	Optional[Series]	Optional target values for target-aware transformations.	None

Returns:

Type	Description
BaseTransformer	Self for method chaining

Raises:

Type	Description
ValueError	If data is invalid

fit_transform(data, y=None)

Fit to data and transform in one step.

This is often more efficient than calling fit() and transform() separately.

Parameters:

Name	Type	Description	Default
data	DataFrame	Input data to fit and transform	required
y	Optional[Series]	Optional target values for target-aware transformations.	None

Returns:

Type	Description
DataFrame	Transformed data

get_feature_names_out()

Get output feature names.

Returns:

Type	Description
List[str]	List of output column names

get_params()

Get fitted parameters.

Returns:

Type	Description
Dict[str, Any]	Dictionary of fitted parameters

transform(data)

Transform data using fitted parameters.

Parameters:

Name	Type	Description	Default
data	DataFrame	Input data to transform	required

Returns:

Type	Description
DataFrame	Transformed data

Raises:

Type	Description
RuntimeError	If transformer hasn't been fitted
ValueError	If data is invalid

bitbullet.transform.core.base.TransformConfig

Bases: BaseModel

Configuration for a transformation step.

This uses Pydantic for validation and serialization, ensuring type safety at runtime.

model_dump_json(**kwargs)

Override to handle non-serializable types.

Registry

bitbullet.transform.core.registry.TransformerRegistry

Central registry for all transformers using the Singleton pattern.

This implements the Registry and Singleton patterns, providing a global point of access for transformer registration and creation.

Example

registry = TransformerRegistry() registry.register('numerical', 'log', LogTransformer) transformer = registry.create('numerical', 'log', config)

__init__()

Initialize registry (only once due to Singleton).

__new__()

Ensure only one instance exists (Singleton pattern).

__repr__()

String representation.

clear()

Clear all registrations (useful for testing).

create(transformer_type, method, config, **kwargs)

Create a transformer instance.

This method uses the Factory pattern to instantiate transformers dynamically based on the registry.

Parameters:

Name	Type	Description	Default
transformer_type	str	Category of transformer	required
method	str	Specific method name	required
config	TransformConfig	Configuration for the transformer	required
**kwargs	Any	Additional arguments passed to transformer	{}

Returns:

Type	Description
BaseTransformer	Configured transformer instance

Raises:

Type	Description
KeyError	If transformer not registered

get_metadata(transformer_type, method)

Get metadata for a specific transformer.

Parameters:

Name	Type	Description	Default
transformer_type	str	Category of transformer	required
method	str	Specific method name	required

Returns:

Type	Description
Dict[str, Any]	Metadata dictionary

is_registered(transformer_type, method)

Check if a transformer is registered.

Parameters:

Name	Type	Description	Default
transformer_type	str	Category of transformer	required
method	str	Specific method name	required

Returns:

Type	Description
bool	True if registered, False otherwise

list_transformers(transformer_type=None)

List all registered transformers.

Parameters:

Name	Type	Description	Default
transformer_type	Optional[str]	Optional filter for specific type	None

Returns:

Type	Description
Dict[str, list[str]]	Dictionary mapping types to lists of methods

register(transformer_type, method, transformer_class, factory=None, **metadata)

Register a transformer class or factory.

Parameters:

Name	Type	Description	Default
transformer_type	str	Category of transformer (e.g., 'numerical', 'categorical')	required
method	str	Specific method name (e.g., 'log', 'standard_scale')	required
transformer_class	Type[BaseTransformer]	The transformer class to register	required
factory	Optional[Callable[..., BaseTransformer]]	Optional factory function for custom instantiation	None
**metadata	Any	Additional metadata about this transformer	{}

Raises:

Type	Description
ValueError	If trying to re-register without override

unregister(transformer_type, method)

Unregister a transformer.

Parameters:

Name	Type	Description	Default
transformer_type	str	Category of transformer	required
method	str	Specific method name	required

Raises:

Type	Description
KeyError	If transformer not registered

Built-in Transformers

bitbullet.transform.transformers.numerical.NumericalTransformer

Bases: BaseTransformer

Flexible numerical transformer supporting multiple methods.

This uses the Strategy pattern internally, selecting the appropriate transformation method based on configuration.

Supported Methods: - log, log1p: Logarithmic transformations - sqrt: Square root transformation - boxcox: Box-Cox power transformation - yeo_johnson: Yeo-Johnson transformation (handles negatives) - standard_scale: Z-score normalization - minmax_scale: Min-max normalization to [0, 1] - robust_scale: Robust scaling using median and IQR - quantile_normal, quantile_uniform: Quantile transformations - reciprocal: 1/x transformation - winsorize: Clip outliers - percentile_binning: Convert to categorical bins using quantiles - arcsine: Arcsine transformation for proportion data

output_dtype(method) classmethod

Declared output dtype for the given method. preserve ⇒ same dtype the transformer accepts as input (numerical for this class).

bitbullet.transform.transformers.categorical.CategoricalTransformer

Bases: BaseTransformer

Categorical transformer with production-grade encoding strategies.

Supported Methods: - label_encode: Integer encoding - onehot_encode: One-hot encoding with sparse support - ordinal_encode: Ordered integer encoding - frequency_encode: Encode by category frequency - target_encode: Target-based encoding (requires y) - bin_rare: Group rare categories into 'other' - hash_encode: Feature hashing for high cardinality

output_dtype(method) classmethod

Declared output dtype for the given method. preserve ⇒ same dtype as input (categorical for this class).

bitbullet.transform.transformers.datetime.DateTimeTransformer

Bases: BaseTransformer

DateTime feature engineering transformer.

Supported Methods: - extract_components: Extract year, month, day, etc. - cyclical_encode: Encode time features as sin/cos for cyclical patterns - time_since: Calculate time since reference date - is_weekend: Binary weekend indicator - is_holiday: Binary holiday indicator (requires holiday calendar)

bitbullet.transform.transformers.target.TargetEncodingTransformer

Bases: BaseTransformer

Target encoder with OOF-aware fit_transform and multiclass support.

describe()

Return an audit-trail summary of the fitted encoder.

All source columns in one step share the same mode/classes/positive_label (driven by target_type + y), so those live at the top level. output_columns_by_input maps each source column to the channel columns produced for it.

Folds and CV params are recorded as configuration only; they are not part of the serialized inference state.

fit_transform(data, y=None)

Fit and return OOF-encoded training data.

Order of operations

1. fit(data, y) learns + stores the full-data map per channel (this is what transform uses at inference).
2. For each source column, compute the OOF encoding using the configured CV strategy. Each fold builds a temporary map from the other folds and encodes the held-out rows with it.
3. Replace source columns with their OOF channel columns.

The per-fold maps are throwaway scratch — only the full-data maps from step 1 are serialized for inference.

EDA Utilities

bitbullet.transform.eda.generate_feature_stats(df)

Generate a comprehensive statistics table for feature exploration to assist in deciding transformations.

Returns:

Type	Description
DataFrame	Tuple containing:
List[str]	stats_df: DataFrame with comprehensive statistics
List[str]	numerical_cols: List of identified numerical columns
Tuple[DataFrame, List[str], List[str]]	categorical_cols: List of identified categorical/text columns

bitbullet.transform.eda.plot_pipeline_transformations(pipeline, X_orig, X_trans)

Interactively generates a before-and-after plot for all features modified by the given TransformPipeline. Automatically handles both numerical and categorical data behaviors.