Spaces:

yusufcalisir
/

Collaborative-Fraud-Intelligence-Simulator

Running

Collaborative-Fraud-Intelligence-Simulator / backend /tests /unit /test_split_isolation.py

deploy: Hugging Face space upload

eff6797 3 days ago

7.82 kB

"""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

	"""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