Phase 5: AI Impact Assessment & Governance · 40 min · Python · scikit-learn · matplotlib
The Problem
A model that performs well at deployment can degrade over time. The world changes, data distributions shift, and bias can emerge in subgroups that weren't in the training data. Without monitoring, you won't know until a SEV-1 incident occurs. With monitoring, you detect drift early and respond before harm is done.
The 3 types of drift
| Drift Type | What Changes | Detection Method | Example |
|---|
| Data drift | Input feature distributions change | KS test, Population Stability Index (PSI) | Income distribution shifts due to economic change |
| Concept drift | The relationship between features and target changes | Model performance drop on recent data | After COVID, employment_years became less predictive of loan repayment |
| Bias drift | Fairness metrics degrade over time | Periodic bias audit on recent predictions | Disparate impact ratio drops from 0.80 to 0.65 over 3 months |
Alerting thresholds
| Metric | Green | Yellow (Investigate) | Red (Act) |
|---|
| Accuracy | Within 2% of baseline | 2-5% drop | >5% drop |
| Disparate Impact Ratio | ≥ 0.80 | 0.70-0.80 | < 0.70 |
| Equalized Odds Difference | < 0.05 | 0.05-0.10 | > 0.10 |
| PSI (Population Stability Index) | < 0.10 | 0.10-0.25 | > 0.25 |
Building the monitoring module
"""
monitoring_dashboard.py
Post-deployment monitoring for drift detection and alerting.
"""
import json
import numpy as np
import pandas as pd
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
from scipy import stats
from datetime import datetime, timedelta
def population_stability_index(expected, actual, bins=10):
"""
Calculate Population Stability Index (PSI).
PSI < 0.10: stable, 0.10-0.25: investigate, > 0.25: significant shift.
"""
expected_pct = np.histogram(expected, bins=bins, density=True)[0]
actual_pct = np.histogram(actual, bins=bins, density=True)[0]
# Avoid division by zero
expected_pct = np.where(expected_pct == 0, 0.0001, expected_pct)
actual_pct = np.where(actual_pct == 0, 0.0001, actual_pct)
psi = np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct))
return round(float(psi), 4)
def ks_test_drift(baseline, current):
"""Kolmogorov-Smirnov test for distribution drift."""
ks_stat, p_value = stats.ks_2samp(baseline, current)
return {
"ks_statistic": round(float(ks_stat), 4),
"p_value": round(float(p_value), 6),
"drift_detected": p_value < 0.05,
}
def check_bias_drift(baseline_di_ratio, current_di_ratio, threshold=0.80):
"""Check if disparate impact ratio has drifted below threshold."""
return {
"baseline_di": baseline_di_ratio,
"current_di": current_di_ratio,
"drift": current_di_ratio < threshold,
"severity": "red" if current_di_ratio < 0.70 else "yellow" if current_di_ratio < 0.80 else "green",
}
def check_performance_drift(baseline_accuracy, current_accuracy):
"""Check if model accuracy has drifted."""
drop = baseline_accuracy - current_accuracy
return {
"baseline_accuracy": round(baseline_accuracy, 4),
"current_accuracy": round(current_accuracy, 4),
"drop_pct": round(drop * 100, 2),
"severity": "red" if drop > 0.05 else "yellow" if drop > 0.02 else "green",
}
def run_monitoring_check(baseline_data, current_data, feature_names,
baseline_accuracy, current_accuracy,
baseline_di, current_di,
output_dir="."):
"""
Run a complete monitoring check across all drift types.
Args:
baseline_data: DataFrame with baseline (training) feature distributions
current_data: DataFrame with current (production) feature distributions
feature_names: List of feature column names
baseline_accuracy: Model accuracy at deployment
current_accuracy: Current model accuracy
baseline_di: Disparate impact ratio at deployment
current_di: Current disparate impact ratio
Returns:
Dict with monitoring results and alerts
"""
results = {
"check_date": datetime.now().isoformat(),
"data_drift": {},
"performance_drift": {},
"bias_drift": {},
"alerts": [],
"overall_status": "green",
}
# 1. Data drift — check each feature
for feature in feature_names:
if feature in baseline_data.columns and feature in current_data.columns:
psi = population_stability_index(baseline_data[feature], current_data[feature])
ks = ks_test_drift(baseline_data[feature], current_data[feature])
severity = "red" if psi > 0.25 else "yellow" if psi > 0.10 else "green"
results["data_drift"][feature] = {
"psi": psi,
"ks_statistic": ks["ks_statistic"],
"ks_p_value": ks["p_value"],
"drift_detected": ks["drift_detected"],
"severity": severity,
}
if severity in ["yellow", "red"]:
results["alerts"].append({
"type": "data_drift",
"feature": feature,
"severity": severity,
"message": f"Feature '{feature}' PSI={psi} ({severity})",
})
# 2. Performance drift
perf = check_performance_drift(baseline_accuracy, current_accuracy)
results["performance_drift"] = perf
if perf["severity"] in ["yellow", "red"]:
results["alerts"].append({
"type": "performance_drift",
"severity": perf["severity"],
"message": f"Accuracy dropped {perf['drop_pct']}% ({perf['severity']})",
})
# 3. Bias drift
bias = check_bias_drift(baseline_di, current_di)
results["bias_drift"] = bias
if bias["severity"] in ["yellow", "red"]:
results["alerts"].append({
"type": "bias_drift",
"severity": bias["severity"],
"message": f"DI ratio dropped from {bias['baseline_di']} to {bias['current_di']} ({bias['severity']})",
})
# Overall status
severities = [d["severity"] for d in results["data_drift"].values()]
severities.append(perf["severity"])
severities.append(bias["severity"])
if "red" in severities:
results["overall_status"] = "red"
elif "yellow" in severities:
results["overall_status"] = "yellow"
else:
results["overall_status"] = "green"
return results
def generate_monitoring_dashboard(results, output_file="monitoring_dashboard.png"):
"""Generate a visual monitoring dashboard."""
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
fig.suptitle("AI Model Monitoring Dashboard", fontsize=16, fontweight='bold')
# 1. Data drift PSI by feature
ax = axes[0, 0]
features = list(results["data_drift"].keys())
psi_values = [results["data_drift"][f]["psi"] for f in features]
colors = ['red' if p > 0.25 else 'orange' if p > 0.10 else 'green' for p in psi_values]
ax.barh(features, psi_values, color=colors)
ax.axvline(x=0.10, color='orange', linestyle='--', label='Investigate (0.10)')
ax.axvline(x=0.25, color='red', linestyle='--', label='Significant (0.25)')
ax.set_xlabel('PSI')
ax.set_title('Data Drift (PSI by Feature)')
ax.legend()
# 2. Performance drift
ax = axes[0, 1]
perf = results["performance_drift"]
ax.bar(["Baseline", "Current"], [perf["baseline_accuracy"], perf["current_accuracy"]],
color=['green', perf["severity"] if perf["severity"] != "green" else 'green'])
ax.set_ylabel('Accuracy')
ax.set_title(f'Performance Drift ({perf["drop_pct"]}% drop)')
ax.set_ylim(0, 1)
# 3. Bias drift
ax = axes[1, 0]
bias = results["bias_drift"]
ax.bar(["Baseline", "Current"], [bias["baseline_di"], bias["current_di"]],
color=['green', bias["severity"] if bias["severity"] != "green" else 'green'])
ax.axhline(y=0.80, color='red', linestyle='--', label='80% rule threshold')
ax.set_ylabel('Disparate Impact Ratio')
ax.set_title('Bias Drift (DI Ratio)')
ax.legend()
# 4. Alert summary
ax = axes[1, 1]
ax.axis('off')
alert_text = f"Overall Status: {results['overall_status'].upper()}\n\nAlerts ({len(results['alerts'])}):\n"
for alert in results["alerts"]:
alert_text += f" [{alert['severity'].upper()}] {alert['message']}\n"
if not results["alerts"]:
alert_text += " No alerts — all metrics within thresholds"
ax.text(0.1, 0.5, alert_text, fontsize=12, fontfamily='monospace',
verticalalignment='center', transform=ax.transAxes,
bbox=dict(boxstyle='round', facecolor='wheat' if results["overall_status"] != "green" else 'lightgreen'))
ax.set_title('Alert Summary')
plt.tight_layout()
plt.savefig(output_file, dpi=150, bbox_inches='tight')
plt.close()
return output_file
if __name__ == "__main__":
# Simulate baseline and current data
np.random.seed(42)
n = 2000
# Baseline (training) data
baseline_df = pd.DataFrame({
"income": np.random.normal(55000, 20000, n).clip(15000, 200000),
"credit_score": np.random.normal(700, 80, n).clip(300, 850),
"debt_ratio": np.random.uniform(0.1, 0.5, n),
})
# Current (production) data — with drift
current_df = pd.DataFrame({
"income": np.random.normal(48000, 22000, n).clip(15000, 200000), # Income shifted down
"credit_score": np.random.normal(680, 90, n).clip(300, 850), # Credit scores shifted
"debt_ratio": np.random.uniform(0.15, 0.6, n), # Debt ratios increased
})
print("Running monitoring check...")
results = run_monitoring_check(
baseline_data=baseline_df,
current_data=current_df,
feature_names=["income", "credit_score", "debt_ratio"],
baseline_accuracy=0.87,
current_accuracy=0.82, # 5% drop
baseline_di=0.80,
current_di=0.68, # Below 80% rule
)
with open("monitoring_results.json", "w") as f:
# .item() turns NumPy numbers and booleans into plain Python ones, which json can write.
json.dump(results, f, indent=2, default=lambda value: value.item())
dashboard = generate_monitoring_dashboard(results)
print(f"\nOverall status: {results['overall_status'].upper()}")
print(f"Alerts: {len(results['alerts'])}")
for alert in results["alerts"]:
print(f" [{alert['severity'].upper()}] {alert['message']}")
print(f"\nDashboard saved to {dashboard}")
print("Results saved to monitoring_results.json")
Your monitoring dashboard shows PSI = 0.18 for the "income" feature (yellow: investigate) but the disparate impact ratio is still 0.83 (green: passes the 80% rule) and accuracy has only dropped 1% (green). A teammate says "only one metric is yellow and the other two are green, so overall status should be green — let's not raise an alert." Does run_monitoring_check's logic agree, and should it?
The code takes the worst severity across every data-drift feature, performance drift, and bias drift, and that becomes the overall status — there's no averaging, and no metric is weighted above another. This "worst-of" design is intentional: a single yellow PSI reading is often the earliest signal that something is starting to shift, well before it shows up as an accuracy drop or a bias drift breach. If the system averaged severities (or let green metrics cancel out a yellow one), it would suppress exactly the early warning the monitoring system was built to catch, and the team would only find out once the disparate impact ratio or accuracy had already crossed into yellow or red — by which point the incident response playbook (Lesson 4) is likely already needed instead of a routine investigation.