executing-plans
executing-plans
Use when executing an implementation plan in the current session as the implementer yourself — your human partner chose inline execution, or no subagent tool is available
ds-model-robustnessImproves model robustness including adversarial robustness, out-of-distribution detection, and uncertainty quantification
Independent GitHub snapshot · Automated classification · Original documentation governs
name: ds-model-robustness compatibility: opencode completeness: 95 content-types:
Comprehensive guide to model robustness in machine learning and data science workflows.
Model Robustness is a critical component of the machine learning workflow. This skill covers:
import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, classification_report
# Generate synthetic dataset with in-distribution and out-of-distribution samples
X_in, y_in = make_classification(n_samples=800, n_features=10, n_classes=2, random_state=42)
X_ood = np.random.uniform(low=-3, high=3, size=(200, 10))
y_ood = np.full(200, -1) # Label for OOD samples
X = np.vstack([X_in, X_ood])
y = np.concatenate([y_in, y_ood])
# Split into training and testing sets
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
# Train a robust ensemble classifier
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)
# Predictions and probability estimates
y_pred = model.predict(X_test)
y_proba = model.predict_proba(X_test)
# Calculate predictive entropy for uncertainty quantification
entropy = -np.sum(y_proba * np.log(y_proba + 1e-10), axis=1)
# OOD detection using Mahalanobis distance approximation via feature variance
train_mean = np.mean(X_train, axis=0)
train_cov = np.cov(X_train.T)
cov_inv = np.linalg.inv(train_cov + 1e-6 * np.eye(train_cov.shape[0]))
mahal_dist = np.array([
(x - train_mean) @ cov_inv @ (x - train_mean).T for x in X_test
])
# Threshold for OOD detection (95th percentile of training distances)
train_mahal = np.array([
(x - train_mean) @ cov_inv @ (x - train_mean).T for x in X_train
])
ood_threshold = np.percentile(train_mahal, 95)
results = {
'accuracy': accuracy_score(y_test, y_pred)
'mean_uncertainty': float(np.mean(entropy))
'ood_samples_detected': int(np.sum(mahal_dist > ood_threshold))
'classification_report': classification_report(y_test, y_pred, zero_division=0)
}
import logging
from typing import Any, Dict, Optional
import numpy as np
import pandas as pd
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import f1_score
logger = logging.getLogger(__name__)
class ModelRobustness:
"""Production implementation of Model Robustness with OOD detection and uncertainty."""
def __init__(self, ood_percentile: float = 0.95, random_state: int = 42):
self.ood_percentile = ood_percentile
self.random_state = random_state
self.model: Optional[GradientBoostingClassifier] = None
self.scaler: Optional[StandardScaler] = None
self.train_stats: Optional[Dict[str, np.ndarray]] = None
def execute(self, data: pd.DataFrame, labels: Optional[np.ndarray] = None) -> Dict[str, Any]:
"""Execute Model Robustness on data"""
if data is None or data.empty:
raise ValueError("Input data cannot be None or empty")
X = data.values
y = labels if labels is not None else None
# Initialize scaler and model if not already fitted
if self.scaler is None:
self.scaler = StandardScaler()
X_scaled = self.scaler.fit_transform(X)
else:
X_scaled = self.scaler.transform(X)
if self.model is None:
self.model = GradientBoostingClassifier(n_estimators=50, random_state=self.random_state)
if y is not None:
self.model.fit(X_scaled, y)
self.train_stats = {
'mean': np.mean(X_scaled, axis=0)
'cov_inv': np.linalg.inv(np.cov(X_scaled.T) + 1e-6 * np.eye(X_scaled.shape[1]))
}
else:
logger.warning("No labels provided for training. Model remains uninitialized.")
return {'status': 'skipped_training', 'data_shape': X.shape}
# Predictions and uncertainty
y_pred = self.model.predict(X_scaled)
y_proba = self.model.predict_proba(X_scaled)
uncertainty = -np.sum(y_proba * np.log(y_proba + 1e-10), axis=1)
# OOD scoring
mahal_dist = np.array([
(x - self.train_stats['mean']) @ self.train_stats['cov_inv'] @ (x - self.train_stats['mean']).T
for x in X_scaled
])
is_ood = mahal_dist > np.percentile(mahal_dist, self.ood_percentile * 100)
metrics = {
'f1_score': float(f1_score(y, y_pred, average='weighted', zero_division=0))
'mean_uncertainty': float(np.mean(uncertainty))
'ood_ratio': float(np.mean(is_ood))
'predictions': y_pred.tolist()
}
return metrics
# BAD: Ignoring uncertainty thresholds and blindly trusting low-confidence predictions
def bad_predict(model, X_test):
predictions = model.predict(X_test)
return predictions # Fails silently on OOD or high-entropy samples
# GOOD: Explicit uncertainty gating and OOD rejection per OWASP ML Security Top 10
def good_predict_with_robustness(model, X_test, scaler, threshold: float = 0.85):
if X_test is None or X_test.size == 0:
raise ValueError("Input tensor cannot be empty")
X_scaled = scaler.transform(X_test)
y_proba = model.predict_proba(X_scaled)
confidence = np.max(y_proba, axis=1)
entropy = -np.sum(y_proba * np.log(y_proba + 1e-10), axis=1)
# Reject samples below confidence threshold or above entropy threshold
mask = (confidence >= threshold) & (entropy <= np.percentile(entropy, 90))
safe_predictions = model.predict(X_scaled[mask])
rejected_count = int(np.sum(~mask))
return safe_predictions, rejected_count
| Pitfall | Problem | Solution | |
Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.