Python案例如何用Scikit-learn做互信息聚类

wen python案例 2026-07-16 178

本文目录导读：

互信息特征选择 + KMeans聚类
使用互信息选择最佳特征
在聚类中使用互信息选择的特征
完整的互信息聚类流程
特征数量和聚类效果分析
实际案例：文本聚类
关键要点

我来介绍如何使用Scikit-learn进行互信息聚类，互信息通常用于特征选择,但也可以配合聚类算法使用。

互信息特征选择 + KMeans聚类

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.datasets import make_classification
from sklearn.feature_selection import SelectKBest, mutual_info_classif
from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import silhouette_score, adjusted_rand_score
# 创建示例数据
X, y = make_classification(n_samples=300, n_features=20, n_informative=5, 
                          n_redundant=5, n_repeated=0, n_classes=3,
                          random_state=42)
print(f"原始数据维度: {X.shape}")
print(f"原始特征数量: {X.shape[1]}")

使用互信息选择最佳特征

# 计算每个特征与目标变量的互信息
mi_scores = mutual_info_classif(X, y, random_state=42)
# 展示互信息得分
feature_mi = pd.DataFrame({
    'feature': [f'特征_{i}' for i in range(X.shape[1])],
    'mi_score': mi_scores
}).sort_values('mi_score', ascending=False)
print("特征互信息得分（前10个）:")
print(feature_mi.head(10))
# 选择最相关的5个特征
selector = SelectKBest(mutual_info_classif, k=5)
X_selected = selector.fit_transform(X, y)
selected_indices = selector.get_support(indices=True)
print(f"\n选择的特征索引: {selected_indices}")

在聚类中使用互信息选择的特征

# 标准化数据
scaler = StandardScaler()
X_selected_scaled = scaler.fit_transform(X_selected)
# 使用KMeans聚类
kmeans = KMeans(n_clusters=3, random_state=42)
cluster_labels = kmeans.fit_predict(X_selected_scaled)
# 评估聚类效果
silhouette_avg = silhouette_score(X_selected_scaled, cluster_labels)
ari_score = adjusted_rand_score(y, cluster_labels)
print(f"轮廓系数: {silhouette_avg:.3f}")
print(f"调整兰德指数: {ari_score:.3f}")
# 可视化聚类结果（使用前两个特征）
plt.figure(figsize=(10, 4))
plt.subplot(121)
plt.scatter(X_selected[:, 0], X_selected[:, 1], c=cluster_labels, cmap='viridis')f'聚类结果 (K={3})')
plt.xlabel(f'特征 {selected_indices[0]}')
plt.ylabel(f'特征 {selected_indices[1]}')
plt.subplot(122)
plt.scatter(X_selected[:, 0], X_selected[:, 1], c=y, cmap='viridis')'真实标签')
plt.xlabel(f'特征 {selected_indices[0]}')
plt.ylabel(f'特征 {selected_indices[1]}')
plt.tight_layout()
plt.show()

完整的互信息聚类流程

from sklearn.feature_selection import mutual_info_regression
from sklearn.cluster import AgglomerativeClustering, DBSCAN
from sklearn.metrics.cluster import mutual_info_score as cluster_mi_score
from sklearn.metrics import normalized_mutual_info_score
class MutualInformationClustering:
    """基于互信息的聚类分析类"""
    def __init__(self, n_features=10):
        self.n_features = n_features
        self.selector = None
        self.cluster_model = None
    def fit_select_features(self, X, y=None):
        """基于互信息选择特征"""
        if y is not None:
            # 有监督：使用互信息分类
            self.selector = SelectKBest(mutual_info_classif, k=self.n_features)
        else:
            # 无监督：使用互信息回归（处理连续变量）
            self.selector = SelectKBest(mutual_info_regression, k=self.n_features)
        X_selected = self.selector.fit_transform(X, y if y is not None else np.zeros(X.shape[0]))
        return X_selected
    def cluster_with_mutual_info(self, X, method='kmeans', n_clusters=3):
        """执行聚类分析"""
        if method == 'kmeans':
            self.cluster_model = KMeans(n_clusters=n_clusters, random_state=42)
        elif method == 'hierarchical':
            self.cluster_model = AgglomerativeClustering(n_clusters=n_clusters)
        elif method == 'dbscan':
            self.cluster_model = DBSCAN(eps=0.5, min_samples=5)
        else:
            raise ValueError("不支持的聚类方法")
        cluster_labels = self.cluster_model.fit_predict(X)
        return cluster_labels
    def evaluate_clustering(self, X, cluster_labels, true_labels=None):
        """评估聚类效果"""
        metrics = {}
        # 轮廓系数
        if len(np.unique(cluster_labels)) > 1:
            metrics['silhouette'] = silhouette_score(X, cluster_labels)
        # 如果有真实标签，计算调整互信息
        if true_labels is not None:
            metrics['nmi'] = normalized_mutual_info_score(true_labels, cluster_labels)
            metrics['ari'] = adjusted_rand_score(true_labels, cluster_labels)
        return metrics
# 使用示例
mi_cluster = MutualInformationClustering(n_features=8)
# 选择特征
X_sel = mi_cluster.fit_select_features(X, y)
# 执行聚类
labels = mi_cluster.cluster_with_mutual_info(X_sel, method='kmeans')
# 评估
metrics = mi_cluster.evaluate_clustering(X_sel, labels, y)
print("聚类评估指标:")
for metric, value in metrics.items():
    print(f"  {metric}: {value:.3f}")

特征数量和聚类效果分析

def analyze_feature_impact(X, y, n_features_range):
    """分析特征数量对聚类效果的影响"""
    results = []
    for n_feat in n_features_range:
        # 特征选择
        selector = SelectKBest(mutual_info_classif, k=min(n_feat, X.shape[1]))
        X_selected = selector.fit_transform(X, y)
        # 聚类
        scaler = StandardScaler()
        X_scaled = scaler.fit_transform(X_selected)
        kmeans = KMeans(n_clusters=3, random_state=42, n_init=10)
        labels = kmeans.fit_predict(X_scaled)
        # 评估
        silhouette = silhouette_score(X_scaled, labels) if len(np.unique(labels)) > 1 else -1
        nmi = normalized_mutual_info_score(y, labels)
        results.append({
            'n_features': n_feat,
            'silhouette': silhouette,
            'nmi': nmi
        })
    return pd.DataFrame(results)
# 分析不同特征数量的影响
n_features_range = range(2, min(15, X.shape[1]), 1)
impact_df = analyze_feature_impact(X, y, n_features_range)
# 可视化
plt.figure(figsize=(12, 4))
plt.subplot(121)
plt.plot(impact_df['n_features'], impact_df['silhouette'], 'o-', color='blue')
plt.xlabel('特征数量')
plt.ylabel('轮廓系数')'特征数量 vs 聚类质量（轮廓系数）')
plt.grid(True, alpha=0.3)
plt.subplot(122)
plt.plot(impact_df['n_features'], impact_df['nmi'], 'o-', color='red')
plt.xlabel('特征数量')
plt.ylabel('NMI分数')'特征数量 vs 聚类质量（归一化互信息）')
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
print("不同特征数量的聚类效果:")
print(impact_df.to_string(index=False))

实际案例：文本聚类

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.datasets import fetch_20newsgroups
# 加载数据
categories = ['sci.space', 'rec.sport.baseball']
newsgroups = fetch_20newsgroups(subset='train', categories=categories, 
                               shuffle=True, random_state=42)
print(f"文本数量: {len(newsgroups.data)}")
print(f"类别: {newsgroups.target_names}")
# TF-IDF向量化
vectorizer = TfidfVectorizer(max_features=1000, stop_words='english')
X_tfidf = vectorizer.fit_transform(newsgroups.data)
# 互信息特征选择
selector = SelectKBest(mutual_info_classif, k=100)
X_selected = selector.fit_transform(X_tfidf.toarray(), newsgroups.target)
# 聚类
scaler = StandardScaler(with_mean=False)  # 稀疏矩阵不需要中心化
X_scaled = scaler.fit_transform(X_selected)
kmeans = KMeans(n_clusters=2, random_state=42, n_init=10)
cluster_labels = kmeans.fit_predict(X_scaled)
# 评估
nmi = normalized_mutual_info_score(newsgroups.target, cluster_labels)
ari = adjusted_rand_score(newsgroups.target, cluster_labels)
print(f"\n文本聚类效果:")
print(f"NMI: {nmi:.3f}")
print(f"ARI: {ari:.3f}")
# 展示每个簇的主要词汇
feature_names = vectorizer.get_feature_names_out()
selected_feature_indices = selector.get_support(indices=True)
selected_features = feature_names[selected_feature_indices]
for cluster_id in range(2):
    cluster_mask = (cluster_labels == cluster_id)
    cluster_size = np.sum(cluster_mask)
    print(f"\n簇 {cluster_id}: {cluster_size} 个文本")