Your ML Model Is Lying to You — Here's How to Catch It.

import numpy as np
from scipy import stats
import json
from datetime import datetime

class InputDriftDetector:
    def __init__(self, reference_data: np.ndarray, feature_name: str):
        """
        reference_data: your training data distribution (sample)
        feature_name: name of the feature you're monitoring
        """
        self.reference = reference_data
        self.feature_name = feature_name
        self.alerts = []

    def check(self, current_data: np.ndarray, threshold: float = 0.05) -> bool:
        """
        Uses KS test to compare current data distribution
        against training distribution.

        p_value < threshold = distributions are significantly different = drift!
        Think of it like: "are these two groups of numbers
        drawn from the same population?"
        """
        ks_stat, p_value = stats.ks_2samp(self.reference, current_data)

        drift_detected = p_value < threshold

        result = {
            "timestamp": datetime.utcnow().isoformat(),
            "feature": self.feature_name,
            "ks_statistic": round(ks_stat, 4),
            "p_value": round(p_value, 4),
            "drift_detected": drift_detected
        }

        if drift_detected:
            self.alerts.append(result)
            print(f"⚠️  Input drift detected in '{self.feature_name}'!")
            print(f"    KS stat: {ks_stat:.4f} | p-value: {p_value:.4f}")
            print(f"    Your input distribution has shifted from training.")

        return drift_detected

# Example usage
# Simulate training distribution: mostly young users (20-35)
training_ages = np.random.normal(loc=28, scale=5, size=1000)

# Simulate current distribution: older users (40-55)
current_ages = np.random.normal(loc=47, scale=6, size=200)

detector = InputDriftDetector(training_ages, feature_name="user_age")
detector.check(current_ages)
# ⚠️  Input drift detected in 'user_age'!

class PredictionDriftDetector:
    def __init__(self, baseline_predictions: list):
        """
        baseline_predictions: list of predictions from a known-good period
        """
        self.baseline = baseline_predictions
        self.baseline_pos_rate = sum(baseline_predictions) / len(baseline_predictions)

    def check(self, current_predictions: list, tolerance: float = 0.1) -> bool:
        """
        Checks if the positive prediction rate has shifted
        significantly from the baseline.

        Like checking: "are we predicting churn for WAY more
        people than we used to? If so — something changed."
        """
        current_pos_rate = sum(current_predictions) / len(current_predictions)
        shift = abs(current_pos_rate - self.baseline_pos_rate)

        drift_detected = shift > tolerance

        print(f"📊 Prediction distribution check:")
        print(f"   Baseline positive rate: {self.baseline_pos_rate:.2%}")
        print(f"   Current positive rate:  {current_pos_rate:.2%}")
        print(f"   Shift: {shift:.2%}")

        if drift_detected:
            print(f"⚠️  Prediction drift detected! Shift exceeds {tolerance:.0%} tolerance.")
            if current_pos_rate > self.baseline_pos_rate:
                print(f"   Model is predicting positive MORE often than before.")
            else:
                print(f"   Model is predicting positive LESS often than before.")
        else:
            print(f"✅ Prediction distribution looks stable.")

        return drift_detected

# Example
baseline = [1 if x > 0.85 else 0 for x in np.random.random(1000)]  # ~15% positive rate
current  = [1 if x > 0.60 else 0 for x in np.random.random(200)]   # ~40% positive rate

detector = PredictionDriftDetector(baseline)
detector.check(current)

class DataQualityMonitor:
    def __init__(self, 
                 baseline_null_rate: float,
                 baseline_value_range: tuple):
        """
        baseline_null_rate: expected % of null values (from training data)
        baseline_value_range: expected (min, max) of values
        """
        self.baseline_null_rate = baseline_null_rate
        self.value_min, self.value_max = baseline_value_range
        self.issues = []

    def check(self, data: list, feature_name: str) -> dict:
        issues_found = []

        # Check null rate
        null_rate = sum(1 for x in data if x is None) / len(data)
        if null_rate > self.baseline_null_rate * 2:  # alert if 2x baseline
            issues_found.append(
                f"Null rate: {null_rate:.1%} (baseline: {self.baseline_null_rate:.1%})"
            )

        # Check value range
        valid_values = [x for x in data if x is not None]
        if valid_values:
            actual_min = min(valid_values)
            actual_max = max(valid_values)

            if actual_min < self.value_min or actual_max > self.value_max:
                issues_found.append(
                    f"Values out of range: [{actual_min}, {actual_max}] "
                    f"(expected: [{self.value_min}, {self.value_max}])"
                )

        if issues_found:
            print(f"⚠️  Data quality issues in '{feature_name}':")
            for issue in issues_found:
                print(f"   - {issue}")
        else:
            print(f"✅ '{feature_name}' data quality looks good.")

        return {"feature": feature_name, "issues": issues_found}

# Example
monitor = DataQualityMonitor(
    baseline_null_rate=0.02,    # 2% nulls in training data
    baseline_value_range=(0, 100)  # values expected between 0 and 100
)

# Simulate degraded data: more nulls, some out-of-range values
bad_data = [None, None, None, None, None, 45, 23, 150, 67, 89, None, 34]
monitor.check(bad_data, feature_name="purchase_amount")

from dataclasses import dataclass, field
from typing import List, Optional
from datetime import datetime
import numpy as np

@dataclass
class DriftAlert:
    timestamp: str
    drift_type: str
    feature: str
    severity: str  # "low", "medium", "high"
    message: str

class MLDriftMonitor:
    """
    A simple, practical drift monitor for production ML systems.

    Think of this as your smoke detector — runs quietly in the
    background, only makes noise when something actually needs
    your attention.
    """

    def __init__(self, model_name: str):
        self.model_name = model_name
        self.alerts: List[DriftAlert] = []
        self.check_count = 0

    def check_input_drift(self, 
                          feature_name: str,
                          reference: np.ndarray, 
                          current: np.ndarray,
                          threshold: float = 0.05):
        from scipy import stats
        _, p_value = stats.ks_2samp(reference, current)

        if p_value < threshold:
            severity = "high" if p_value < 0.01 else "medium"
            self.alerts.append(DriftAlert(
                timestamp=datetime.utcnow().isoformat(),
                drift_type="input_drift",
                feature=feature_name,
                severity=severity,
                message=f"Input distribution shifted (p={p_value:.4f})"
            ))

    def check_prediction_drift(self,
                               baseline_preds: list,
                               current_preds: list,
                               tolerance: float = 0.1):
        baseline_rate = sum(baseline_preds) / len(baseline_preds)
        current_rate = sum(current_preds) / len(current_preds)
        shift = abs(current_rate - baseline_rate)

        if shift > tolerance:
            severity = "high" if shift > 0.2 else "medium"
            self.alerts.append(DriftAlert(
                timestamp=datetime.utcnow().isoformat(),
                drift_type="prediction_drift",
                feature="model_output",
                severity=severity,
                message=f"Prediction rate shifted by {shift:.1%}"
            ))

    def check_data_quality(self,
                           feature_name: str,
                           data: list,
                           max_null_rate: float = 0.05):
        null_rate = sum(1 for x in data if x is None) / len(data)

        if null_rate > max_null_rate:
            severity = "high" if null_rate > 0.2 else "low"
            self.alerts.append(DriftAlert(
                timestamp=datetime.utcnow().isoformat(),
                drift_type="data_quality",
                feature=feature_name,
                severity=severity,
                message=f"Null rate: {null_rate:.1%} (max allowed: {max_null_rate:.1%})"
            ))

    def report(self):
        print(f"\n{'='*50}")
        print(f"Drift Monitor Report — {self.model_name}")
        print(f"{'='*50}")

        if not self.alerts:
            print("✅ No drift detected. Model looks healthy.")
            return

        high   = [a for a in self.alerts if a.severity == "high"]
        medium = [a for a in self.alerts if a.severity == "medium"]
        low    = [a for a in self.alerts if a.severity == "low"]

        print(f"Total alerts: {len(self.alerts)}")
        print(f"  🔴 High:   {len(high)}")
        print(f"  🟡 Medium: {len(medium)}")
        print(f"  🟢 Low:    {len(low)}")
        print()

        for alert in self.alerts:
            icon = {"high": "🔴", "medium": "🟡", "low": "🟢"}[alert.severity]
            print(f"{icon} [{alert.drift_type}] {alert.feature}")
            print(f"   {alert.message}")
            print(f"   Detected at: {alert.timestamp}")
            print()

# Usage example
monitor = MLDriftMonitor("churn_classifier_v2")

# Simulate some drift
training_data = np.random.normal(28, 5, 1000)
current_data  = np.random.normal(47, 6, 200)
monitor.check_input_drift("user_age", training_data, current_data)

baseline_preds = [1 if x > 0.85 else 0 for x in np.random.random(1000)]
current_preds  = [1 if x > 0.60 else 0 for x in np.random.random(200)]
monitor.check_prediction_drift(baseline_preds, current_preds)

messy_data = [None, None, 45, None, 23, None, None, 67, None, 89]
monitor.check_data_quality("purchase_amount", messy_data)

monitor.report()

import json
from datetime import datetime

def log_prediction(input_data: dict, prediction, confidence: float):
    """
    The simplest possible monitoring setup.
    Just log everything. You'll thank yourself later.
    """
    log_entry = {
        "timestamp": datetime.utcnow().isoformat(),
        "inputs": input_data,
        "prediction": prediction,
        "confidence": round(confidence, 4)
    }

    # Write to a log file (or your logging system of choice)
    with open("predictions.log", "a") as f:
        f.write(json.dumps(log_entry) + "\n")