Fairness is not a one-time achievement—it requires continuous monitoring and regular audits. This chapter covers systems and practices for maintaining fair AI in production, following 弘益人間 principles.
AI systems can drift over time as data distributions change, user populations evolve, and societal norms shift. Without ongoing monitoring, a system that was fair at deployment can become biased over time.
Track fairness metrics continuously alongside traditional performance metrics.
# Fairness Monitoring Dashboard
import numpy as np
from datetime import datetime, timedelta
class FairnessMonitor:
"""
Real-time fairness monitoring system
"""
def __init__(self, protected_attributes, fairness_thresholds):
self.protected_attributes = protected_attributes
self.fairness_thresholds = fairness_thresholds
self.metrics_history = []
self.alerts = []
def log_prediction(self, features, prediction, true_label=None, timestamp=None):
"""
Log individual prediction for monitoring
"""
if timestamp is None:
timestamp = datetime.now()
record = {
'timestamp': timestamp,
'features': features,
'prediction': prediction,
'true_label': true_label,
'protected_attrs': {
attr: features.get(attr) for attr in self.protected_attributes
}
}
self.metrics_history.append(record)
def compute_realtime_metrics(self, window_hours=24):
"""
Compute fairness metrics for recent predictions
"""
cutoff_time = datetime.now() - timedelta(hours=window_hours)
recent_records = [r for r in self.metrics_history
if r['timestamp'] > cutoff_time]
if not recent_records:
return {}
# Extract data
predictions = np.array([r['prediction'] for r in recent_records])
metrics = {}
for attr in self.protected_attributes:
groups = {}
for record in recent_records:
group = record['protected_attrs'][attr]
if group not in groups:
groups[group] = []
groups[group].append(record['prediction'])
# Calculate demographic parity
positive_rates = {}
for group, preds in groups.items():
positive_rates[group] = np.mean(preds)
if len(positive_rates) > 1:
max_rate = max(positive_rates.values())
min_rate = min(positive_rates.values())
disparity = max_rate - min_rate
metrics[attr] = {
'positive_rates': positive_rates,
'disparity': disparity,
'status': 'OK' if disparity < self.fairness_thresholds.get(attr, 0.1) else 'ALERT'
}
if disparity >= self.fairness_thresholds.get(attr, 0.1):
self.raise_alert(attr, disparity, positive_rates)
return metrics
def raise_alert(self, attribute, disparity, rates):
"""
Raise fairness alert
"""
alert = {
'timestamp': datetime.now(),
'attribute': attribute,
'disparity': disparity,
'rates': rates,
'severity': 'HIGH' if disparity > 0.2 else 'MEDIUM'
}
self.alerts.append(alert)
print(f"⚠️ FAIRNESS ALERT: {attribute} disparity = {disparity:.3f}")
def generate_report(self):
"""
Generate monitoring report
"""
print("FAIRNESS MONITORING REPORT")
print("=" * 70)
print(f"Report Time: {datetime.now()}")
print(f"Total Predictions Logged: {len(self.metrics_history)}")
print(f"Active Alerts: {len([a for a in self.alerts if (datetime.now() - a['timestamp']).hours < 24])}")
current_metrics = self.compute_realtime_metrics()
print("\nCURRENT FAIRNESS STATUS:")
print("-" * 70)
for attr, data in current_metrics.items():
print(f"\n{attr}:")
print(f" Disparity: {data['disparity']:.3f} [{data['status']}]")
for group, rate in data['positive_rates'].items():
print(f" {group}: {rate:.3f}")
print("\n弘益人間 - Maintaining fairness for all users")
print("=" * 70)
# Usage
monitor = FairnessMonitor(
protected_attributes=['gender', 'race'],
fairness_thresholds={'gender': 0.1, 'race': 0.1}
)
Detect when data distributions or model behavior changes significantly.
# Distribution Drift Detection
from scipy import stats
class DriftDetector:
"""
Detect distribution drift in features and predictions
"""
def __init__(self, reference_data, significance_level=0.05):
self.reference_data = reference_data
self.significance_level = significance_level
def detect_feature_drift(self, current_data, feature_name):
"""
Use Kolmogorov-Smirnov test to detect feature drift
"""
reference_values = self.reference_data[feature_name]
current_values = current_data[feature_name]
# KS test
statistic, p_value = stats.ks_2samp(reference_values, current_values)
drift_detected = p_value < self.significance_level
result = {
'feature': feature_name,
'statistic': statistic,
'p_value': p_value,
'drift_detected': drift_detected
}
if drift_detected:
print(f"⚠️ DRIFT DETECTED in {feature_name}")
print(f" KS Statistic: {statistic:.4f}, p-value: {p_value:.4f}")
return result
def detect_prediction_drift(self, current_predictions):
"""
Detect drift in prediction distribution
"""
reference_predictions = self.reference_data['predictions']
# Compare distributions
ref_mean = reference_predictions.mean()
curr_mean = current_predictions.mean()
ref_std = reference_predictions.std()
curr_std = current_predictions.std()
# Z-test for mean difference
z_score = (curr_mean - ref_mean) / (ref_std / np.sqrt(len(current_predictions)))
p_value = 2 * (1 - stats.norm.cdf(abs(z_score)))
drift_detected = p_value < self.significance_level
print(f"Prediction Drift Analysis:")
print(f" Reference mean: {ref_mean:.3f}")
print(f" Current mean: {curr_mean:.3f}")
print(f" Z-score: {z_score:.3f}")
print(f" Drift detected: {drift_detected}")
return drift_detected
def comprehensive_drift_check(self, current_data, protected_attrs):
"""
Check for drift across all features and groups
"""
print("\nCOMPREHENSIVE DRIFT CHECK")
print("=" * 70)
drift_report = {'features': {}, 'groups': {}}
# Check each feature
for feature in current_data.columns:
if feature in protected_attrs:
continue
result = self.detect_feature_drift(current_data, feature)
drift_report['features'][feature] = result
# Check group-specific drift
for attr in protected_attrs:
for group in current_data[attr].unique():
mask = current_data[attr] == group
group_predictions = current_data[mask]['predictions']
ref_mask = self.reference_data[attr] == group
ref_group_predictions = self.reference_data[ref_mask]['predictions']
# Compare group distributions
statistic, p_value = stats.ks_2samp(ref_group_predictions, group_predictions)
drift_report['groups'][f"{attr}_{group}"] = {
'statistic': statistic,
'p_value': p_value,
'drift_detected': p_value < self.significance_level
}
return drift_report
Run comprehensive fairness audits on a regular schedule.
# Automated Audit Scheduler
import schedule
import time
from datetime import datetime
class AutomatedAuditor:
"""
Automated fairness auditing system
"""
def __init__(self, model, data_source, protected_attrs):
self.model = model
self.data_source = data_source
self.protected_attrs = protected_attrs
self.audit_history = []
def run_audit(self):
"""
Execute comprehensive fairness audit
"""
print(f"\n{'='*70}")
print(f"AUTOMATED FAIRNESS AUDIT - {datetime.now()}")
print(f"{'='*70}")
# Fetch recent data
data = self.data_source.get_recent_data()
X = data[self.model.feature_names]
y_true = data['label']
# Generate predictions
y_pred = self.model.predict(X)
audit_results = {
'timestamp': datetime.now(),
'sample_size': len(data),
'overall_accuracy': np.mean(y_true == y_pred),
'group_metrics': {},
'alerts': []
}
# Evaluate fairness for each protected attribute
for attr in self.protected_attrs:
groups = data[attr].unique()
group_metrics = {}
for group in groups:
mask = data[attr] == group
accuracy = np.mean(y_true[mask] == y_pred[mask])
positive_rate = np.mean(y_pred[mask])
group_metrics[group] = {
'accuracy': accuracy,
'positive_rate': positive_rate,
'sample_size': mask.sum()
}
# Check for disparities
accuracies = [m['accuracy'] for m in group_metrics.values()]
acc_disparity = max(accuracies) - min(accuracies)
rates = [m['positive_rate'] for m in group_metrics.values()]
rate_disparity = max(rates) - min(rates)
audit_results['group_metrics'][attr] = {
'metrics': group_metrics,
'accuracy_disparity': acc_disparity,
'rate_disparity': rate_disparity
}
# Generate alerts
if acc_disparity > 0.1:
audit_results['alerts'].append({
'type': 'accuracy_disparity',
'attribute': attr,
'severity': 'HIGH',
'value': acc_disparity
})
if rate_disparity > 0.1:
audit_results['alerts'].append({
'type': 'demographic_parity',
'attribute': attr,
'severity': 'MEDIUM',
'value': rate_disparity
})
# Store audit results
self.audit_history.append(audit_results)
# Print summary
self.print_audit_summary(audit_results)
# Send alerts if necessary
if audit_results['alerts']:
self.send_alerts(audit_results['alerts'])
return audit_results
def print_audit_summary(self, results):
"""
Print audit results summary
"""
print(f"\nAudit Summary:")
print(f" Sample Size: {results['sample_size']}")
print(f" Overall Accuracy: {results['overall_accuracy']:.3f}")
print(f" Alerts Generated: {len(results['alerts'])}")
for attr, data in results['group_metrics'].items():
print(f"\n{attr}:")
print(f" Accuracy Disparity: {data['accuracy_disparity']:.3f}")
print(f" Rate Disparity: {data['rate_disparity']:.3f}")
print("\n弘益人間 - Regular audits ensure continued fairness")
def send_alerts(self, alerts):
"""
Send alerts to stakeholders
"""
for alert in alerts:
print(f"\n🚨 ALERT: {alert['type']} - {alert['severity']}")
print(f" Attribute: {alert['attribute']}")
print(f" Value: {alert['value']:.3f}")
def schedule_audits(self, frequency='daily'):
"""
Schedule regular audits
"""
if frequency == 'daily':
schedule.every().day.at("02:00").do(self.run_audit)
elif frequency == 'weekly':
schedule.every().monday.at("02:00").do(self.run_audit)
elif frequency == 'hourly':
schedule.every().hour.do(self.run_audit)
print(f"Automated audits scheduled: {frequency}")
# Run scheduler
while True:
schedule.run_pending()
time.sleep(60)
# Usage
# auditor = AutomatedAuditor(model, data_source, ['gender', 'race'])
# auditor.schedule_audits('daily')
# Performance Degradation Monitor
class PerformanceDegradationDetector:
"""
Detect gradual performance degradation over time
"""
def __init__(self, baseline_metrics, degradation_threshold=0.05):
self.baseline_metrics = baseline_metrics
self.degradation_threshold = degradation_threshold
self.metric_history = []
def check_degradation(self, current_metrics):
"""
Compare current metrics against baseline
"""
degradations = {}
for metric_name, baseline_value in self.baseline_metrics.items():
current_value = current_metrics.get(metric_name)
if current_value is None:
continue
degradation = baseline_value - current_value
degradation_pct = (degradation / baseline_value) * 100
degradations[metric_name] = {
'baseline': baseline_value,
'current': current_value,
'degradation': degradation,
'degradation_pct': degradation_pct,
'threshold_exceeded': degradation > self.degradation_threshold
}
# Log metrics
self.metric_history.append({
'timestamp': datetime.now(),
'metrics': current_metrics,
'degradations': degradations
})
# Report significant degradations
significant_degradations = {
k: v for k, v in degradations.items()
if v['threshold_exceeded']
}
if significant_degradations:
print("⚠️ PERFORMANCE DEGRADATION DETECTED")
for metric, data in significant_degradations.items():
print(f" {metric}: {data['baseline']:.3f} → {data['current']:.3f} "
f"({data['degradation_pct']:+.1f}%)")
return degradations
def analyze_trends(self, window_size=30):
"""
Analyze performance trends over time
"""
if len(self.metric_history) < window_size:
print("Insufficient history for trend analysis")
return
recent_history = self.metric_history[-window_size:]
print("\nPERFORMANCE TREND ANALYSIS")
print("=" * 70)
for metric_name in self.baseline_metrics.keys():
values = [h['metrics'].get(metric_name, 0) for h in recent_history]
# Linear regression to detect trends
x = np.arange(len(values))
slope, intercept = np.polyfit(x, values, 1)
trend = "improving" if slope > 0 else "degrading"
trend_strength = abs(slope) * window_size
print(f"\n{metric_name}:")
print(f" Trend: {trend}")
print(f" Slope: {slope:.6f}")
print(f" Projected change over {window_size} periods: {trend_strength:.3f}")
if slope < -0.001:
print(f" ⚠️ Degrading trend detected")
return True
# Stakeholder Report Generator
class StakeholderReportGenerator:
"""
Generate comprehensive fairness reports for stakeholders
"""
def __init__(self, monitor, auditor):
self.monitor = monitor
self.auditor = auditor
def generate_executive_summary(self):
"""
Create executive summary of fairness status
"""
print("\n" + "=" * 70)
print("EXECUTIVE SUMMARY - AI FAIRNESS STATUS")
print("=" * 70)
print(f"Report Date: {datetime.now().strftime('%Y-%m-%d')}")
# Get latest audit
if self.auditor.audit_history:
latest_audit = self.auditor.audit_history[-1]
print(f"\nOverall Status: ", end="")
if not latest_audit['alerts']:
print("✓ COMPLIANT")
else:
high_alerts = [a for a in latest_audit['alerts'] if a['severity'] == 'HIGH']
if high_alerts:
print("✗ CRITICAL ISSUES")
else:
print("⚠ ATTENTION NEEDED")
print(f"\nKey Metrics:")
print(f" Total Predictions: {latest_audit['sample_size']:,}")
print(f" Overall Accuracy: {latest_audit['overall_accuracy']:.1%}")
print(f" Active Alerts: {len(latest_audit['alerts'])}")
print(f"\nFairness Assessment by Protected Attribute:")
for attr, data in latest_audit['group_metrics'].items():
print(f"\n{attr}:")
print(f" Max Accuracy Disparity: {data['accuracy_disparity']:.1%}")
print(f" Max Rate Disparity: {data['rate_disparity']:.1%}")
status = "✓" if (data['accuracy_disparity'] < 0.1 and
data['rate_disparity'] < 0.1) else "⚠"
print(f" Status: {status}")
print("\n" + "=" * 70)
print("弘益人間 - Commitment to serving all users fairly")
print("=" * 70)
def generate_detailed_report(self, output_file):
"""
Generate detailed report with visualizations
"""
# In production, this would create charts, tables, etc.
with open(output_file, 'w') as f:
f.write("DETAILED FAIRNESS AUDIT REPORT\n")
f.write("=" * 70 + "\n\n")
# Include all audit data
for audit in self.auditor.audit_history[-10:]: # Last 10 audits
f.write(f"\nAudit Date: {audit['timestamp']}\n")
f.write(f"Accuracy: {audit['overall_accuracy']:.3f}\n")
f.write(f"Alerts: {len(audit['alerts'])}\n")
f.write("-" * 70 + "\n")
print(f"Detailed report saved to: {output_file}")
# Usage
# report_gen = StakeholderReportGenerator(monitor, auditor)
# report_gen.generate_executive_summary()