Update app.py

app.py

@@ -1,6 +1,7 @@
 import time 
 import math
 
+import numpy as np
 import pandas as pd
 import matplotlib.pyplot as plt
 import seaborn as sns
@@ -14,6 +15,31 @@ from sklearn.metrics import classification_report
 LOGS_DATASET_URI = 'pgurazada1/machine-failure-mlops-demo-logs'
 
 
+# Load and cache training data
+
+dataset = fetch_openml(data_id=42890, as_frame=True, parser="auto")
+data_df = dataset.data
+
+target = 'Machine failure'
+numeric_features = [
+    'Air temperature [K]',
+    'Process temperature [K]',
+    'Rotational speed [rpm]',
+    'Torque [Nm]',
+    'Tool wear [min]'
+]
+
+categorical_features = ['Type']
+
+X = data_df[numeric_features + categorical_features]
+y = data_df[target]
+
+Xtrain, Xtest, ytrain, ytest = train_test_split(
+    X, y,
+    test_size=0.2,
+    random_state=42
+)
+
 def get_data():
     """
     Connect to the HuggingFace dataset where the logs are stored.
@@ -24,38 +50,19 @@ def get_data():
 
     return sample_df
 
-def load_training_data():
-    dataset = fetch_openml(data_id=42890, as_frame=True, parser="auto")
-    data_df = dataset.data
     
-    target = 'Machine failure'
-    numeric_features = [
-        'Air temperature [K]',
-        'Process temperature [K]',
-        'Rotational speed [rpm]',
-        'Torque [Nm]',
-        'Tool wear [min]'
-    ]
-    
-    categorical_features = ['Type']
-
-    X = data_df[numeric_features + categorical_features]
-    y = data_df[target]
-
-    Xtrain, Xtest, ytrain, ytest = train_test_split(
-        X, y,
-        test_size=0.2,
-        random_state=42
-    )
-
-    return Xtrain, ytrain
-
 def check_model_drift():
+    """
+    Check proportion of machine failure as compared to
+    its proportion in training data. If the deviation is more than
+    2 standard deviations, flag a model drift.
+    """
     sample_df = get_data()
     p_pos_label_training_data = 0.03475
     training_data_size = 8000
     
     n_0 = sample_df.prediction.value_counts()[0]
+    
     try:
         n_1 = sample_df.prediction.value_counts()[1]
     except Exception as e:
@@ -67,11 +74,68 @@ def check_model_drift():
     p_diff = abs(p_pos_label_training_data - p_pos_label_sample_logs)
     
     if p_diff > 2 * math.sqrt(variance):
-        return "Model Drift Detected! Check logs!"
+        return "Model Drift Detected! Check Logs!"
     else:
         return "No Model Drift!"
 
+
+def psi(actual_proportions, expected_proportions):
+
+    psi_values = (actual_proportions - expected_proportions) * \
+        np.log(actual_proportions / expected_proportions)
+
+    return sum(psi_values)
     
+
+def check_data_drift():
+    """
+    Compare training data features and live features. If the deviation is
+    more than 2 standard deviations, flag data drift. 
+    Numeric features and catagorical features are dealt with separately.
+    """
+    sample_df = get_data()
+    data_drift_status = {}
+    
+    numeric_features = [
+        'Air temperature [K]',
+        'Process temperature [K]',
+        'Rotational speed [rpm]',
+        'Torque [Nm]',
+        'Tool wear [min]'
+    ]
+    
+    categorical_features = ['Type']
+
+    # Numeric features
+    
+    for feature in numeric_features:
+        mean_feature_training_data = Xtrain[feature].mean()
+        std_feature_training_data = Xtrain[feature].std()
+
+        mean_feature_sample_logs = sample_df[feature].mean()
+
+        mean_diff = abs(mean_feature_training_data - mean_feature_sample_logs)
+
+        if mean_diff > 2 * std_feature_training_data:
+            data_drift_status[feature] = "Data Drift Detected! Check Logs!"
+        else:
+            data_drift_status[feature] = "No Data Drift!"
+
+    # Categorical feature Type
+
+    live_proportions = sample_df['Type'].value_counts(normalize=True).values
+    training_proportions = Xtrain['Type'].value_counts(normalize=True).values
+
+    psi_value = psi(live_proportions, training_proportions)
+
+    if psi_value > 0.1:
+        data_drift_status['Type'] = "Data Drift Detected! Check Logs!"
+    else:
+        data_drift_status['Type'] = "No Data Drift!"
+
+    return data_drift_status
+    
+
 with gr.Blocks() as demo:
     gr.Markdown("# Real-time Monitoring Dashboard")
 
@@ -81,4 +145,11 @@ with gr.Blocks() as demo:
         with gr.Column():
             gr.Textbox(check_model_drift, every=5, label="Model Drift Status")
 
+    gr.Markdown("Data drift detection (every 5 seconds)")
+    
+    with gr.Row():
+        with gr.Column():
+            gr.DataFrame(check_data_drift, every=5, label="Data Drift Status")
+
+
 demo.queue().launch()