Create cluster_utils.py

Contains choose_k(), compute_cluster_stats(), identify_extreme_features()

cluster_utils.py

@@ -0,0 +1,65 @@
+from sklearn.metrics import silhouette_score
+from sklearn.cluster import KMeans
+
+#========== Before Final Clustering ========== 
+def choose_k(X_pca, k_range=(2, 12)):
+    best_k = 2
+    best_score = -1
+    
+    for k in range(k_range[0], k_range[1]):
+        km = KMeans(n_clusters=k, random_state=42)
+        labels = km.fit_predict(X_pca)
+        score = silhouette_score(X_pca, labels)
+        
+        if score > best_score:
+            best_score = score
+            best_k = k
+            
+    return best_k
+
+#========== During Cluster Analysis ========== 
+# function to save cluster stats
+def compute_cluster_stats(X_processed, labels, feature_names):
+    df = pd.DataFrame(X_processed, columns=feature_names)
+    df['cluster'] = labels
+
+    stats = {}
+
+    for cluster_id in sorted(df['cluster'].unique()):
+        cluster_data = df[df['cluster'] == cluster_id].drop(columns=['cluster'])
+
+        stats[cluster_id] = {
+            "count": len(cluster_data),
+            "mean": cluster_data.mean().to_dict(),
+            "median": cluster_data.median().to_dict(),
+            "std": cluster_data.std().to_dict(),
+            "min": cluster_data.min().to_dict(),
+            "max": cluster_data.max().to_dict(),
+            "range": (cluster_data.max() - cluster_data.min()).to_dict(),
+        }
+
+    return stats
+
+# function to identify core features of the clusters
+def identify_extreme_features(X_processed, labels, feature_names, threshold=1.0):
+    df = pd.DataFrame(X_processed, columns=feature_names)
+    df['cluster'] = labels
+
+    global_mean = df.drop(columns=['cluster']).mean()
+    global_std = df.drop(columns=['cluster']).std()
+
+    extremes = {}
+
+    for cluster_id in sorted(df['cluster'].unique()):
+        cluster_mean = df[df['cluster'] == cluster_id].drop(columns=['cluster']).mean()
+        
+        z_scores = ((cluster_mean - global_mean) / global_std).abs()
+
+        extreme_features = z_scores[z_scores > threshold].sort_values(ascending=False)
+
+        extremes[cluster_id] = {
+            "features": extreme_features.index.tolist(),
+            "z_scores": extreme_features.to_dict()
+        }
+
+    return extremes