Spaces:
Sleeping
Sleeping
File size: 5,374 Bytes
57b5041 3a77c6f 57b5041 3a77c6f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 | """
Model Performance Monitor
=========================
Trains a lightweight RandomForest on reference data, then evaluates
it on production data to surface concept drift via accuracy degradation.
"""
import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import (
accuracy_score,
precision_score,
recall_score,
f1_score,
roc_auc_score,
)
class ModelMonitor:
"""Simulates a trained ML model and monitors its production behavior."""
def __init__(self, random_state: int = 42):
self.random_state = random_state
self.model = RandomForestClassifier(
n_estimators=100,
max_depth=6,
random_state=self.random_state,
)
self._is_fitted = False
self.reference_metrics: dict = {}
self.production_metrics: dict = {}
# Training helpers
def _create_target(self, df: pd.DataFrame) -> pd.Series:
"""
Synthesise a binary target from the feature set so we can train
a classifier without requiring real labels.
"""
np.random.seed(self.random_state)
numeric = df.select_dtypes(include=[np.number])
z = numeric.apply(lambda c: (c - c.mean()) / (c.std() + 1e-8))
score = z.mean(axis=1) + np.random.normal(0, 0.3, len(df))
return (score > score.median()).astype(int)
def fit(self, reference_df: pd.DataFrame) -> "ModelMonitor":
"""Train the surrogate model on the reference dataset."""
X = reference_df.select_dtypes(include=[np.number])
y = self._create_target(reference_df)
X_train, X_val, y_train, y_val = train_test_split(
X, y, test_size=0.2, random_state=self.random_state
)
self.model.fit(X_train, y_train)
self._is_fitted = True
val_preds = self.model.predict(X_val)
val_proba = self.model.predict_proba(X_val)[:, 1]
self.reference_metrics = self._evaluate(y_val, val_preds, val_proba)
return self
# Tracking methods
def track_accuracy(self, production_df: pd.DataFrame) -> dict:
"""Return accuracy on production data vs. reference baseline."""
self._check_fitted()
X = production_df.select_dtypes(include=[np.number])
y = self._create_target(production_df)
preds = self.model.predict(X)
prod_acc = accuracy_score(y, preds)
ref_acc = self.reference_metrics.get("accuracy", 0)
return {
"reference_accuracy": round(ref_acc, 4),
"production_accuracy": round(prod_acc, 4),
"accuracy_drop": round(ref_acc - prod_acc, 4),
"concept_drift_detected": (ref_acc - prod_acc) > 0.05,
}
def track_predictions(self, production_df: pd.DataFrame) -> dict:
"""Prediction distribution stats on production data."""
self._check_fitted()
X = production_df.select_dtypes(include=[np.number])
proba = self.model.predict_proba(X)[:, 1]
preds = self.model.predict(X)
return {
"mean_probability": round(float(proba.mean()), 4),
"std_probability": round(float(proba.std()), 4),
"positive_rate": round(float(preds.mean()), 4),
"prediction_count": int(len(preds)),
}
def generate_report(self, production_df: pd.DataFrame) -> dict:
"""Full comparison report: reference vs. production."""
self._check_fitted()
X = production_df.select_dtypes(include=[np.number])
y = self._create_target(production_df)
preds = self.model.predict(X)
proba = self.model.predict_proba(X)[:, 1]
self.production_metrics = self._evaluate(y, preds, proba)
degradation = {
metric: round(
self.reference_metrics.get(metric, 0)
- self.production_metrics.get(metric, 0),
4,
)
for metric in self.reference_metrics
}
return {
"reference_metrics": self.reference_metrics,
"production_metrics": self.production_metrics,
"metric_degradation": degradation,
"concept_drift_detected": any(v > 0.05 for v in degradation.values()),
"feature_importances": dict(
zip(
X.columns.tolist(),
[round(fi, 4) for fi in self.model.feature_importances_],
)
),
}
# Internals
@staticmethod
def _evaluate(
y_true: np.ndarray,
y_pred: np.ndarray,
y_proba: np.ndarray,
) -> dict:
return {
"accuracy": round(accuracy_score(y_true, y_pred), 4),
"precision": round(precision_score(y_true, y_pred, zero_division=0), 4),
"recall": round(recall_score(y_true, y_pred, zero_division=0), 4),
"f1": round(f1_score(y_true, y_pred, zero_division=0), 4),
"auc_roc": round(roc_auc_score(y_true, y_proba), 4),
}
def _check_fitted(self):
if not self._is_fitted:
raise RuntimeError("Call .fit(reference_df) before monitoring.")
|