机器学习——无监督学习：聚类算法深度解析(K-Means、DBSCAN、GMM)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
import numpy as np
from sklearn.metrics import silhouette_score
import matplotlib.pyplot as plt

class KMeans:
    def __init__(self, n_clusters=3, max_iter=100, random_state=42):
        self.n_clusters = n_clusters
        self.max_iter = max_iter
        self.random_state = random_state
        self.centroids = None
        self.labels = None
        
    def initialize_centroids(self, X):
        """随机初始化簇心"""
        np.random.seed(self.random_state)
        indices = np.random.choice(X.shape[0], self.n_clusters, replace=False)
        return X[indices]
    
    def assign_clusters(self, X):
        """E步：分配样本到最近的簇"""
        distances = np.zeros((X.shape[0], self.n_clusters))
        for k in range(self.n_clusters):
            distances[:, k] = np.linalg.norm(X - self.centroids[k], axis=1)
        return np.argmin(distances, axis=1)
    
    def update_centroids(self, X, labels):
        """M步：更新簇心"""
        new_centroids = np.zeros((self.n_clusters, X.shape[1]))
        for k in range(self.n_clusters):
            cluster_points = X[labels == k]
            if len(cluster_points) > 0:
                new_centroids[k] = cluster_points.mean(axis=0)
            else:
                # 如果某簇为空，随机选择一个样本
                new_centroids[k] = X[np.random.choice(X.shape[0])]
        return new_centroids
    
    def compute_wcss(self, X, labels):
        """计算WCSS"""
        wcss = 0
        for k in range(self.n_clusters):
            cluster_points = X[labels == k]
            wcss += np.sum(np.linalg.norm(cluster_points - self.centroids[k], axis=1) ** 2)
        return wcss
    
    def fit(self, X):
        """训练K-Means"""
        self.centroids = self.initialize_centroids(X)
        
        for iteration in range(self.max_iter):
            # E步
            labels = self.assign_clusters(X)
            
            # M步
            new_centroids = self.update_centroids(X, labels)
            
            # 检查收敛
            if np.allclose(self.centroids, new_centroids):
                print(f"K-Means在第{iteration}次迭代收敛")
                break
            
            self.centroids = new_centroids
        
        self.labels = labels
        return self
    
    def predict(self, X):
        """预测新数据的簇标签"""
        distances = np.zeros((X.shape[0], self.n_clusters))
        for k in range(self.n_clusters):
            distances[:, k] = np.linalg.norm(X - self.centroids[k], axis=1)
        return np.argmin(distances, axis=1)


# 示例：使用鸢尾花数据集
from sklearn.datasets import load_iris
from sklearn.preprocessing import StandardScaler

iris = load_iris()
X = iris.data
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)

# 使用肘部法则选择最优K
wcss_values = []
k_range = range(1, 11)
for k in k_range:
    kmeans = KMeans(n_clusters=k, random_state=42)
    kmeans.fit(X_scaled)
    wcss_values.append(kmeans.compute_wcss(X_scaled, kmeans.labels))

plt.figure(figsize=(10, 6))
plt.plot(k_range, wcss_values, 'bo-')
plt.xlabel('簇数 K')
plt.ylabel('WCSS')
plt.title('肘部法则：选择最优K值')
plt.grid(True)
plt.show()

# 训练最优K-Means
kmeans = KMeans(n_clusters=3, random_state=42)
kmeans.fit(X_scaled)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
import numpy as np
from sklearn.neighbors import NearestNeighbors
import matplotlib.pyplot as plt

class DBSCAN:
    def __init__(self, eps=0.5, min_pts=5):
        self.eps = eps
        self.min_pts = min_pts
        self.labels = None
        
    def get_neighbors(self, X, point_idx):
        """获取点的ε-邻域"""
        distances = np.linalg.norm(X - X[point_idx], axis=1)
        neighbors = np.where(distances <= self.eps)[0]
        return neighbors
    
    def expand_cluster(self, X, labels, point_idx, cluster_id, visited):
        """扩展簇"""
        # 获取初始邻域
        neighbors = self.get_neighbors(X, point_idx)
        
        if len(neighbors) < self.min_pts:
            # 核心点条件不满足
            return False
        
        # 标记初始点
        labels[point_idx] = cluster_id
        
        # 广度优先搜索扩展簇
        queue = list(neighbors)
        while queue:
            current_idx = queue.pop(0)
            
            if visited[current_idx]:
                continue
            
            visited[current_idx] = True
            
            # 获取当前点的邻域
            current_neighbors = self.get_neighbors(X, current_idx)
            
            # 如果是核心点，扩展邻域
            if len(current_neighbors) >= self.min_pts:
                for neighbor_idx in current_neighbors:
                    if not visited[neighbor_idx]:
                        queue.append(neighbor_idx)
            
            # 如果未分配，加入当前簇
            if labels[current_idx] == -1:
                labels[current_idx] = cluster_id
        
        return True
    
    def fit(self, X):
        """训练DBSCAN"""
        n_samples = X.shape[0]
        self.labels = np.full(n_samples, -1, dtype=int)  # -1表示噪声点
        visited = np.zeros(n_samples, dtype=bool)
        
        cluster_id = 0
        
        for i in range(n_samples):
            if visited[i]:
                continue
            
            visited[i] = True
            
            if self.expand_cluster(X, self.labels, i, cluster_id, visited):
                cluster_id += 1
        
        return self
    
    def fit_predict(self, X):
        """训练并返回标签"""
        self.fit(X)
        return self.labels
    
    @staticmethod
    def get_eps_by_kdistance(X, min_pts=5, k=None):
        """使用K-distance图选择eps"""
        if k is None:
            k = min_pts
        
        # 计算k-nearest距离
        nbrs = NearestNeighbors(n_neighbors=k+1).fit(X)
        distances, indices = nbrs.kneighbors(X)
        
        # 第k个邻域的距离（排除自身）
        k_distances = distances[:, k]
        k_distances = np.sort(k_distances)
        
        # 绘制K-distance图
        plt.figure(figsize=(10, 6))
        plt.plot(k_distances)
        plt.xlabel('数据点索引（排序）')
        plt.ylabel('第{}个邻近距离'.format(k))
        plt.title('K-distance图：选择eps值')
        plt.grid(True)
        plt.show()
        
        return k_distances


# 示例
from sklearn.datasets import make_moons

# 生成月形数据
X, _ = make_moons(n_samples=300, noise=0.05, random_state=42)

# 获取eps值
k_distances = DBSCAN.get_eps_by_kdistance(X, min_pts=5)

# 训练DBSCAN
dbscan = DBSCAN(eps=0.2, min_pts=5)
labels = dbscan.fit_predict(X)

# 统计结果
n_clusters = len(set(labels)) - (1 if -1 in labels else 0)
n_noise = list(labels).count(-1)

print(f"聚类数: {n_clusters}")
print(f"噪声点数: {n_noise}")

# 可视化
plt.figure(figsize=(10, 6))
unique_labels = set(labels)
colors = plt.cm.Spectral(np.linspace(0, 1, len(unique_labels)))

for label, color in zip(unique_labels, colors):
    if label == -1:
        color = 'k'  # 噪声点用黑色
    
    mask = labels == label
    plt.scatter(X[mask, 0], X[mask, 1], c=[color], label=f'Cluster {label}', s=50)

plt.title('DBSCAN聚类结果')
plt.legend()
plt.show()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
import numpy as np
from scipy.stats import multivariate_normal
from sklearn.metrics import silhouette_score
import matplotlib.pyplot as plt

class GaussianMixtureModel:
    def __init__(self, n_components=3, max_iter=100, tol=1e-4, 
                 covariance_type='full', random_state=42):
        """
        初始化GMM
        
        参数：
        - n_components: 高斯分布数量
        - max_iter: 最大迭代次数
        - tol: 收敛阈值
        - covariance_type: 协方差类型 ('full', 'diag', 'tied', 'spherical')
        - random_state: 随机种子
        """
        self.n_components = n_components
        self.max_iter = max_iter
        self.tol = tol
        self.covariance_type = covariance_type
        self.random_state = random_state
        
        # 模型参数
        self.weights = None
        self.means = None
        self.covariances = None
        self.responsibilities = None
        
    def initialize_parameters(self, X):
        """初始化参数"""
        n_samples, n_features = X.shape
        np.random.seed(self.random_state)
        
        # 随机初始化混合系数
        self.weights = np.ones(self.n_components) / self.n_components
        
        # 从数据中随机选择K个点作为初始均值
        indices = np.random.choice(n_samples, self.n_components, replace=False)
        self.means = X[indices].copy()
        
        # 初始化协方差矩阵
        if self.covariance_type == 'full':
            self.covariances = np.array([np.eye(n_features) for _ in range(self.n_components)])
        elif self.covariance_type == 'diag':
            self.covariances = np.array([np.ones(n_features) for _ in range(self.n_components)])
        elif self.covariance_type == 'tied':
            self.covariances = np.eye(n_features)
        elif self.covariance_type == 'spherical':
            self.covariances = np.ones(self.n_components)
    
    def compute_gaussian_probability(self, X, mean, covariance):
        """计算高斯概率"""
        if self.covariance_type == 'spherical':
            # 球形协方差
            var = covariance
            numerator = np.exp(-(np.sum((X - mean) ** 2, axis=1) / (2 * var)))
            denominator = np.sqrt((2 * np.pi * var) ** X.shape[1])
            return numerator / denominator
        else:
            # 完全协方差
            return multivariate_normal.pdf(X, mean, covariance)
    
    def e_step(self, X):
        """E步：计算责任度"""
        n_samples = X.shape[0]
        self.responsibilities = np.zeros((n_samples, self.n_components))
        
        # 计算每个高斯分布的概率
        for k in range(self.n_components):
            self.responsibilities[:, k] = (
                self.weights[k] * 
                self.compute_gaussian_probability(X, self.means[k], self.covariances[k])
            )
        
        # 归一化为后验概率
        total_prob = np.sum(self.responsibilities, axis=1, keepdims=True)
        self.responsibilities = self.responsibilities / (total_prob + 1e-10)
        
        return -np.sum(np.log(total_prob + 1e-10))
    
    def m_step(self, X):
        """M步：更新参数"""
        n_samples, n_features = X.shape
        
        # 有效样本数
        N_k = np.sum(self.responsibilities, axis=0)
        
        # 更新混合系数
        self.weights = N_k / n_samples
        
        # 更新均值
        self.means = (self.responsibilities.T @ X) / (N_k[:, np.newaxis] + 1e-10)
        
        # 更新协方差矩阵
        if self.covariance_type == 'full':
            self.covariances = np.array([
                ((self.responsibilities[:, k:k+1] * (X - self.means[k])).T @ 
                 (X - self.means[k])) / (N_k[k] + 1e-10)
                for k in range(self.n_components)
            ])
        elif self.covariance_type == 'diag':
            self.covariances = np.array([
                np.mean(self.responsibilities[:, k:k+1] * (X - self.means[k])**2, axis=0)
                for k in range(self.n_components)
            ])
        elif self.covariance_type == 'spherical':
            self.covariances = np.array([
                np.mean(self.responsibilities[:, k] * 
                       np.sum((X - self.means[k])**2, axis=1)) / n_features
                for k in range(self.n_components)
            ])
    
    def fit(self, X):
        """训练GMM"""
        self.initialize_parameters(X)
        prev_log_likelihood = -np.inf
        
        for iteration in range(self.max_iter):
            # E步
            log_likelihood = self.e_step(X)
            
            # 检查收敛
            if abs(log_likelihood - prev_log_likelihood) < self.tol:
                print(f"GMM在第{iteration}次迭代收敛")
                break
            
            # M步
            self.m_step(X)
            
            prev_log_likelihood = log_likelihood
        
        return self
    
    def predict(self, X):
        """预测聚类标签"""
        _, labels = self.predict_proba(X)
        return labels
    
    def predict_proba(self, X):
        """预测概率"""
        responsibilities = np.zeros((X.shape[0], self.n_components))
        
        for k in range(self.n_components):
            responsibilities[:, k] = (
                self.weights[k] * 
                self.compute_gaussian_probability(X, self.means[k], self.covariances[k])
            )
        
        total_prob = np.sum(responsibilities, axis=1, keepdims=True)
        responsibilities = responsibilities / (total_prob + 1e-10)
        labels = np.argmax(responsibilities, axis=1)
        
        return responsibilities, labels
    
    def bic(self, X):
        """计算BIC值"""
        n_samples = X.shape[0]
        
        # 计算对数似然
        log_likelihood = self.e_step(X)
        
        # 参数个数
        if self.covariance_type == 'full':
            n_params = (self.n_components * X.shape[1] +  # 均值
                       self.n_components * X.shape[1] * (X.shape[1] + 1) // 2 +  # 协方差
                       self.n_components - 1)  # 混合系数
        else:
            n_params = (self.n_components * X.shape[1] +  # 均值
                       self.n_components +  # 方差
                       self.n_components - 1)  # 混合系数
        
        bic = -2 * log_likelihood + n_params * np.log(n_samples)
        return bic


# 示例：使用鸢尾花数据
from sklearn.datasets import load_iris
from sklearn.preprocessing import StandardScaler

iris = load_iris()
X = iris.data
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)

# 使用BIC选择最优组件数
bic_values = []
k_range = range(1, 11)
for k in k_range:
    gmm = GaussianMixtureModel(n_components=k, random_state=42)
    gmm.fit(X_scaled)
    bic_values.append(gmm.bic(X_scaled))

plt.figure(figsize=(10, 6))
plt.plot(k_range, bic_values, 'bo-')
plt.xlabel('组件数')
plt.ylabel('BIC值')
plt.title('BIC准则：选择最优组件数')
plt.grid(True)
plt.show()

# 训练最优GMM
best_k = k_range[np.argmin(bic_values)]
gmm = GaussianMixtureModel(n_components=best_k, random_state=42)
gmm.fit(X_scaled)

labels = gmm.predict(X_scaled)
print(f"最优组件数: {best_k}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler
import matplotlib.pyplot as plt
from mpl_toolkits.mplot3d import Axes3D

# 生成模拟数据
np.random.seed(42)
n_samples = 500

# VIP客户：高收入，高消费
vip_customers = np.random.normal([55, 80000, 75, 4, 150], [5, 10000, 10, 1, 30], (150, 5))

# Regular客户：中等收入，中等消费
regular_customers = np.random.normal([40, 50000, 50, 2.5, 100], [8, 15000, 15, 1, 40], (200, 5))

# At-risk客户：低收入，低消费
atrisk_customers = np.random.normal([35, 30000, 25, 1, 50], [10, 10000, 15, 0.5, 20], (150, 5))

# 合并数据
data = np.vstack([vip_customers, regular_customers, atrisk_customers])

# 创建DataFrame
df = pd.DataFrame(data, columns=['Age', 'Annual_Income', 'Spending_Score', 
                                 'Purchase_Frequency', 'Avg_Order_Value'])

# 数据标准化
scaler = StandardScaler()
X = scaler.fit_transform(df)

print("数据形状:", X.shape)
print("数据统计信息:")
print(df.describe())

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
from sklearn.metrics import silhouette_score, davies_bouldin_score, calinski_harabasz_score

# 1. 使用K-Means
from sklearn.cluster import KMeans

kmeans = KMeans(n_clusters=3, random_state=42, n_init=10)
kmeans_labels = kmeans.fit_predict(X)

print("=== K-Means 结果 ===")
print(f"轮廓系数: {silhouette_score(X, kmeans_labels):.4f}")
print(f"Davies-Bouldin指数: {davies_bouldin_score(X, kmeans_labels):.4f}")
print(f"Calinski-Harabasz指数: {calinski_harabasz_score(X, kmeans_labels):.4f}")

# 2. 使用DBSCAN
from sklearn.cluster import DBSCAN
from sklearn.neighbors import NearestNeighbors

# 选择eps
neighbors = NearestNeighbors(n_neighbors=5)
neighbors_fit = neighbors.fit(X)
distances, indices = neighbors_fit.kneighbors(X)
distances = np.sort(distances[:, 4], axis=0)

dbscan = DBSCAN(eps=0.8, min_pts=5)
dbscan_labels = dbscan.fit_predict(X)

print("\n=== DBSCAN 结果 ===")
n_clusters_dbscan = len(set(dbscan_labels)) - (1 if -1 in dbscan_labels else 0)
n_noise = list(dbscan_labels).count(-1)
print(f"聚类数: {n_clusters_dbscan}")
print(f"噪声点数: {n_noise}")

if n_clusters_dbscan > 1 and n_noise < len(dbscan_labels) - 1:
    mask = dbscan_labels != -1
    print(f"轮廓系数: {silhouette_score(X[mask], dbscan_labels[mask]):.4f}")

# 3. 使用GMM
from sklearn.mixture import GaussianMixture

gmm = GaussianMixture(n_components=3, random_state=42)
gmm_labels = gmm.fit_predict(X)

print("\n=== GMM 结果 ===")
print(f"轮廓系数: {silhouette_score(X, gmm_labels):.4f}")
print(f"Davies-Bouldin指数: {davies_bouldin_score(X, gmm_labels):.4f}")
print(f"Calinski-Harabasz指数: {calinski_harabasz_score(X, gmm_labels):.4f}")
print(f"BIC: {gmm.bic(X):.2f}")
print(f"AIC: {gmm.aic(X):.2f}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# 使用K-Means结果进行分析
df['Cluster'] = kmeans_labels
df['Cluster_Name'] = df['Cluster'].map({
    0: 'Cluster_0',
    1: 'Cluster_1',
    2: 'Cluster_2'
})

# 分析每个簇的特征
print("=== 各簇特征分析 ===")
for cluster in range(3):
    cluster_data = df[df['Cluster'] == cluster]
    print(f"\nCluster {cluster} (样本数: {len(cluster_data)}):")
    print(cluster_data[['Age', 'Annual_Income', 'Spending_Score', 
                        'Purchase_Frequency', 'Avg_Order_Value']].describe())
    
    # 计算簇的中心特征
    print(f"平均年龄: {cluster_data['Age'].mean():.1f}")
    print(f"平均年收入: ${cluster_data['Annual_Income'].mean():.0f}")
    print(f"平均消费分数: {cluster_data['Spending_Score'].mean():.1f}")
    print(f"平均购买频率: {cluster_data['Purchase_Frequency'].mean():.2f}次/月")
    print(f"平均订单金额: ${cluster_data['Avg_Order_Value'].mean():.0f}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
# 2D可视化：使用PCA降维
from sklearn.decomposition import PCA

pca = PCA(n_components=2)
X_pca = pca.fit_transform(X)

fig, axes = plt.subplots(1, 3, figsize=(15, 5))

# K-Means可视化
ax = axes[0]
scatter = ax.scatter(X_pca[:, 0], X_pca[:, 1], c=kmeans_labels, cmap='viridis', s=50, alpha=0.6)
ax.scatter(pca.transform(scaler.inverse_transform(kmeans.cluster_centers_))[:, 0],
          pca.transform(scaler.inverse_transform(kmeans.cluster_centers_))[:, 1],
          c='red', marker='X', s=200, edgecolors='black', linewidths=2)
ax.set_title('K-Means聚类结果')
ax.set_xlabel(f'PC1 ({pca.explained_variance_ratio_[0]:.1%})')
ax.set_ylabel(f'PC2 ({pca.explained_variance_ratio_[1]:.1%})')
plt.colorbar(scatter, ax=ax)

# DBSCAN可视化
ax = axes[1]
unique_labels = set(dbscan_labels)
colors = plt.cm.Spectral(np.linspace(0, 1, len(unique_labels)))
for label, color in zip(unique_labels, colors):
    if label == -1:
        color = 'k'
    class_member_mask = (dbscan_labels == label)
    xy = X_pca[class_member_mask]
    ax.scatter(xy[:, 0], xy[:, 1], c=[color], label=f'Cluster {label}', s=50, alpha=0.6)
ax.set_title('DBSCAN聚类结果')
ax.set_xlabel(f'PC1 ({pca.explained_variance_ratio_[0]:.1%})')
ax.set_ylabel(f'PC2 ({pca.explained_variance_ratio_[1]:.1%})')

# GMM可视化
ax = axes[2]
scatter = ax.scatter(X_pca[:, 0], X_pca[:, 1], c=gmm_labels, cmap='viridis', s=50, alpha=0.6)
ax.set_title('GMM聚类结果')
ax.set_xlabel(f'PC1 ({pca.explained_variance_ratio_[0]:.1%})')
ax.set_ylabel(f'PC2 ({pca.explained_variance_ratio_[1]:.1%})')
plt.colorbar(scatter, ax=ax)

plt.tight_layout()
plt.show()

# 3D可视化：Age vs Annual_Income vs Spending_Score
fig = plt.figure(figsize=(12, 5))

ax = fig.add_subplot(121, projection='3d')
scatter = ax.scatter(df['Age'], df['Annual_Income'], df['Spending_Score'], 
                    c=kmeans_labels, cmap='viridis', s=50, alpha=0.6)
ax.set_xlabel('Age')
ax.set_ylabel('Annual Income')
ax.set_zlabel('Spending Score')
ax.set_title('K-Means 3D可视化')
plt.colorbar(scatter, ax=ax)

ax = fig.add_subplot(122, projection='3d')
scatter = ax.scatter(df['Age'], df['Annual_Income'], df['Spending_Score'], 
                    c=gmm_labels, cmap='viridis', s=50, alpha=0.6)
ax.set_xlabel('Age')
ax.set_ylabel('Annual Income')
ax.set_zlabel('Spending Score')
ax.set_title('GMM 3D可视化')
plt.colorbar(scatter, ax=ax)

plt.tight_layout()
plt.show()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
# 根据聚类结果制定营销策略
print("=== 客户分群营销策略 ===\n")

for cluster in range(3):
    cluster_data = df[df['Cluster'] == cluster]
    avg_income = cluster_data['Annual_Income'].mean()
    avg_spending = cluster_data['Spending_Score'].mean()
    avg_frequency = cluster_data['Purchase_Frequency'].mean()
    avg_order = cluster_data['Avg_Order_Value'].mean()
    
    print(f"Cluster {cluster}:")
    print(f"  客户数: {len(cluster_data)} ({len(cluster_data)/len(df)*100:.1f}%)")
    print(f"  平均年收入: ${avg_income:.0f}")
    print(f"  消费分数: {avg_spending:.1f}")
    print(f"  购买频率: {avg_frequency:.2f}次/月")
    print(f"  平均订单金额: ${avg_order:.0f}")
    
    # 分类和建议
    if avg_spending > 60:
        print(f"  分类: VIP客户 (高消费)")
        print(f"  建议:")
        print(f"    - 提供专属优惠和会员权益")
        print(f"    - 专业客户服务支持")
        print(f"    - 定期推送高端产品和新品")
    elif avg_spending > 40:
        print(f"  分类: Regular客户 (中等消费)")
        print(f"  建议:")
        print(f"    - 提供个性化推荐")
        print(f"    - 适度优惠活动")
        print(f"    - 鼓励提高购买频率")
    else:
        print(f"  分类: At-risk客户 (低消费)")
        print(f"  建议:")
        print(f"    - 实施挽留策略")
        print(f"    - 提供激励性优惠")
        print(f"    - 调查流失原因")
    print()