Universal Data Cleaner

A small, production-oriented tabular cleaning transformer for pandas + scikit-learn.

It is designed for the part of an ML pipeline that is easy to get wrong: cleaning training data and inference data with exactly the same learned rules.

Why this exists

A common notebook pattern is:

df["Age"] = df["Age"].fillna(df["Age"].median())

That leaks information when it is run before the train/test split, and it silently breaks at inference time because the median is recomputed on the new batch.

DataCleaner fits on training data, stores the learned values, and reuses them in transform() and in saved artifacts.

What it handles

• numeric coercion and median / mean / constant imputation
• optional IQR outlier capping (learned from training data only)
• categorical mode / constant imputation
• optional unseen-category policy: keep or replace
• boolean coercion and imputation
• datetime parsing plus calendar feature extraction
• optional missing-value indicator columns
• optional constant-column dropping
• stable output schema and column order
• declarative column contracts (roles, allowed values, ranges, patterns, uniqueness)
• automatic train-schema contract inference and transform-time drift reports
• on_drift policy: raise, warn, or ignore
• identifier columns that are never numeric-coerced, median-imputed, or capped
• joblib save / load with artifact versioning
• scikit-learn Pipeline, clone, and get_feature_names_out compatibility

The cleaner never mutates the input DataFrame and never drops inference rows.

Install from PyPI

pip install universal-data-cleaner

The distribution name is universal-data-cleaner; the import name is data_cleaner.

from data_cleaner import CleanerConfig, DataCleaner

Development setup with uv

git clone https://github.com/devlprkhan/universal-data-cleaner.git
cd universal-data-cleaner
uv sync

Run the test suite:

uv run pytest -q

Open the reference notebook:

uv run jupyter lab Universal_Data_Cleaner.ipynb

The notebook is also verified headlessly in CI-style environments with nbclient; both tools are in the dev dependency group.

Install directly from GitHub

uv add "git+https://github.com/devlprkhan/universal-data-cleaner.git"
# or
pip install "git+https://github.com/devlprkhan/universal-data-cleaner.git"

Quick start

import pandas as pd
from data_cleaner import CleanerConfig, DataCleaner

train = pd.DataFrame(
    {
        "age": [34, 41, None, 29],
        "salary": [70_000, 82_000, 91_000, 55_000],
        "dept": ["IT", "HR", "IT", None],
        "joined": ["2022-01-03", "2021-11-20", "2023-04-01", "bad-date"],
    }
)

config = CleanerConfig(
    datetime_columns=("joined",),
    add_missing_indicators=True,
    drop_constant_columns=True,
)

cleaner = DataCleaner(config).fit(train)
train_clean = cleaner.transform(train)
print(cleaner.cleaning_summary())
print(cleaner.outlier_summary())

For a column that should be numeric even though it arrived as dirty strings:

config = CleanerConfig(
    numeric_columns=("salary",),          # force numeric parsing
    datetime_columns=("joined",),        # parse dates instead of treating them as categories
    unseen_category="keep",              # or "fill" to replace unseen labels
)

Leak-safe workflow

from sklearn.model_selection import train_test_split

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

cleaner = DataCleaner(config).fit(X_train)

X_train_clean = cleaner.transform(X_train)
X_test_clean = cleaner.transform(X_test)   # uses train statistics only

Save the fitted cleaner with the model artifact:

cleaner.save("artifacts/data_cleaner.joblib")
restored = DataCleaner.load("artifacts/data_cleaner.joblib")

Column contracts and drift detection

A contract makes the assumptions about your data explicit and checks them again on every batch.

from data_cleaner import ColumnSpec, DataContract, CleanerConfig, DataCleaner

contract = DataContract(
    columns=[
        ColumnSpec("customer_id", role="id"),
        ColumnSpec("age", role="numeric", min_value=0, max_value=120),
        ColumnSpec("plan", role="category", allowed_values=("free", "pro", "team")),
    ],
)

cleaner = DataCleaner(
    CleanerConfig(contract=contract, on_drift="warn")
).fit(X_train)

X_test_clean = cleaner.transform(X_test)   # drift is detected, never mutates
print(cleaner.drift_report())

If you do not pass a contract, a conservative one is inferred from the training schema automatically. Drift is then reported without any configuration:

cleaner = DataCleaner().fit(X_train)
cleaner.transform(X_test)
report = cleaner.drift_report()

if not report.empty:
    print(report[["column", "check", "severity", "message"]])

Supported drift checks: missing_column, unexpected_column, dtype_change, unexpected_missing, unexpected_category, value_out_of_range, pattern_mismatch, and duplicate_values.

Control what happens when drift is found with CleanerConfig.on_drift:

role="id" is worth calling out: identifier columns are treated as text, never numeric-imputed, never outlier-capped, and rendered without a trailing .0. Use role="ignore" to drop operational columns from the output.

Contracts can be version-controlled and shared between training and serving:

contract.to_json()                       # store it with the model artifact
restored = DataContract.from_json(payload)

# Optional YAML support requires `pip install pyyaml`
# restored = DataContract.from_yaml(contract.to_yaml())

scikit-learn Pipeline

from sklearn.compose import ColumnTransformer
from sklearn.linear_model import Ridge
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler

numeric = ["age", "salary", "joined__year"]
categorical = ["dept"]
boolean = []

preprocess = ColumnTransformer(
    [
        ("num", StandardScaler(), numeric),
        ("cat", OneHotEncoder(handle_unknown="ignore"), categorical),
        ("bool", "passthrough", boolean),
    ],
    remainder="drop",
)

pipeline = Pipeline(
    [
        ("cleaner", DataCleaner(config)),
        ("features", preprocess),
        ("model", Ridge(alpha=1.0)),
    ]
)
pipeline.fit(X_train, y_train)
predictions = pipeline.predict(X_test)

Public API

Important production notes

• Fit the cleaner after splitting and only on training data.
• Do not pass IDs, free-text notes, or the target column to the cleaner.
• Mark identifier columns with ColumnSpec(..., role="id") so they are never treated as numeric measurements.
• Set on_drift="raise" in production when an unexpected schema change should stop the job instead of silently cleaning it.
• Keep DataSource-style constant columns out of the model, or let drop_constant_columns=True remove them.
• unseen_category="keep" is the default because one-hot encoders with handle_unknown="ignore" can handle an unseen label safely.
• Datetime columns are replaced by calendar features. Their missing values are represented by the generated feature NaNs plus an optional missing indicator.
• IQR capping is skipped automatically when the IQR is zero, so constant or near-constant numeric columns are not accidentally destroyed.

Project layout

.
├── data_cleaner.py                    # reusable library module
├── Universal_Data_Cleaner.ipynb       # end-to-end reference workflow
├── tests/test_data_cleaner.py         # unit and integration tests
├── pyproject.toml                     # package metadata and dev tooling
├── uv.lock                            # reproducible dependency lock
├── README.md
├── ROADMAP.md
├── CHANGELOG.md
├── CONTRIBUTING.md
├── SECURITY.md
└── LICENSE

Contributing

Contributions are welcome. See CONTRIBUTING.md for the local setup, test command, and pull-request expectations.

License

Released under the MIT License.