File size: 7,820 Bytes
eff6797 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 | """Unit tests for Data Preprocessor and Temporal Split Isolation.
Verifies:
- Strict train-split isolation (parameters learned only from training partition).
- Handling of unseen test categorical tokens (mapped to zero-vector without crashing).
- Zero data snooping (outliers/distributions in test have 0% effect on train statistics).
- Chronological temporal ordering (t_train_max <= t_val_min <= t_test_min).
- Full state serialization and deserialization fidelity.
"""
from __future__ import annotations
import numpy as np
import pandas as pd
import pytest
from app.application.services.dataloader import temporal_split_dataset
from app.application.services.feature_service import FeatureService
from app.application.services.preprocessor import DataPreprocessor, PreprocessingStateError
class TestDataPreprocessorIsolation:
"""Test suite ensuring zero-leakage preprocessing and split isolation."""
def test_preprocessor_not_fitted_raises_error(self) -> None:
"""Calling transform before fit must raise PreprocessingStateError."""
prep = DataPreprocessor()
df = pd.DataFrame({"col1": [1.0, 2.0, 3.0]})
with pytest.raises(PreprocessingStateError, match="must be fitted before transform"):
prep.transform(df)
def test_preprocessor_fit_transform_strict_isolation(self) -> None:
"""Preprocessor statistics must derive strictly from training set."""
train_data = pd.DataFrame({
"amount": [100.0, 200.0, 300.0],
"balance": [1000.0, 2000.0, 3000.0],
})
test_data = pd.DataFrame({
"amount": [999999.0, 888888.0], # Extreme outliers in test set
"balance": [5000.0, 6000.0],
})
prep = DataPreprocessor(numeric_strategy="standardize", impute_strategy="mean")
X_train_transformed = prep.fit_transform(train_data)
# Means and stds must reflect ONLY train_data
assert prep.means_["amount"] == pytest.approx(200.0)
assert prep.means_["balance"] == pytest.approx(2000.0)
assert prep.stds_["amount"] > 0
assert prep.stds_["balance"] > 0
# Transforming test data must use train statistics without mutating them
X_test_transformed = prep.transform(test_data)
# Verify mean of amount has NOT changed
assert prep.means_["amount"] == pytest.approx(200.0)
assert X_train_transformed.shape == (3, 2)
assert X_test_transformed.shape == (2, 2)
def test_unseen_categorical_vocabulary_isolation(self) -> None:
"""Categories present only in test set must be mapped to zero-vector without leaking."""
train_df = pd.DataFrame({
"amount": [10.0, 20.0, 30.0],
"channel": ["MOBILE", "WEB", "MOBILE"],
})
test_df = pd.DataFrame({
"amount": [40.0, 50.0],
"channel": ["ATM", "BRANCH"], # Completely unseen in train
})
prep = DataPreprocessor()
X_train = prep.fit_transform(train_df)
X_test = prep.transform(test_df)
assert "MOBILE" in prep.categories_["channel"]
assert "WEB" in prep.categories_["channel"]
assert "ATM" not in prep.categories_["channel"]
assert "BRANCH" not in prep.categories_["channel"]
# Number of output features: 1 numeric ("amount") + 2 categories ("MOBILE", "WEB")
assert prep.feature_names_out_ == ["amount", "channel_MOBILE", "channel_WEB"]
assert X_train.shape[1] == 3
assert X_test.shape[1] == 3
# Test rows for ATM and BRANCH must have zeros for all train category indicators
assert np.all(X_test[:, 1:] == 0.0)
def test_preprocessor_outlier_clipping(self) -> None:
"""Outlier clipping bounds transformed values to clip_std_factor."""
train_df = pd.DataFrame({"feat": [0.0, 1.0, 2.0, 3.0, 4.0]})
test_df = pd.DataFrame({"feat": [10000.0, -10000.0]})
prep = DataPreprocessor(numeric_strategy="standardize", clip_outliers=True, clip_std_factor=5.0)
prep.fit(train_df)
X_test = prep.transform(test_df)
assert np.all(X_test <= 5.0)
assert np.all(X_test >= -5.0)
def test_serialization_fidelity(self) -> None:
"""Serialized preprocessor state must reconstitute identical transformation behavior."""
df = pd.DataFrame({
"amount": [100.0, 200.0, np.nan, 400.0],
"category": ["TRANSFER", "CASH_OUT", "TRANSFER", "PAYMENT"],
})
prep = DataPreprocessor(numeric_strategy="minmax", impute_strategy="median")
X_transformed_orig = prep.fit_transform(df)
state_dict = prep.to_dict()
restored_prep = DataPreprocessor.from_dict(state_dict)
X_transformed_restored = restored_prep.transform(df)
np.testing.assert_allclose(X_transformed_orig, X_transformed_restored, rtol=1e-5)
class TestTemporalSplitIsolation:
"""Test suite ensuring chronological arrow of time is enforced without lookahead."""
def test_temporal_split_monotonic_order(self) -> None:
"""Verifies t_train_max <= t_val_min <= t_test_min."""
n_samples = 100
df = pd.DataFrame({
"step": np.random.default_rng(42).permutation(np.arange(n_samples)),
"amount": np.random.default_rng(42).uniform(10, 500, size=n_samples),
"isFraud": (np.random.default_rng(42).random(n_samples) < 0.1).astype(int),
})
train_df, val_df, test_df, meta = FeatureService.temporal_split(
df,
time_col="step",
train_ratio=0.70,
val_ratio=0.15,
test_ratio=0.15,
label_col="isFraud",
)
assert meta.is_strictly_chronological is True
assert len(train_df) == 70
assert len(val_df) == 15
assert len(test_df) == 15
# Check temporal order boundaries
assert meta.train_time_max is not None and meta.val_time_min is not None
assert meta.val_time_max is not None and meta.test_time_min is not None
assert float(meta.train_time_max) <= float(meta.val_time_min)
assert float(meta.val_time_max) <= float(meta.test_time_min)
# Check internal ascending order
assert train_df["step"].is_monotonic_increasing
assert val_df["step"].is_monotonic_increasing
assert test_df["step"].is_monotonic_increasing
def test_temporal_split_boundary_overlap_detection(self) -> None:
"""Boundary overlap is flagged when boundary timestamps match across partitions."""
# 10 records all at timestamp 5.0
df = pd.DataFrame({
"step": [5.0] * 10,
"amount": list(range(10)),
})
_, _, _, meta = FeatureService.temporal_split(
df,
time_col="step",
train_ratio=0.6,
val_ratio=0.2,
test_ratio=0.2,
)
assert meta.boundary_overlap is True
def test_dataloader_temporal_split_integration(self) -> None:
"""Verifies dataloader temporal split helper on PaySim mock dataset."""
from app.application.services.synthetic_dataset_generators import generate_synthetic_paysim
paysim_data = generate_synthetic_paysim(n_mock_txns=200)
split_result = temporal_split_dataset(
paysim_data,
train_ratio=0.70,
val_ratio=0.15,
test_ratio=0.15,
preprocess=True,
)
assert split_result["is_strictly_chronological"] is True
assert split_result["X_train"].shape[0] == 140
assert split_result["X_val"].shape[0] == 30
assert split_result["X_test"].shape[0] == 30
assert split_result["preprocessor"] is not None
assert split_result["preprocessor"].is_fitted is True
|