Model Validation

Model Validation

Overview

Model validation ensures models meet quality standards before production deployment. It encompasses offline evaluation, online testing, and continuous monitoring to catch performance degradation, data drift, and model failures.

When to Use This Reference

• Implementing offline model evaluation strategies
• Setting up A/B testing frameworks
• Building shadow deployment pipelines
• Creating model comparison workflows
• Implementing continuous model monitoring

When NOT to Use

• Quick model prototyping
• One-off analysis without deployment
• Models with no production requirements

Offline Evaluation

Comprehensive Evaluation Suite

from dataclasses import dataclass
from typing import Optional
import numpy as np
import pandas as pd
from sklearn.metrics import (
    accuracy_score, precision_score, recall_score, f1_score,
    roc_auc_score, average_precision_score, confusion_matrix,
    mean_squared_error, mean_absolute_error, r2_score,
)

@dataclass
class ClassificationMetrics:
    """Classification model metrics."""
    accuracy: float
    precision: float
    recall: float
    f1: float
    roc_auc: Optional[float]
    pr_auc: Optional[float]
    confusion_matrix: np.ndarray

    def to_dict(self) -> dict:
        return {
            "accuracy": self.accuracy,
            "precision": self.precision,
            "recall": self.recall,
            "f1": self.f1,
            "roc_auc": self.roc_auc,
            "pr_auc": self.pr_auc,
        }

@dataclass
class RegressionMetrics:
    """Regression model metrics."""
    mse: float
    rmse: float
    mae: float
    r2: float
    mape: Optional[float]

    def to_dict(self) -> dict:
        return {
            "mse": self.mse,
            "rmse": self.rmse,
            "mae": self.mae,
            "r2": self.r2,
            "mape": self.mape,
        }

class ModelEvaluator:
    """Comprehensive model evaluation."""

    def __init__(self, task_type: str = "classification"):
        self.task_type = task_type

    def evaluate_classification(
        self,
        y_true: np.ndarray,
        y_pred: np.ndarray,
        y_prob: Optional[np.ndarray] = None,
        average: str = "weighted",
    ) -> ClassificationMetrics:
        """Evaluate classification model."""
        roc_auc = None
        pr_auc = None

        if y_prob is not None:
            if len(np.unique(y_true)) == 2:
                # Binary classification
                if y_prob.ndim == 2:
                    y_prob_pos = y_prob[:, 1]
                else:
                    y_prob_pos = y_prob
                roc_auc = roc_auc_score(y_true, y_prob_pos)
                pr_auc = average_precision_score(y_true, y_prob_pos)
            else:
                # Multiclass
                roc_auc = roc_auc_score(
                    y_true, y_prob, multi_class="ovr", average=average
                )

        return ClassificationMetrics(
            accuracy=accuracy_score(y_true, y_pred),
            precision=precision_score(y_true, y_pred, average=average, zero_division=0),
            recall=recall_score(y_true, y_pred, average=average, zero_division=0),
            f1=f1_score(y_true, y_pred, average=average, zero_division=0),
            roc_auc=roc_auc,
            pr_auc=pr_auc,
            confusion_matrix=confusion_matrix(y_true, y_pred),
        )

    def evaluate_regression(
        self,
        y_true: np.ndarray,
        y_pred: np.ndarray,
    ) -> RegressionMetrics:
        """Evaluate regression model."""
        mse = mean_squared_error(y_true, y_pred)

        # MAPE (handle zero values)
        mask = y_true != 0
        if mask.any():
            mape = np.mean(np.abs((y_true[mask] - y_pred[mask]) / y_true[mask])) * 100
        else:
            mape = None

        return RegressionMetrics(
            mse=mse,
            rmse=np.sqrt(mse),
            mae=mean_absolute_error(y_true, y_pred),
            r2=r2_score(y_true, y_pred),
            mape=mape,
        )

    def evaluate_by_segment(
        self,
        y_true: np.ndarray,
        y_pred: np.ndarray,
        segments: np.ndarray,
        y_prob: Optional[np.ndarray] = None,
    ) -> dict:
        """Evaluate model performance by segment."""
        results = {}

        for segment in np.unique(segments):
            mask = segments == segment

            if self.task_type == "classification":
                segment_prob = y_prob[mask] if y_prob is not None else None
                metrics = self.evaluate_classification(
                    y_true[mask], y_pred[mask], segment_prob
                )
            else:
                metrics = self.evaluate_regression(y_true[mask], y_pred[mask])

            results[segment] = metrics.to_dict()

        return results

Cross-Validation Framework

from sklearn.model_selection import (
    KFold, StratifiedKFold, TimeSeriesSplit, cross_val_score
)
import numpy as np
from typing import Callable

class CrossValidator:
    """Cross-validation framework for model evaluation."""

    def __init__(
        self,
        n_splits: int = 5,
        shuffle: bool = True,
        random_state: int = 42,
    ):
        self.n_splits = n_splits
        self.shuffle = shuffle
        self.random_state = random_state

    def validate_classification(
        self,
        model,
        X: np.ndarray,
        y: np.ndarray,
        stratified: bool = True,
    ) -> dict:
        """Run stratified k-fold cross-validation for classification."""
        if stratified:
            cv = StratifiedKFold(
                n_splits=self.n_splits,
                shuffle=self.shuffle,
                random_state=self.random_state,
            )
        else:
            cv = KFold(
                n_splits=self.n_splits,
                shuffle=self.shuffle,
                random_state=self.random_state,
            )

        evaluator = ModelEvaluator("classification")
        fold_metrics = []

        for fold, (train_idx, val_idx) in enumerate(cv.split(X, y)):
            X_train, X_val = X[train_idx], X[val_idx]
            y_train, y_val = y[train_idx], y[val_idx]

            # Clone and train model
            from sklearn.base import clone
            fold_model = clone(model)
            fold_model.fit(X_train, y_train)

            y_pred = fold_model.predict(X_val)
            y_prob = None
            if hasattr(fold_model, "predict_proba"):
                y_prob = fold_model.predict_proba(X_val)

            metrics = evaluator.evaluate_classification(y_val, y_pred, y_prob)
            fold_metrics.append(metrics.to_dict())

        return self._aggregate_cv_results(fold_metrics)

    def validate_time_series(
        self,
        model,
        X: np.ndarray,
        y: np.ndarray,
        gap: int = 0,
    ) -> dict:
        """Run time series cross-validation."""
        cv = TimeSeriesSplit(n_splits=self.n_splits, gap=gap)
        evaluator = ModelEvaluator("regression")
        fold_metrics = []

        for train_idx, val_idx in cv.split(X):
            X_train, X_val = X[train_idx], X[val_idx]
            y_train, y_val = y[train_idx], y[val_idx]

            from sklearn.base import clone
            fold_model = clone(model)
            fold_model.fit(X_train, y_train)

            y_pred = fold_model.predict(X_val)
            metrics = evaluator.evaluate_regression(y_val, y_pred)
            fold_metrics.append(metrics.to_dict())

        return self._aggregate_cv_results(fold_metrics)

    def _aggregate_cv_results(self, fold_metrics: list[dict]) -> dict:
        """Aggregate metrics across folds."""
        keys = fold_metrics[0].keys()
        aggregated = {}

        for key in keys:
            values = [m[key] for m in fold_metrics if m[key] is not None]
            if values:
                aggregated[key] = {
                    "mean": np.mean(values),
                    "std": np.std(values),
                    "min": np.min(values),
                    "max": np.max(values),
                    "values": values,
                }

        return aggregated

Model Comparison

Statistical Comparison

from scipy import stats
import numpy as np
from dataclasses import dataclass

@dataclass
class ComparisonResult:
    """Model comparison statistical result."""
    model_a_mean: float
    model_b_mean: float
    difference: float
    p_value: float
    significant: bool
    confidence_interval: tuple[float, float]
    test_used: str

class ModelComparator:
    """Statistical comparison of model performance."""

    def __init__(self, significance_level: float = 0.05):
        self.significance_level = significance_level

    def paired_t_test(
        self,
        scores_a: np.ndarray,
        scores_b: np.ndarray,
    ) -> ComparisonResult:
        """Paired t-test for CV score comparison."""
        statistic, p_value = stats.ttest_rel(scores_a, scores_b)

        differences = scores_a - scores_b
        mean_diff = np.mean(differences)
        std_diff = np.std(differences, ddof=1)
        n = len(differences)

        # 95% confidence interval
        t_critical = stats.t.ppf(1 - self.significance_level / 2, n - 1)
        margin = t_critical * std_diff / np.sqrt(n)
        ci = (mean_diff - margin, mean_diff + margin)

        return ComparisonResult(
            model_a_mean=np.mean(scores_a),
            model_b_mean=np.mean(scores_b),
            difference=mean_diff,
            p_value=p_value,
            significant=p_value < self.significance_level,
            confidence_interval=ci,
            test_used="paired_t_test",
        )

    def wilcoxon_test(
        self,
        scores_a: np.ndarray,
        scores_b: np.ndarray,
    ) -> ComparisonResult:
        """Wilcoxon signed-rank test (non-parametric)."""
        statistic, p_value = stats.wilcoxon(scores_a, scores_b)

        differences = scores_a - scores_b
        mean_diff = np.mean(differences)

        # Bootstrap confidence interval
        ci = self._bootstrap_ci(differences)

        return ComparisonResult(
            model_a_mean=np.mean(scores_a),
            model_b_mean=np.mean(scores_b),
            difference=mean_diff,
            p_value=p_value,
            significant=p_value < self.significance_level,
            confidence_interval=ci,
            test_used="wilcoxon",
        )

    def mcnemar_test(
        self,
        y_true: np.ndarray,
        pred_a: np.ndarray,
        pred_b: np.ndarray,
    ) -> ComparisonResult:
        """McNemar's test for classifier comparison."""
        # Build contingency table
        correct_a = (pred_a == y_true)
        correct_b = (pred_b == y_true)

        # b: A correct, B wrong; c: A wrong, B correct
        b = np.sum(correct_a & ~correct_b)
        c = np.sum(~correct_a & correct_b)

        if b + c < 25:
            # Use exact binomial test for small samples
            p_value = stats.binom_test(b, b + c, 0.5)
        else:
            # Use chi-square approximation
            statistic = (abs(b - c) - 1) ** 2 / (b + c)
            p_value = 1 - stats.chi2.cdf(statistic, 1)

        acc_a = np.mean(correct_a)
        acc_b = np.mean(correct_b)

        return ComparisonResult(
            model_a_mean=acc_a,
            model_b_mean=acc_b,
            difference=acc_a - acc_b,
            p_value=p_value,
            significant=p_value < self.significance_level,
            confidence_interval=(None, None),
            test_used="mcnemar",
        )

    def _bootstrap_ci(
        self,
        data: np.ndarray,
        n_bootstrap: int = 10000,
        alpha: float = 0.05,
    ) -> tuple[float, float]:
        """Calculate bootstrap confidence interval."""
        bootstrapped_means = []

        for _ in range(n_bootstrap):
            sample = np.random.choice(data, size=len(data), replace=True)
            bootstrapped_means.append(np.mean(sample))

        lower = np.percentile(bootstrapped_means, alpha / 2 * 100)
        upper = np.percentile(bootstrapped_means, (1 - alpha / 2) * 100)

        return (lower, upper)

A/B Testing

Online Experiment Framework

from dataclasses import dataclass
from datetime import datetime
from typing import Optional
import numpy as np
import hashlib
import json

@dataclass
class Experiment:
    """A/B test experiment configuration."""
    experiment_id: str
    name: str
    control_model: str
    treatment_model: str
    traffic_split: float  # Fraction to treatment
    start_time: datetime
    end_time: Optional[datetime]
    metrics: list[str]
    minimum_sample_size: int
    status: str = "active"

class ABTestRouter:
    """Route traffic between control and treatment."""

    def __init__(self, experiment: Experiment):
        self.experiment = experiment

    def get_variant(self, user_id: str) -> str:
        """Deterministically assign user to variant."""
        # Hash user_id for consistent assignment
        hash_input = f"{self.experiment.experiment_id}:{user_id}"
        hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16)
        normalized = hash_value / (2**128)

        if normalized < self.experiment.traffic_split:
            return "treatment"
        return "control"

    def get_model(self, user_id: str) -> str:
        """Get model to use for user."""
        variant = self.get_variant(user_id)

        if variant == "treatment":
            return self.experiment.treatment_model
        return self.experiment.control_model

class ABTestAnalyzer:
    """Analyze A/B test results."""

    def __init__(self, significance_level: float = 0.05):
        self.significance_level = significance_level

    def analyze_conversion(
        self,
        control_conversions: int,
        control_total: int,
        treatment_conversions: int,
        treatment_total: int,
    ) -> dict:
        """Analyze conversion rate experiment."""
        control_rate = control_conversions / control_total
        treatment_rate = treatment_conversions / treatment_total

        # Two-proportion z-test
        pooled_rate = (control_conversions + treatment_conversions) / (
            control_total + treatment_total
        )
        se = np.sqrt(
            pooled_rate * (1 - pooled_rate) * (1/control_total + 1/treatment_total)
        )

        z_stat = (treatment_rate - control_rate) / se
        p_value = 2 * (1 - stats.norm.cdf(abs(z_stat)))

        # Relative lift
        lift = (treatment_rate - control_rate) / control_rate if control_rate > 0 else 0

        # Confidence interval for difference
        se_diff = np.sqrt(
            control_rate * (1 - control_rate) / control_total +
            treatment_rate * (1 - treatment_rate) / treatment_total
        )
        z_critical = stats.norm.ppf(1 - self.significance_level / 2)
        ci = (
            (treatment_rate - control_rate) - z_critical * se_diff,
            (treatment_rate - control_rate) + z_critical * se_diff,
        )

        return {
            "control_rate": control_rate,
            "treatment_rate": treatment_rate,
            "absolute_difference": treatment_rate - control_rate,
            "relative_lift": lift,
            "p_value": p_value,
            "significant": p_value < self.significance_level,
            "confidence_interval": ci,
            "control_sample_size": control_total,
            "treatment_sample_size": treatment_total,
        }

    def analyze_continuous_metric(
        self,
        control_values: np.ndarray,
        treatment_values: np.ndarray,
    ) -> dict:
        """Analyze continuous metric (e.g., revenue, time)."""
        control_mean = np.mean(control_values)
        treatment_mean = np.mean(treatment_values)

        # Welch's t-test (unequal variances)
        statistic, p_value = stats.ttest_ind(
            treatment_values, control_values, equal_var=False
        )

        lift = (treatment_mean - control_mean) / control_mean if control_mean > 0 else 0

        # Confidence interval
        se_diff = np.sqrt(
            np.var(control_values) / len(control_values) +
            np.var(treatment_values) / len(treatment_values)
        )
        t_critical = stats.t.ppf(
            1 - self.significance_level / 2,
            min(len(control_values), len(treatment_values)) - 1
        )
        ci = (
            (treatment_mean - control_mean) - t_critical * se_diff,
            (treatment_mean - control_mean) + t_critical * se_diff,
        )

        return {
            "control_mean": control_mean,
            "treatment_mean": treatment_mean,
            "absolute_difference": treatment_mean - control_mean,
            "relative_lift": lift,
            "p_value": p_value,
            "significant": p_value < self.significance_level,
            "confidence_interval": ci,
            "control_sample_size": len(control_values),
            "treatment_sample_size": len(treatment_values),
        }

    def calculate_sample_size(
        self,
        baseline_rate: float,
        minimum_detectable_effect: float,
        power: float = 0.8,
    ) -> int:
        """Calculate required sample size per variant."""
        alpha = self.significance_level
        z_alpha = stats.norm.ppf(1 - alpha / 2)
        z_beta = stats.norm.ppf(power)

        p1 = baseline_rate
        p2 = baseline_rate * (1 + minimum_detectable_effect)

        p_bar = (p1 + p2) / 2

        n = (
            (z_alpha * np.sqrt(2 * p_bar * (1 - p_bar)) +
             z_beta * np.sqrt(p1 * (1 - p1) + p2 * (1 - p2))) ** 2 /
            (p2 - p1) ** 2
        )

        return int(np.ceil(n))

Shadow Deployment

Shadow Mode Pipeline

from dataclasses import dataclass
from datetime import datetime
from typing import Any, Optional
import logging
import json

logger = logging.getLogger(__name__)

@dataclass
class PredictionComparison:
    """Comparison of production and shadow predictions."""
    request_id: str
    timestamp: datetime
    production_prediction: Any
    shadow_prediction: Any
    production_latency_ms: float
    shadow_latency_ms: float
    agreement: bool
    features: Optional[dict] = None

class ShadowDeployment:
    """Shadow deployment for model validation."""

    def __init__(
        self,
        production_model,
        shadow_model,
        log_path: str = "/var/log/shadow_predictions.jsonl",
    ):
        self.production_model = production_model
        self.shadow_model = shadow_model
        self.log_path = log_path
        self.comparisons: list[PredictionComparison] = []

    def predict(
        self,
        features: dict,
        request_id: str = None,
    ) -> Any:
        """Get production prediction, run shadow in parallel."""
        import time
        import uuid
        import concurrent.futures

        request_id = request_id or str(uuid.uuid4())

        # Production prediction (synchronous, used for response)
        prod_start = time.time()
        production_pred = self.production_model.predict(features)
        prod_latency = (time.time() - prod_start) * 1000

        # Shadow prediction (async, logged but not returned)
        def run_shadow():
            shadow_start = time.time()
            shadow_pred = self.shadow_model.predict(features)
            shadow_latency = (time.time() - shadow_start) * 1000
            return shadow_pred, shadow_latency

        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
            future = executor.submit(run_shadow)

            try:
                shadow_pred, shadow_latency = future.result(timeout=5.0)

                comparison = PredictionComparison(
                    request_id=request_id,
                    timestamp=datetime.utcnow(),
                    production_prediction=production_pred,
                    shadow_prediction=shadow_pred,
                    production_latency_ms=prod_latency,
                    shadow_latency_ms=shadow_latency,
                    agreement=self._check_agreement(production_pred, shadow_pred),
                    features=features,
                )

                self._log_comparison(comparison)

            except concurrent.futures.TimeoutError:
                logger.warning(f"Shadow prediction timed out for {request_id}")

        return production_pred

    def _check_agreement(self, prod_pred: Any, shadow_pred: Any) -> bool:
        """Check if predictions agree."""
        if isinstance(prod_pred, (list, np.ndarray)):
            return np.allclose(prod_pred, shadow_pred, rtol=1e-3)
        return prod_pred == shadow_pred

    def _log_comparison(self, comparison: PredictionComparison) -> None:
        """Log comparison to file."""
        log_entry = {
            "request_id": comparison.request_id,
            "timestamp": comparison.timestamp.isoformat(),
            "production_prediction": str(comparison.production_prediction),
            "shadow_prediction": str(comparison.shadow_prediction),
            "production_latency_ms": comparison.production_latency_ms,
            "shadow_latency_ms": comparison.shadow_latency_ms,
            "agreement": comparison.agreement,
        }

        with open(self.log_path, "a") as f:
            f.write(json.dumps(log_entry) + "\n")

        self.comparisons.append(comparison)

    def analyze_shadow_performance(self) -> dict:
        """Analyze shadow model performance."""
        if not self.comparisons:
            return {}

        agreements = [c.agreement for c in self.comparisons]
        prod_latencies = [c.production_latency_ms for c in self.comparisons]
        shadow_latencies = [c.shadow_latency_ms for c in self.comparisons]

        return {
            "total_comparisons": len(self.comparisons),
            "agreement_rate": np.mean(agreements),
            "production_latency_p50": np.percentile(prod_latencies, 50),
            "production_latency_p99": np.percentile(prod_latencies, 99),
            "shadow_latency_p50": np.percentile(shadow_latencies, 50),
            "shadow_latency_p99": np.percentile(shadow_latencies, 99),
            "latency_difference_mean": np.mean(
                [s - p for s, p in zip(shadow_latencies, prod_latencies)]
            ),
        }

Validation Pipeline Integration

Complete Validation Workflow

from enum import Enum
from dataclasses import dataclass
from typing import Optional

class ValidationStatus(Enum):
    PASSED = "passed"
    FAILED = "failed"
    WARNING = "warning"

@dataclass
class ValidationResult:
    """Result of a validation check."""
    check_name: str
    status: ValidationStatus
    message: str
    details: Optional[dict] = None

class ModelValidator:
    """Complete model validation workflow."""

    def __init__(
        self,
        accuracy_threshold: float = 0.8,
        latency_threshold_ms: float = 100,
        drift_threshold: float = 0.2,
    ):
        self.accuracy_threshold = accuracy_threshold
        self.latency_threshold_ms = latency_threshold_ms
        self.drift_threshold = drift_threshold
        self.results: list[ValidationResult] = []

    def validate_performance(
        self,
        y_true: np.ndarray,
        y_pred: np.ndarray,
    ) -> ValidationResult:
        """Validate model performance metrics."""
        evaluator = ModelEvaluator("classification")
        metrics = evaluator.evaluate_classification(y_true, y_pred)

        if metrics.accuracy >= self.accuracy_threshold:
            status = ValidationStatus.PASSED
            message = f"Accuracy {metrics.accuracy:.4f} meets threshold"
        else:
            status = ValidationStatus.FAILED
            message = f"Accuracy {metrics.accuracy:.4f} below threshold {self.accuracy_threshold}"

        result = ValidationResult(
            check_name="performance",
            status=status,
            message=message,
            details=metrics.to_dict(),
        )
        self.results.append(result)
        return result

    def validate_latency(
        self,
        model,
        sample_input: np.ndarray,
        n_iterations: int = 100,
    ) -> ValidationResult:
        """Validate inference latency."""
        import time

        latencies = []
        for _ in range(n_iterations):
            start = time.time()
            model.predict(sample_input)
            latencies.append((time.time() - start) * 1000)

        p50 = np.percentile(latencies, 50)
        p99 = np.percentile(latencies, 99)

        if p99 <= self.latency_threshold_ms:
            status = ValidationStatus.PASSED
            message = f"P99 latency {p99:.2f}ms meets threshold"
        elif p50 <= self.latency_threshold_ms:
            status = ValidationStatus.WARNING
            message = f"P50 OK but P99 {p99:.2f}ms exceeds threshold"
        else:
            status = ValidationStatus.FAILED
            message = f"P99 latency {p99:.2f}ms exceeds threshold"

        result = ValidationResult(
            check_name="latency",
            status=status,
            message=message,
            details={"p50_ms": p50, "p99_ms": p99, "mean_ms": np.mean(latencies)},
        )
        self.results.append(result)
        return result

    def validate_data_compatibility(
        self,
        model,
        expected_features: list[str],
        sample_data: pd.DataFrame,
    ) -> ValidationResult:
        """Validate model accepts expected input format."""
        missing_features = set(expected_features) - set(sample_data.columns)
        extra_features = set(sample_data.columns) - set(expected_features)

        if missing_features:
            status = ValidationStatus.FAILED
            message = f"Missing features: {missing_features}"
        elif extra_features:
            status = ValidationStatus.WARNING
            message = f"Extra features will be ignored: {extra_features}"
        else:
            status = ValidationStatus.PASSED
            message = "All expected features present"

        # Try inference
        try:
            model.predict(sample_data[expected_features].head(1))
        except Exception as e:
            status = ValidationStatus.FAILED
            message = f"Inference failed: {str(e)}"

        result = ValidationResult(
            check_name="data_compatibility",
            status=status,
            message=message,
            details={
                "missing_features": list(missing_features),
                "extra_features": list(extra_features),
            },
        )
        self.results.append(result)
        return result

    def validate_vs_baseline(
        self,
        y_true: np.ndarray,
        new_pred: np.ndarray,
        baseline_pred: np.ndarray,
    ) -> ValidationResult:
        """Validate new model vs baseline."""
        comparator = ModelComparator()
        comparison = comparator.mcnemar_test(y_true, new_pred, baseline_pred)

        new_acc = accuracy_score(y_true, new_pred)
        baseline_acc = accuracy_score(y_true, baseline_pred)

        if new_acc >= baseline_acc:
            if comparison.significant:
                status = ValidationStatus.PASSED
                message = f"Significant improvement: {new_acc:.4f} vs {baseline_acc:.4f}"
            else:
                status = ValidationStatus.WARNING
                message = f"Improvement not significant: {new_acc:.4f} vs {baseline_acc:.4f}"
        else:
            if comparison.significant:
                status = ValidationStatus.FAILED
                message = f"Significant regression: {new_acc:.4f} vs {baseline_acc:.4f}"
            else:
                status = ValidationStatus.WARNING
                message = f"Minor regression: {new_acc:.4f} vs {baseline_acc:.4f}"

        result = ValidationResult(
            check_name="baseline_comparison",
            status=status,
            message=message,
            details={
                "new_accuracy": new_acc,
                "baseline_accuracy": baseline_acc,
                "p_value": comparison.p_value,
            },
        )
        self.results.append(result)
        return result

    def get_summary(self) -> dict:
        """Get validation summary."""
        passed = sum(1 for r in self.results if r.status == ValidationStatus.PASSED)
        warnings = sum(1 for r in self.results if r.status == ValidationStatus.WARNING)
        failed = sum(1 for r in self.results if r.status == ValidationStatus.FAILED)

        overall_status = (
            ValidationStatus.FAILED if failed > 0
            else ValidationStatus.WARNING if warnings > 0
            else ValidationStatus.PASSED
        )

        return {
            "overall_status": overall_status.value,
            "passed": passed,
            "warnings": warnings,
            "failed": failed,
            "results": [
                {
                    "check": r.check_name,
                    "status": r.status.value,
                    "message": r.message,
                }
                for r in self.results
            ],
        }

Best Practices

Validation Checklist

VALIDATION_CHECKLIST = {
    "offline": [
        "Accuracy/performance metrics meet threshold",
        "Cross-validation shows consistent performance",
        "Model outperforms or matches baseline",
        "Metrics stable across data segments",
    ],
    "pre_deployment": [
        "Inference latency within SLA",
        "Memory usage acceptable",
        "Input/output schema validated",
        "Model serialization/loading works",
    ],
    "shadow": [
        "Shadow predictions logged successfully",
        "Agreement rate with production acceptable",
        "No latency regression",
        "Error rate within bounds",
    ],
    "ab_test": [
        "Sufficient sample size reached",
        "Statistical significance achieved",
        "No negative impact on guardrail metrics",
        "Business metrics improved",
    ],
}

Related References

• training-pipelines.md - Model training before validation
• experiment-tracking.md - Logging validation results
• pipeline-orchestration.md - Automated validation workflows
• feature-engineering.md - Feature validation

Cross-Reference Skills

• Data Engineer - Data quality validation
• DevOps Engineer - Deployment pipeline integration