# 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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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
