tools-only 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 validationexperiment-tracking.md- Logging validation resultspipeline-orchestration.md- Automated validation workflowsfeature-engineering.md- Feature validation
Cross-Reference Skills
- Data Engineer - Data quality validation
- DevOps Engineer - Deployment pipeline integration