1. Why Train-Only Preprocessing Matters

Data preparation is part of the machine-learning model. Any preprocessing step that learns information from data, such as a median, mean, most-frequent category, scaling parameter or category vocabulary, must be fitted using the training set only.

If preprocessing statistics are calculated from the complete dataset before the train-test split, information from the test set influences the training process. This is called data leakage. Leakage can make evaluation results appear better than the model's true performance on unseen data.

2. Prerequisites

You should understand Python variables, functions, dictionaries and basic pandas DataFrame operations. This implementation uses pandas for data inspection and scikit-learn for train-test splitting and reusable preprocessing.

python -m pip install pandas numpy scikit-learn

import numpy as np
import pandas as pd
import sklearn

print("NumPy:", np.__version__)
print("pandas:", pd.__version__)
print("scikit-learn:", sklearn.__version__)

3. Create a Practice CSV

If you already have a CSV file, you can skip this section and update the file path in the next step. The following code creates a small dataset containing numeric values, categories, missing values, an invalid numeric value and one duplicate row.

from pathlib import Path
import pandas as pd

sample_data = pd.DataFrame({
    "customer_id": [101, 102, 103, 104, 105, 106, 107, 107],
    "age": [25, 41, None, 36, 29, 52, 44, 44],
    "monthly_income": [35000, 62000, 48000, None, 41000, 79000, "unknown", "unknown"],
    "city": ["Kolkata", "Delhi", None, "Mumbai", " kolkata ", "Delhi", "Pune", "Pune"],
    "signup_date": [
        "2026-01-05", "2026-01-10", "invalid", "2026-02-01",
        "2026-02-08", None, "2026-03-15", "2026-03-15"
    ],
    "purchased": [1, 0, 1, 0, 1, 1, 0, 0]
})

csv_path = Path("customer_training_data.csv")
sample_data.to_csv(csv_path, index=False)
print(f"Created: {csv_path.resolve()}")

4. Load the CSV Safely

Use a function that checks the file path, loads the file and confirms that data is present. Fail early with a clear error instead of allowing an empty or incorrect dataset to move further into the pipeline.

from pathlib import Path
import pandas as pd


def load_csv(file_path: str) -> pd.DataFrame:
    """Load a CSV file and perform basic structural validation."""
    path = Path(file_path)

    if not path.exists():
        raise FileNotFoundError(f"CSV file not found: {path.resolve()}")

    if path.suffix.lower() != ".csv":
        raise ValueError(f"Expected a .csv file, received: {path.suffix}")

    df = pd.read_csv(path)

    if df.empty:
        raise ValueError("The CSV file contains no data rows.")

    if not df.columns.is_unique:
        raise ValueError("The CSV contains duplicate column names.")

    return df


df = load_csv("customer_training_data.csv")
print(df.head())

5. Inspect Shape, Types and Sample Values

Inspection should happen before cleaning. Record the number of rows and columns, inferred dtypes, missing-value counts, duplicate counts and representative values.

def inspect_dataframe(df: pd.DataFrame) -> None:
    """Print a compact diagnostic report for a DataFrame."""
    print("\n--- SHAPE ---")
    print(f"Rows: {df.shape[0]}, Columns: {df.shape[1]}")

    print("\n--- FIRST FIVE ROWS ---")
    print(df.head())

    print("\n--- DATA TYPES ---")
    print(df.dtypes)

    print("\n--- DATAFRAME INFO ---")
    df.info()

    print("\n--- MISSING VALUES ---")
    missing = pd.DataFrame({
        "missing_count": df.isna().sum(),
        "missing_percent": df.isna().mean().mul(100).round(2)
    })
    print(missing.sort_values("missing_percent", ascending=False))

    print("\n--- DUPLICATE ROWS ---")
    print(df.duplicated().sum())

    print("\n--- DESCRIPTIVE SUMMARY ---")
    print(df.describe(include="all").transpose())


inspect_dataframe(df)

6. Why Inferred Types May Be Wrong

CSV files do not store rich database-style schemas. pandas infers each column's dtype from its values. A numeric column containing a value such as "unknown" may be loaded as object instead of a numerical dtype.

7. Perform Deterministic Structural Cleaning

Some transformations do not learn statistical information from the dataset. Examples include normalizing column names, converting known invalid tokens to missing values, trimming strings and parsing dates. These deterministic rules may be encapsulated separately from learned preprocessing.

import numpy as np
import pandas as pd


def clean_structure(df: pd.DataFrame) -> pd.DataFrame:
    """
    Apply deterministic cleaning rules.

    This function does not calculate means, medians, modes or scaling
    parameters. It returns a new DataFrame and leaves the input unchanged.
    """
    cleaned = df.copy()

    # Normalize column names.
    cleaned.columns = (
        cleaned.columns
        .str.strip()
        .str.lower()
        .str.replace(r"[^a-z0-9]+", "_", regex=True)
        .str.strip("_")
    )

    # Remove exact duplicate records.
    cleaned = cleaned.drop_duplicates().reset_index(drop=True)

    # Convert known invalid tokens to missing values.
    missing_tokens = ["", " ", "na", "n/a", "none", "null", "unknown", "?"]
    cleaned = cleaned.replace(missing_tokens, np.nan)

    # Convert fields according to the known schema.
    if "monthly_income" in cleaned.columns:
        cleaned["monthly_income"] = pd.to_numeric(
            cleaned["monthly_income"], errors="coerce"
        )

    if "age" in cleaned.columns:
        cleaned["age"] = pd.to_numeric(cleaned["age"], errors="coerce")

    if "signup_date" in cleaned.columns:
        cleaned["signup_date"] = pd.to_datetime(
            cleaned["signup_date"], errors="coerce"
        )

    # Standardize known text fields.
    if "city" in cleaned.columns:
        cleaned["city"] = cleaned["city"].astype("string").str.strip().str.title()

    return cleaned


clean_df = clean_structure(df)
inspect_dataframe(clean_df)

8. Validate the Expected Schema

A preprocessing pipeline should verify required columns and basic assumptions. This prevents a renamed or missing source field from producing a confusing error later.

def validate_schema(df: pd.DataFrame, target_column: str) -> None:
    """Validate required columns and simple business assumptions."""
    required_columns = {
        "customer_id",
        "age",
        "monthly_income",
        "city",
        "signup_date",
        target_column
    }

    missing_columns = required_columns.difference(df.columns)
    if missing_columns:
        raise ValueError(
            "Required columns are missing: "
            + ", ".join(sorted(missing_columns))
        )

    if df[target_column].isna().any():
        raise ValueError("The target column contains missing values.")

    valid_target_values = {0, 1}
    actual_target_values = set(df[target_column].dropna().unique())
    if not actual_target_values.issubset(valid_target_values):
        raise ValueError(
            f"Unexpected target values: {sorted(actual_target_values)}"
        )

    if "age" in df.columns:
        invalid_age = df["age"].dropna().lt(0) | df["age"].dropna().gt(120)
        if invalid_age.any():
            raise ValueError("Age contains values outside the allowed range 0-120.")


TARGET_COLUMN = "purchased"
validate_schema(clean_df, TARGET_COLUMN)

9. Separate Features and Target

The target must not be included in the feature-preprocessing inputs. Otherwise the transformation can accidentally expose the answer to the model.

TARGET_COLUMN = "purchased"

X = clean_df.drop(columns=[TARGET_COLUMN])
y = clean_df[TARGET_COLUMN].astype("int64")

print("Feature shape:", X.shape)
print("Target shape:", y.shape)
print("Target distribution:")
print(y.value_counts(normalize=True).sort_index())

10. Split Before Learning Imputation Values

Perform the train-test split before fitting imputers, encoders or scalers. For classification, stratify=y attempts to preserve the target distribution in both subsets.

from sklearn.model_selection import train_test_split

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

print("X_train:", X_train.shape)
print("X_test:", X_test.shape)
print("y_train:", y_train.shape)
print("y_test:", y_test.shape)

11. Decide Which Columns Enter the Model

An identifier such as customer_id identifies a record but usually should not be treated as a numerical predictive feature. Dates can be transformed into meaningful components such as account age, year, month or day of week.

def build_model_features(df: pd.DataFrame) -> pd.DataFrame:
    """Create deterministic model features from the structurally cleaned data."""
    features = df.copy()

    # Drop a record identifier that should not be learned as a numeric quantity.
    features = features.drop(columns=["customer_id"], errors="ignore")

    # Convert date into model-friendly components.
    if "signup_date" in features.columns:
        features["signup_year"] = features["signup_date"].dt.year
        features["signup_month"] = features["signup_date"].dt.month
        features["signup_day_of_week"] = features["signup_date"].dt.dayofweek
        features = features.drop(columns=["signup_date"])

    return features


X_train_features = build_model_features(X_train)
X_test_features = build_model_features(X_test)

print(X_train_features.dtypes)

12. Inspect Missing Values in Training Data

Review missingness after the split and feature creation. The training subset determines the imputation strategy and learned replacement values.

def missing_value_report(df: pd.DataFrame) -> pd.DataFrame:
    """Return count, percentage and dtype for each column."""
    report = pd.DataFrame({
        "dtype": df.dtypes.astype(str),
        "missing_count": df.isna().sum(),
        "missing_percent": df.isna().mean().mul(100).round(2)
    })

    return report.sort_values(
        ["missing_percent", "missing_count"],
        ascending=False
    )


print(missing_value_report(X_train_features))

13. Select Numeric and Categorical Columns

For this implementation, numeric features receive median imputation and standard scaling. Categorical features receive most-frequent imputation and one-hot encoding.

numeric_features = X_train_features.select_dtypes(
    include=["number"]
).columns.tolist()

categorical_features = X_train_features.select_dtypes(
    include=["object", "string", "category", "bool"]
).columns.tolist()

print("Numeric features:", numeric_features)
print("Categorical features:", categorical_features)

14. Build a Reusable Train-Only Preprocessor

A scikit-learn ColumnTransformer applies different pipelines to different column groups. The object learns values only when fit() or fit_transform() is called.

from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler


def build_preprocessor(
    numeric_features: list[str],
    categorical_features: list[str]
) -> ColumnTransformer:
    """Build an unfitted preprocessing graph for numeric and categorical data."""

    numeric_pipeline = Pipeline(steps=[
        ("imputer", SimpleImputer(strategy="median", add_indicator=True)),
        ("scaler", StandardScaler())
    ])

    categorical_pipeline = Pipeline(steps=[
        ("imputer", SimpleImputer(strategy="most_frequent")),
        (
            "encoder",
            OneHotEncoder(handle_unknown="ignore", sparse_output=False)
        )
    ])

    return ColumnTransformer(
        transformers=[
            ("numeric", numeric_pipeline, numeric_features),
            ("categorical", categorical_pipeline, categorical_features)
        ],
        remainder="drop",
        verbose_feature_names_out=False
    )


preprocessor = build_preprocessor(
    numeric_features=numeric_features,
    categorical_features=categorical_features
)

15. Fit Only on Training Data

This is the most important implementation step. Call fit_transform() on the training features. Call only transform() on the test features.

X_train_processed = preprocessor.fit_transform(X_train_features)
X_test_processed = preprocessor.transform(X_test_features)

feature_names = preprocessor.get_feature_names_out()

X_train_processed = pd.DataFrame(
    X_train_processed,
    columns=feature_names,
    index=X_train_features.index
)

X_test_processed = pd.DataFrame(
    X_test_processed,
    columns=feature_names,
    index=X_test_features.index
)

print("Processed training shape:", X_train_processed.shape)
print("Processed test shape:", X_test_processed.shape)
print(X_train_processed.head())

16. Inspect Learned Training Statistics

The fitted pipelines retain the values learned from the training subset. Inspecting them helps verify the implementation and document the preprocessing behavior.

numeric_imputer = (
    preprocessor
    .named_transformers_["numeric"]
    .named_steps["imputer"]
)

categorical_imputer = (
    preprocessor
    .named_transformers_["categorical"]
    .named_steps["imputer"]
)

print("Numeric columns:", numeric_features)
print("Training medians:", numeric_imputer.statistics_)

print("Categorical columns:", categorical_features)
print("Training category replacements:", categorical_imputer.statistics_)

17. Validate the Processed Output

A preprocessing function should verify that training and test outputs have compatible schemas and contain no unexpected missing values.

def validate_processed_data(
    X_train_processed: pd.DataFrame,
    X_test_processed: pd.DataFrame,
    y_train: pd.Series,
    y_test: pd.Series
) -> None:
    """Run post-transformation quality checks."""

    if list(X_train_processed.columns) != list(X_test_processed.columns):
        raise ValueError("Training and test feature columns do not match.")

    if X_train_processed.isna().any().any():
        raise ValueError("Processed training data still contains missing values.")

    if X_test_processed.isna().any().any():
        raise ValueError("Processed test data still contains missing values.")

    if len(X_train_processed) != len(y_train):
        raise ValueError("Training feature and target row counts do not match.")

    if len(X_test_processed) != len(y_test):
        raise ValueError("Test feature and target row counts do not match.")

    print("Processed data validation passed.")


validate_processed_data(
    X_train_processed,
    X_test_processed,
    y_train,
    y_test
)

18. Complete Runnable Implementation

The following script combines the full workflow into reusable functions. Save it as train_only_preprocessing.py, place the CSV file in the expected location and run it from the terminal or your Python environment.

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path

import joblib
import numpy as np
import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler


@dataclass
class PreparedData:
    """Container returned by the train-only preprocessing workflow."""

    X_train: pd.DataFrame
    X_test: pd.DataFrame
    y_train: pd.Series
    y_test: pd.Series
    preprocessor: ColumnTransformer
    numeric_features: list[str]
    categorical_features: list[str]


def load_csv(file_path: str | Path) -> pd.DataFrame:
    """Load a non-empty CSV with unique column names."""
    path = Path(file_path)

    if not path.exists():
        raise FileNotFoundError(f"CSV file not found: {path.resolve()}")

    if path.suffix.lower() != ".csv":
        raise ValueError(f"Expected a CSV file, received: {path.suffix}")

    df = pd.read_csv(path)

    if df.empty:
        raise ValueError("The CSV file contains no data rows.")

    if not df.columns.is_unique:
        raise ValueError("Duplicate column names were detected.")

    return df


def clean_structure(df: pd.DataFrame) -> pd.DataFrame:
    """Apply deterministic, non-statistical cleaning rules."""
    cleaned = df.copy()

    cleaned.columns = (
        cleaned.columns
        .str.strip()
        .str.lower()
        .str.replace(r"[^a-z0-9]+", "_", regex=True)
        .str.strip("_")
    )

    cleaned = cleaned.drop_duplicates().reset_index(drop=True)

    missing_tokens = ["", " ", "na", "n/a", "none", "null", "unknown", "?"]
    cleaned = cleaned.replace(missing_tokens, np.nan)

    if "age" in cleaned.columns:
        cleaned["age"] = pd.to_numeric(cleaned["age"], errors="coerce")

    if "monthly_income" in cleaned.columns:
        cleaned["monthly_income"] = pd.to_numeric(
            cleaned["monthly_income"], errors="coerce"
        )

    if "city" in cleaned.columns:
        cleaned["city"] = cleaned["city"].astype("string").str.strip().str.title()

    if "signup_date" in cleaned.columns:
        cleaned["signup_date"] = pd.to_datetime(
            cleaned["signup_date"], errors="coerce"
        )

    return cleaned


def build_model_features(df: pd.DataFrame) -> pd.DataFrame:
    """Create deterministic model inputs without learning dataset statistics."""
    features = df.copy()
    features = features.drop(columns=["customer_id"], errors="ignore")

    if "signup_date" in features.columns:
        features["signup_year"] = features["signup_date"].dt.year
        features["signup_month"] = features["signup_date"].dt.month
        features["signup_day_of_week"] = features["signup_date"].dt.dayofweek
        features = features.drop(columns=["signup_date"])

    return features


def build_preprocessor(
    numeric_features: list[str],
    categorical_features: list[str]
) -> ColumnTransformer:
    """Build an unfitted preprocessing object."""

    numeric_pipeline = Pipeline(steps=[
        ("imputer", SimpleImputer(strategy="median", add_indicator=True)),
        ("scaler", StandardScaler())
    ])

    categorical_pipeline = Pipeline(steps=[
        ("imputer", SimpleImputer(strategy="most_frequent")),
        (
            "encoder",
            OneHotEncoder(handle_unknown="ignore", sparse_output=False)
        )
    ])

    return ColumnTransformer(
        transformers=[
            ("numeric", numeric_pipeline, numeric_features),
            ("categorical", categorical_pipeline, categorical_features)
        ],
        remainder="drop",
        verbose_feature_names_out=False
    )


def prepare_train_test_data(
    file_path: str | Path,
    target_column: str,
    test_size: float = 0.25,
    random_state: int = 42
) -> PreparedData:
    """
    Load a CSV and build leakage-safe train and test matrices.

    Learned preprocessing values are fitted exclusively on X_train.
    The fitted object is reused to transform X_test.
    """
    raw_df = load_csv(file_path)
    clean_df = clean_structure(raw_df)

    if target_column not in clean_df.columns:
        raise ValueError(f"Target column not found: {target_column}")

    if clean_df[target_column].isna().any():
        raise ValueError("Target column contains missing values.")

    X = clean_df.drop(columns=[target_column])
    y = clean_df[target_column]

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

    X_train_features = build_model_features(X_train)
    X_test_features = build_model_features(X_test)

    numeric_features = X_train_features.select_dtypes(
        include=["number"]
    ).columns.tolist()

    categorical_features = X_train_features.select_dtypes(
        include=["object", "string", "category", "bool"]
    ).columns.tolist()

    if not numeric_features and not categorical_features:
        raise ValueError("No supported model features were found.")

    preprocessor = build_preprocessor(
        numeric_features=numeric_features,
        categorical_features=categorical_features
    )

    train_array = preprocessor.fit_transform(X_train_features)
    test_array = preprocessor.transform(X_test_features)

    feature_names = preprocessor.get_feature_names_out()

    X_train_processed = pd.DataFrame(
        train_array,
        columns=feature_names,
        index=X_train_features.index
    )

    X_test_processed = pd.DataFrame(
        test_array,
        columns=feature_names,
        index=X_test_features.index
    )

    if X_train_processed.isna().any().any():
        raise ValueError("Processed training data contains missing values.")

    if X_test_processed.isna().any().any():
        raise ValueError("Processed test data contains missing values.")

    if list(X_train_processed.columns) != list(X_test_processed.columns):
        raise ValueError("Processed training and test schemas differ.")

    return PreparedData(
        X_train=X_train_processed,
        X_test=X_test_processed,
        y_train=y_train.copy(),
        y_test=y_test.copy(),
        preprocessor=preprocessor,
        numeric_features=numeric_features,
        categorical_features=categorical_features
    )


def main() -> None:
    prepared = prepare_train_test_data(
        file_path="customer_training_data.csv",
        target_column="purchased",
        test_size=0.25,
        random_state=42
    )

    print("Training shape:", prepared.X_train.shape)
    print("Test shape:", prepared.X_test.shape)
    print("Numeric features:", prepared.numeric_features)
    print("Categorical features:", prepared.categorical_features)
    print("\nProcessed training preview:")
    print(prepared.X_train.head())

    # Persist the fitted preprocessor for later inference.
    joblib.dump(prepared.preprocessor, "customer_preprocessor.joblib")
    print("\nSaved fitted preprocessor: customer_preprocessor.joblib")


if __name__ == "__main__":
    main()

19. Install the Persistence Dependency

The complete implementation uses joblib to save the fitted preprocessing object. It is normally installed as a dependency of scikit-learn, but it can also be installed explicitly.

python -m pip install joblib

20. Run the Script

python train_only_preprocessing.py

The script should print training and test shapes, detected feature groups and a preview of the processed training data. It also saves the fitted preprocessor as customer_preprocessor.joblib.

21. Transform New Unseen Data

Production or future records must pass through the same deterministic feature-building logic and the same fitted preprocessor. Do not fit the preprocessor again.

import joblib
import pandas as pd

# Load the fitted training-only preprocessor.
preprocessor = joblib.load("customer_preprocessor.joblib")

new_customers = pd.DataFrame({
    "customer_id": [201, 202],
    "age": [33, None],
    "monthly_income": [55000, 47000],
    "city": ["Kolkata", "Bengaluru"],
    "signup_date": ["2026-04-01", "2026-04-03"]
})

new_customers = clean_structure(new_customers)
new_features = build_model_features(new_customers)
new_array = preprocessor.transform(new_features)

new_processed = pd.DataFrame(
    new_array,
    columns=preprocessor.get_feature_names_out(),
    index=new_features.index
)

print(new_processed)

22. Prefer a Full Model Pipeline

When a model is added, combine preprocessing and the estimator in one scikit-learn Pipeline. Calling fit() on the pipeline fits preprocessing and the model using training data. Calling predict() automatically reuses the fitted transformations.

from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import Pipeline

model_pipeline = Pipeline(steps=[
    ("preprocessor", build_preprocessor(
        numeric_features=numeric_features,
        categorical_features=categorical_features
    )),
    ("model", LogisticRegression(max_iter=1000))
])

model_pipeline.fit(X_train_features, y_train)
predictions = model_pipeline.predict(X_test_features)

print(predictions)

23. Cross-Validation Without Leakage

For cross-validation, pass the complete pipeline to the validation function. scikit-learn then fits a separate preprocessor within each training fold and transforms that fold's validation subset with the corresponding fold-fitted object.

from sklearn.model_selection import StratifiedKFold, cross_val_score

cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)

scores = cross_val_score(
    model_pipeline,
    X_train_features,
    y_train,
    cv=cv,
    scoring="accuracy"
)

print("Fold scores:", scores)
print("Mean accuracy:", scores.mean())

24. Train-Only Statistical Derivation

Suppose a numerical feature contains training values \(x_1, x_2, \ldots, x_m\). A training mean is:

Standardization of a training or test value uses parameters learned from training data:

The test subset must not contribute to \(\mu_{train}\), \(\sigma_{train}\), a training median or a category vocabulary.

25. Missing-Value Strategy Is a Modeling Decision

The median and most-frequent strategies used in this implementation are reasonable baselines, not universal answers. Choose a strategy based on the source and meaning of missingness.

26. Time-Based Data Requires a Different Split

Random train-test splitting is not appropriate when the objective is to predict the future from the past. For time-dependent records, sort by time and reserve the latest period for validation or testing.

sorted_df = clean_df.sort_values("signup_date").reset_index(drop=True)

split_index = int(len(sorted_df) * 0.80)
train_df = sorted_df.iloc[:split_index].copy()
test_df = sorted_df.iloc[split_index:].copy()

X_train = train_df.drop(columns=[TARGET_COLUMN])
y_train = train_df[TARGET_COLUMN]

X_test = test_df.drop(columns=[TARGET_COLUMN])
y_test = test_df[TARGET_COLUMN]

27. Common Implementation Errors

28. Production Checklist

29. Guided Practice Tasks

30. Final Workflow Summary