未记录事实核验 · 状态：维护中
记录仅说明核验时间与维护状态，不保证适用于所有环境；使用前请留意正文中的版本、适用条件与风险提示。
warning 适用边界与维护说明
本文适用于三种聚类算法的学习、参数比较及模拟RFM客户细分练习，不代表真实客户数据验证的业务结论。自动eps、MinPts=2×维度、指标推荐K值及算法选择助手均应作为候选建议，而非普适最优规则；find_optimal_k使用时须保证候选簇数小于样本数，且实际标签满足评分条件。混合类型数据需先明确编码和距离方案，不能直接将选择助手的推荐当作可执行方案。排除噪声后的评分仅描述保留样本，跨算法比较须结合有效样本覆盖范围、噪声比例和业务解释；若DBSCAN成为候选算法，输出中的-1仍表示噪声而非普通客户簇。RFM特征区分度图使用原始量纲的簇均值标准差，不能据柱高直接比较R、F、M的重要性。本次仅作结构与文本审阅，未运行代码或作外部事实核验。
Python聚类算法：无监督学习核心技术
聚类是无监督学习的重要分支，通过数据内在结构发现隐藏的模式和群体。本文深入探讨K-means、层次聚类和DBSCAN三种核心算法的数学原理、实现方法和实际应用。
聚类算法基础理论
无监督学习本质
聚类属于无监督学习范畴，目标是在没有标签信息的情况下，根据数据特征的相似性将样本分组：
$$\text{Input: } X = \{x_1, x_2, ..., x_n\} \in \mathbb{R}^{n \times d}$$$$\text{Output: } C = \{C_1, C_2, ..., C_k\}, \quad C_i \cap C_j = \emptyset \text{ for } i \neq j$$其中：
- $X$ 是 $n$ 个 $d$ 维样本的数据矩阵
- $C$ 是 $k$ 个不相交的簇集合
- 对于覆盖全部样本的硬划分聚类，每个样本 $x_i$ 属于且仅属于一个簇；DBSCAN等含噪声识别的算法允许部分样本不归入任何簇。
相似性度量
聚类算法的核心是相似性度量。常用距离度量包括：
- 欧氏距离：$d(x_i, x_j) = \sqrt{\sum_{k=1}^{d} (x_{ik} - x_{jk})^2}$
- 曼哈顿距离：$d(x_i, x_j) = \sum_{k=1}^{d} |x_{ik} - x_{jk}|$
- 余弦距离：$d(x_i, x_j) = 1 - \frac{x_i \cdot x_j}{||x_i|| \cdot ||x_j||}$
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.cluster import KMeans, AgglomerativeClustering, DBSCAN
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import silhouette_score, davies_bouldin_score, calinski_harabasz_score
from sklearn.datasets import make_blobs
K-means聚类算法
数学原理
K-means算法通过最小化簇内平方误差和（WCSS）来进行聚类：
$$J = \sum_{i=1}^{k} \sum_{x \in C_i} ||x - \mu_i||^2$$其中：
- $k$ 是预设的簇数量
- $C_i$ 是第 $i$ 个簇
- $\mu_i$ 是第 $i$ 个簇的中心点
算法步骤：
- 随机初始化 $k$ 个簇中心 $\mu_1, \mu_2, ..., \mu_k$
- 分配步骤：将每个点分配给最近的簇中心 $$c^{(i)} = \arg\min_j ||x^{(i)} - \mu_j||^2$$
- 更新步骤：重新计算每个簇的中心点 $$\mu_j = \frac{1}{|C_j|} \sum_{x \in C_j} x$$
- 重复步骤2和3直至收敛
K-means++初始化优化
标准K-means对初始中心点敏感，K-means++改进了初始化策略：
def demonstrate_kmeans():
"""K-means算法演示"""
# 生成测试数据
np.random.seed(42)
X, y_true = make_blobs(n_samples=300, centers=4, cluster_std=1.5, random_state=42)
# 标准化数据
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)
# 比较不同初始化方法
init_methods = ['random', 'k-means++']
results = {}
plt.figure(figsize=(12, 5))
for i, init_method in enumerate(init_methods):
# 训练模型
kmeans = KMeans(n_clusters=4, init=init_method, random_state=42, n_init=1)
labels = kmeans.fit_predict(X_scaled)
centers = scaler.inverse_transform(kmeans.cluster_centers_)
# 计算评估指标
silhouette = silhouette_score(X_scaled, labels)
davies_bouldin = davies_bouldin_score(X_scaled, labels)
results[init_method] = {
'silhouette_score': silhouette,
'davies_bouldin_score': davies_bouldin,
'inertia': kmeans.inertia_
}
# 可视化结果
plt.subplot(1, 2, i+1)
scatter = plt.scatter(X[:, 0], X[:, 1], c=labels, cmap='viridis', alpha=0.6)
plt.scatter(centers[:, 0], centers[:, 1], c='red', marker='x',
s=200, linewidths=3, label='聚类中心')
plt.title(f'{init_method} 初始化\n轮廓系数: {silhouette:.3f}')
plt.xlabel('特征1')
plt.ylabel('特征2')
plt.legend()
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
# 输出比较结果
print("初始化方法比较:")
for method, metrics in results.items():
print(f"{method}:")
print(f" 轮廓系数: {metrics['silhouette_score']:.3f}")
print(f" Davies-Bouldin指数: {metrics['davies_bouldin_score']:.3f}")
print(f" 簇内平方和: {metrics['inertia']:.2f}")
return results
# 运行演示
results = demonstrate_kmeans()
最优K值选择
使用肘部法则和轮廓分析确定最优簇数：
def find_optimal_k(X, max_k=10):
"""寻找最优K值"""
k_range = range(2, max_k + 1)
inertias = []
silhouette_scores = []
davies_bouldin_scores = []
for k in k_range:
kmeans = KMeans(n_clusters=k, init='k-means++', random_state=42)
labels = kmeans.fit_predict(X)
inertias.append(kmeans.inertia_)
silhouette_scores.append(silhouette_score(X, labels))
davies_bouldin_scores.append(davies_bouldin_score(X, labels))
# 可视化评估结果
fig, axes = plt.subplots(1, 3, figsize=(15, 4))
# 肘部法则
axes[0].plot(k_range, inertias, 'bo-')
axes[0].set_xlabel('K值')
axes[0].set_ylabel('簇内平方和 (WCSS)')
axes[0].set_title('肘部法则')
axes[0].grid(True, alpha=0.3)
# 轮廓系数
axes[1].plot(k_range, silhouette_scores, 'ro-')
axes[1].set_xlabel('K值')
axes[1].set_ylabel('轮廓系数')
axes[1].set_title('轮廓系数分析')
axes[1].grid(True, alpha=0.3)
# Davies-Bouldin指数
axes[2].plot(k_range, davies_bouldin_scores, 'go-')
axes[2].set_xlabel('K值')
axes[2].set_ylabel('Davies-Bouldin指数')
axes[2].set_title('Davies-Bouldin指数 (越小越好)')
axes[2].grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
# 推荐最优K值
optimal_k_silhouette = k_range[np.argmax(silhouette_scores)]
optimal_k_db = k_range[np.argmin(davies_bouldin_scores)]
print(f"基于轮廓系数的最优K值: {optimal_k_silhouette}")
print(f"基于Davies-Bouldin指数的最优K值: {optimal_k_db}")
return k_range, inertias, silhouette_scores, davies_bouldin_scores
层次聚类算法
算法原理
层次聚类构建数据点的层次结构，分为两种策略：
凝聚层次聚类（Agglomerative）：自底向上，从单个数据点开始逐步合并 分裂层次聚类（Divisive）：自顶向下，从整体数据开始逐步分割
连接准则
不同连接准则定义簇间距离的计算方法：
-
Ward连接：最小化合并后簇的方差增量 $$d(C_i, C_j) = \sqrt{\frac{|C_i| \cdot |C_j|}{|C_i| + |C_j|}} ||\mu_i - \mu_j||$$
-
Complete连接：两簇中最远点对的距离 $$d(C_i, C_j) = \max_{x \in C_i, y \in C_j} d(x, y)$$
-
Average连接：两簇所有点对距离的平均值 $$d(C_i, C_j) = \frac{1}{|C_i| \cdot |C_j|} \sum_{x \in C_i} \sum_{y \in C_j} d(x, y)$$
-
Single连接：两簇中最近点对的距离 $$d(C_i, C_j) = \min_{x \in C_i, y \in C_j} d(x, y)$$
from sklearn.datasets import make_circles
from scipy.cluster.hierarchy import dendrogram, linkage
from scipy.spatial.distance import pdist
def hierarchical_clustering_demo():
"""层次聚类演示"""
# 生成测试数据
np.random.seed(42)
X, y_true = make_blobs(n_samples=150, centers=3, cluster_std=1.2, random_state=42)
# 标准化数据
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)
# 比较不同连接方法
linkage_methods = ['ward', 'complete', 'average', 'single']
plt.figure(figsize=(16, 10))
for i, method in enumerate(linkage_methods):
# 层次聚类
hierarchical = AgglomerativeClustering(n_clusters=3, linkage=method)
labels = hierarchical.fit_predict(X_scaled)
# 计算评估指标
silhouette = silhouette_score(X_scaled, labels)
# 聚类结果可视化
plt.subplot(2, 4, i+1)
scatter = plt.scatter(X[:, 0], X[:, 1], c=labels, cmap='viridis', alpha=0.7)
plt.title(f'{method.title()}连接\n轮廓系数: {silhouette:.3f}')
plt.xlabel('特征1')
plt.ylabel('特征2')
plt.grid(True, alpha=0.3)
# 树状图可视化
plt.subplot(2, 4, i+5)
if method == 'ward':
# Ward方法需要使用欧氏距离
Z = linkage(X_scaled, method=method)
else:
Z = linkage(X_scaled, method=method, metric='euclidean')
dendrogram(Z, truncate_mode='level', p=3, no_labels=True)
plt.title(f'{method.title()}连接\n树状图')
plt.xlabel('样本索引')
plt.ylabel('距离')
plt.tight_layout()
plt.show()
return X_scaled, y_true
# 运行演示
X_scaled, y_true = hierarchical_clustering_demo()
DBSCAN密度聚类
算法原理
DBSCAN（Density-Based Spatial Clustering of Applications with Noise）基于密度连接性进行聚类，能发现任意形状的簇并识别异常点。
核心概念：
- 核心点：ε-邻域内至少包含MinPts个点的点
- 边界点：不是核心点但在某个核心点的ε-邻域内
- 噪声点：既不是核心点也不是边界点的点
数学定义： 给定点$p$和参数$\varepsilon$、$MinPts$：
- $N_\varepsilon(p) = \{q \in X | d(p,q) \leq \varepsilon\}$ （ε-邻域）
- 核心点：$|N_\varepsilon(p)| \geq MinPts$
- 密度直达：若$q \in N_\varepsilon(p)$且$p$为核心点，则$q$从$p$密度直达
- 密度可达：存在点链$p_1, p_2, ..., p_n$使得$p_{i+1}$从$p_i$密度直达
from sklearn.datasets import make_moons, make_circles
from sklearn.neighbors import NearestNeighbors
def dbscan_demonstration():
"""DBSCAN算法演示"""
# 生成不同形状的数据集
np.random.seed(42)
# 月牙形数据
X_moons, _ = make_moons(n_samples=300, noise=0.1, random_state=42)
# 同心圆数据
X_circles, _ = make_circles(n_samples=300, noise=0.05, factor=0.6, random_state=42)
# 不规则形状数据
X_blobs, _ = make_blobs(n_samples=300, centers=4, cluster_std=0.60, random_state=42)
datasets = [X_moons, X_circles, X_blobs]
dataset_names = ['月牙形', '同心圆', '球形簇']
plt.figure(figsize=(15, 10))
for i, (X, name) in enumerate(zip(datasets, dataset_names)):
# 标准化数据
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)
# 原始数据
plt.subplot(3, 4, i*4 + 1)
plt.scatter(X[:, 0], X[:, 1], alpha=0.6)
plt.title(f'{name}\n原始数据')
plt.grid(True, alpha=0.3)
# K-means聚类
plt.subplot(3, 4, i*4 + 2)
kmeans = KMeans(n_clusters=2, random_state=42)
kmeans_labels = kmeans.fit_predict(X_scaled)
plt.scatter(X[:, 0], X[:, 1], c=kmeans_labels, cmap='viridis', alpha=0.6)
plt.title(f'{name}\nK-means')
plt.grid(True, alpha=0.3)
# 自动找最优eps参数
neighbors = NearestNeighbors(n_neighbors=4)
neighbors_fit = neighbors.fit(X_scaled)
distances, indices = neighbors_fit.kneighbors(X_scaled)
distances = np.sort(distances[:, -1])
# 使用二阶差分启发式选取候选eps，不保证最优
knee_point = np.argmax(np.diff(np.diff(distances))) + 2
optimal_eps = distances[knee_point] if knee_point < len(distances) else 0.3
# DBSCAN聚类
plt.subplot(3, 4, i*4 + 3)
dbscan = DBSCAN(eps=optimal_eps, min_samples=4)
dbscan_labels = dbscan.fit_predict(X_scaled)
plt.scatter(X[:, 0], X[:, 1], c=dbscan_labels, cmap='viridis', alpha=0.6)
n_clusters = len(set(dbscan_labels)) - (1 if -1 in dbscan_labels else 0)
n_noise = list(dbscan_labels).count(-1)
plt.title(f'{name}\nDBSCAN\n簇数:{n_clusters}, 噪声:{n_noise}')
plt.grid(True, alpha=0.3)
# K-distance图（用于eps选择）
plt.subplot(3, 4, i*4 + 4)
plt.plot(distances)
plt.axvline(x=knee_point, color='red', linestyle='--', alpha=0.7)
plt.axhline(y=optimal_eps, color='red', linestyle='--', alpha=0.7)
plt.title(f'{name}\nK-distance图\neps={optimal_eps:.3f}')
plt.xlabel('点索引（按距离排序）')
plt.ylabel('4-距离')
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
return datasets, dataset_names
# 运行演示
datasets, names = dbscan_demonstration()
聚类评估指标
核心评估方法
轮廓系数（Silhouette Score）： $$s(i) = \frac{b(i) - a(i)}{\max\{a(i), b(i)\}}$$
其中：
- $a(i)$：样本$i$到同簇内其他点的平均距离
- $b(i)$：样本$i$到最近邻簇内所有点的平均距离
- 取值范围：[-1, 1]，越大越好
Davies-Bouldin指数： $$DB = \frac{1}{k} \sum_{i=1}^{k} \max_{j \neq i} \left\{ \frac{\sigma_i + \sigma_j}{d(c_i, c_j)} \right\}$$
其中$\sigma_i$是簇$i$内点到中心的平均距离，$d(c_i, c_j)$是簇中心间距离。取值越小越好。
def evaluate_clustering_metrics(X, labels):
"""聚类评估指标计算"""
if len(set(labels)) < 2:
return None
# 处理噪声点
if -1 in labels:
mask = labels != -1
X_clean = X[mask]
labels_clean = labels[mask]
else:
X_clean = X
labels_clean = labels
if not (2 <= len(set(labels_clean)) < len(labels_clean)):
return None
# 计算评估指标
silhouette = silhouette_score(X_clean, labels_clean)
davies_bouldin = davies_bouldin_score(X_clean, labels_clean)
calinski_harabasz = calinski_harabasz_score(X_clean, labels_clean)
metrics = {
'silhouette_score': silhouette,
'davies_bouldin_score': davies_bouldin,
'calinski_harabasz_score': calinski_harabasz,
'n_clusters': len(set(labels_clean)),
'n_noise': list(labels).count(-1) if -1 in labels else 0
}
return metrics
客户细分实战案例
RFM客户价值分析
基于客户消费行为的RFM模型进行聚类分析：
def customer_segmentation_case():
"""客户细分完整案例"""
# 生成模拟客户RFM数据
np.random.seed(42)
n_customers = 1000
# 创建不同客户群体的特征
# 高价值客户：最近购买、频次高、金额大
high_value = pd.DataFrame({
'recency': np.random.exponential(10, 200), # 最近购买天数
'frequency': np.random.poisson(15, 200), # 购买频次
'monetary': np.random.lognormal(8, 0.5, 200) # 消费金额
})
# 中等价值客户
medium_value = pd.DataFrame({
'recency': np.random.exponential(30, 400),
'frequency': np.random.poisson(8, 400),
'monetary': np.random.lognormal(6.5, 0.8, 400)
})
# 低价值客户
low_value = pd.DataFrame({
'recency': np.random.exponential(60, 300),
'frequency': np.random.poisson(3, 300),
'monetary': np.random.lognormal(5, 1, 300)
})
# 流失客户
churned = pd.DataFrame({
'recency': np.random.exponential(120, 100),
'frequency': np.random.poisson(1, 100),
'monetary': np.random.lognormal(4, 1.2, 100)
})
# 合并数据
df = pd.concat([high_value, medium_value, low_value, churned], ignore_index=True)
true_labels = np.array([0]*200 + [1]*400 + [2]*300 + [3]*100)
# 添加噪声和异常值
df['recency'] = np.clip(df['recency'], 1, 365)
df['frequency'] = np.clip(df['frequency'], 1, 50)
df['monetary'] = np.clip(df['monetary'], 10, 10000)
print("客户RFM数据统计:")
print(df.describe())
# 特征标准化
scaler = StandardScaler()
X_scaled = scaler.fit_transform(df)
# 比较不同聚类算法
algorithms = {
'K-means': KMeans(n_clusters=4, random_state=42),
'Hierarchical': AgglomerativeClustering(n_clusters=4, linkage='ward'),
'DBSCAN': DBSCAN(eps=0.8, min_samples=10)
}
results = {}
plt.figure(figsize=(15, 10))
for i, (name, algorithm) in enumerate(algorithms.items()):
# 训练模型
labels = algorithm.fit_predict(X_scaled)
# 评估指标
metrics = evaluate_clustering_metrics(X_scaled, labels)
if metrics:
results[name] = metrics
results[name]['labels'] = labels
# 可视化前两个主成分
plt.subplot(2, 3, i+1)
from sklearn.decomposition import PCA
pca = PCA(n_components=2)
X_pca = pca.fit_transform(X_scaled)
scatter = plt.scatter(X_pca[:, 0], X_pca[:, 1], c=labels, cmap='viridis', alpha=0.6)
plt.title(f'{name}\n轮廓系数: {metrics["silhouette_score"]:.3f}' if metrics else name)
plt.xlabel(f'PC1 ({pca.explained_variance_ratio_[0]:.1%})')
plt.ylabel(f'PC2 ({pca.explained_variance_ratio_[1]:.1%})')
plt.grid(True, alpha=0.3)
# RFM特征分布
if i < 3: # 只为前3个算法绘制
plt.subplot(2, 3, i+4)
feature_importance = []
for j in range(3): # R, F, M
unique_labels = set(labels) - {-1} if -1 in labels else set(labels)
if len(unique_labels) > 1:
cluster_means = []
for label in sorted(unique_labels):
mask = labels == label
cluster_means.append(df.iloc[mask, j].mean())
feature_importance.append(np.std(cluster_means))
else:
feature_importance.append(0)
plt.bar(['Recency', 'Frequency', 'Monetary'], feature_importance)
plt.title(f'{name}\nRFM特征区分度')
plt.ylabel('标准差')
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
# 选择最佳算法
if results:
best_algorithm = max(results.keys(), key=lambda x: results[x]['silhouette_score'])
print(f"\n按当前有效样本轮廓系数选出的候选算法: {best_algorithm}（还需结合噪声比例和业务解释比较）")
best_labels = results[best_algorithm]['labels']
df['cluster'] = best_labels
# 客户群体特征分析
print("\n客户群体RFM特征分析:")
cluster_analysis = df.groupby('cluster').agg({
'recency': ['mean', 'std'],
'frequency': ['mean', 'std'],
'monetary': ['mean', 'std'],
'cluster': 'count'
}).round(2)
cluster_analysis.columns = ['_'.join(col).strip() for col in cluster_analysis.columns]
print(cluster_analysis)
# 客户群体标签化
cluster_profiles = {}
cluster_stats = df.groupby('cluster')[['recency', 'frequency', 'monetary']].mean()
for cluster in cluster_stats.index:
r_score = 'Low' if cluster_stats.loc[cluster, 'recency'] > df['recency'].median() else 'High'
f_score = 'High' if cluster_stats.loc[cluster, 'frequency'] > df['frequency'].median() else 'Low'
m_score = 'High' if cluster_stats.loc[cluster, 'monetary'] > df['monetary'].median() else 'Low'
profile = f"R:{r_score}-F:{f_score}-M:{m_score}"
cluster_profiles[cluster] = profile
print("\n客户群体标签:")
for cluster, profile in cluster_profiles.items():
count = (df['cluster'] == cluster).sum()
print(f"群体 {cluster}: {profile} ({count}人, {count/len(df)*100:.1f}%)")
return df, results
# 运行客户细分案例
df_customers, clustering_results = customer_segmentation_case()
最佳实践与优化建议
算法选择决策树
def clustering_algorithm_selector(data_shape, data_type, noise_tolerance):
"""聚类算法选择助手"""
n_samples, n_features = data_shape
recommendations = []
if data_type == 'spherical': # 球形簇
if n_samples < 10000:
recommendations.append("K-means (快速收敛)")
recommendations.append("层次聚类-Ward (可解释性强)")
else:
recommendations.append("Mini-batch K-means (内存友好)")
elif data_type == 'arbitrary': # 任意形状
if noise_tolerance == 'high':
recommendations.append("DBSCAN (处理噪声)")
else:
recommendations.append("谱聚类 (非凸簇)")
elif data_type == 'mixed': # 混合类型
recommendations.append("K-means (基线)")
recommendations.append("DBSCAN (发现异常)")
recommendations.append("层次聚类 (层次结构)")
print(f"数据特征: {n_samples}样本 × {n_features}特征")
print(f"数据类型: {data_type}")
print(f"噪声容忍度: {noise_tolerance}")
print("推荐算法:")
for i, rec in enumerate(recommendations, 1):
print(f"{i}. {rec}")
return recommendations
参数调优策略
- K-means: 使用肘部法则 + 轮廓分析
- DBSCAN: K-distance图确定eps，MinPts=2×维度
- 层次聚类: 根据业务需求选择连接准则
性能优化建议
- 大数据集: 使用Mini-batch K-means或采样
- 高维数据: 先进行PCA降维
- 混合数据类型: 使用Gower距离
- 在线聚类: 考虑流式算法
通过本文的学习，你已经掌握了Python中三种核心聚类算法的原理、实现和应用。在实际项目中，建议根据数据特征和业务需求选择合适的算法，并通过多种评估指标验证聚类效果。
相关链接：
标签：#Python #聚类算法 #无监督学习 #K-means #层次聚类 #DBSCAN #客户细分
创建时间：2024-06-29
最后更新：2025-01-02
字数统计：约6800字
阅读时长：45-50分钟