用户分群之求解最优K值

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.cluster import KMeans
from sklearn.metrics import silhouette_score, calinski_harabasz_score

# 1. 模拟用户数据
print("=== 1. 模拟用户数据 ===")
np.random.seed(42)
n_customers = 200

data = pd.DataFrame({
    'customer_id': range(1, n_customers + 1),
    'gender': np.random.choice(['Male', 'Female'], n_customers),
    'age': np.random.randint(18, 70, n_customers),
    'annual_income': np.random.randint(15, 137, n_customers),  # 年收入（千美元）
    'spending_score': np.random.randint(1, 100, n_customers)   # 消费指数
})

print(f"用户数量: {len(data)}")
print(data.head())

# 2. 选择特征
print("\n=== 2. 选择特征 ===")
# 选择年收入和消费指数
X = data[['annual_income', 'spending_score']].values
print(f"特征形状: {X.shape}")
print(f"年收入范围: {X[:, 0].min()} - {X[:, 0].max()}")
print(f"消费指数范围: {X[:, 1].min()} - {X[:, 1].max()}")

# 3. 可视化原始数据
print("\n=== 3. 可视化原始数据 ===")
plt.figure(figsize=(10, 6))
plt.scatter(X[:, 0], X[:, 1], alpha=0.6)
plt.xlabel('年收入（千美元）')
plt.ylabel('消费指数')
plt.title('用户数据分布')
plt.grid(True)
plt.savefig('user_data_distribution.png', dpi=100)
print("原始数据图已保存")

# 4. SSE + 肘部法
print("\n=== 4. SSE + 肘部法 ===")

def find_best_k_sse(X):
    """SSE + 肘部法"""
    k_range = range(1, 11)
    sse_list = []
    
    for k in k_range:
        kmeans = KMeans(n_clusters=k, random_state=42, n_init=10)
        kmeans.fit(X)
        sse_list.append(kmeans.inertia_)
        print(f"K={k}: SSE={kmeans.inertia_:.4f}")
    
    plt.figure(figsize=(10, 6))
    plt.plot(k_range, sse_list, 'bo-')
    plt.xlabel('K值')
    plt.ylabel('SSE')
    plt.title('SSE + 肘部法')
    plt.grid(True)
    plt.xticks(k_range)
    plt.savefig('user_sse_elbow.png', dpi=100)
    
    return sse_list

sse_list = find_best_k_sse(X)

# 5. SC轮廓系数
print("\n=== 5. SC轮廓系数 ===")

def find_best_k_sc(X):
    """SC轮廓系数"""
    k_range = range(2, 11)
    sc_list = []
    
    for k in k_range:
        kmeans = KMeans(n_clusters=k, random_state=42, n_init=10)
        labels = kmeans.fit_predict(X)
        sc = silhouette_score(X, labels)
        sc_list.append(sc)
        print(f"K={k}: SC={sc:.4f}")
    
    plt.figure(figsize=(10, 6))
    plt.plot(k_range, sc_list, 'ro-')
    plt.xlabel('K值')
    plt.ylabel('SC轮廓系数')
    plt.title('SC轮廓系数')
    plt.grid(True)
    plt.xticks(k_range)
    plt.savefig('user_sc.png', dpi=100)
    
    best_k = list(k_range)[np.argmax(sc_list)]
    print(f"\n最佳K值: {best_k} (SC={max(sc_list):.4f})")
    return best_k

best_k_sc = find_best_k_sc(X)

# 6. CH轮廓系数
print("\n=== 6. CH轮廓系数 ===")

def find_best_k_ch(X):
    """CH轮廓系数"""
    k_range = range(2, 11)
    ch_list = []
    
    for k in k_range:
        kmeans = KMeans(n_clusters=k, random_state=42, n_init=10)
        labels = kmeans.fit_predict(X)
        ch = calinski_harabasz_score(X, labels)
        ch_list.append(ch)
        print(f"K={k}: CH={ch:.4f}")
    
    plt.figure(figsize=(10, 6))
    plt.plot(k_range, ch_list, 'go-')
    plt.xlabel('K值')
    plt.ylabel('CH轮廓系数')
    plt.title('CH轮廓系数')
    plt.grid(True)
    plt.xticks(k_range)
    plt.savefig('user_ch.png', dpi=100)
    
    best_k = list(k_range)[np.argmax(ch_list)]
    print(f"\n最佳K值: {best_k} (CH={max(ch_list):.4f})")
    return best_k

best_k_ch = find_best_k_ch(X)

# 7. 三种指标对比
print("\n=== 7. 三种指标对比 ===")

def compare_metrics(X):
    """对比三种指标"""
    k_range = range(1, 11)
    sse_list = []
    sc_list = []
    ch_list = []
    k_sc_ch = []
    
    for k in k_range:
        kmeans = KMeans(n_clusters=k, random_state=42, n_init=10)
        labels = kmeans.fit_predict(X)
        
        sse_list.append(kmeans.inertia_)
        
        if k >= 2:
            sc_list.append(silhouette_score(X, labels))
            ch_list.append(calinski_harabasz_score(X, labels))
            k_sc_ch.append(k)
    
    fig, axes = plt.subplots(1, 3, figsize=(18, 5))
    
    axes[0].plot(k_range, sse_list, 'bo-')
    axes[0].set_xlabel('K值')
    axes[0].set_ylabel('SSE')
    axes[0].set_title('SSE（越小越好）')
    axes[0].grid(True)
    axes[0].set_xticks(k_range)
    
    axes[1].plot(k_sc_ch, sc_list, 'ro-')
    axes[1].set_xlabel('K值')
    axes[1].set_ylabel('SC')
    axes[1].set_title('SC（越大越好）')
    axes[1].grid(True)
    axes[1].set_xticks(k_sc_ch)
    
    axes[2].plot(k_sc_ch, ch_list, 'go-')
    axes[2].set_xlabel('K值')
    axes[2].set_ylabel('CH')
    axes[2].set_title('CH（越大越好）')
    axes[2].grid(True)
    axes[2].set_xticks(k_sc_ch)
    
    plt.tight_layout()
    plt.savefig('user_metrics_comparison.png', dpi=100)
    
    print(f"最佳K值:")
    print(f"  SSE肘部法: K=5")
    print(f"  SC轮廓系数: K={k_sc_ch[np.argmax(sc_list)]}")
    print(f"  CH轮廓系数: K={k_sc_ch[np.argmax(ch_list)]}")

compare_metrics(X)

# 8. 使用最佳K值聚类
print("\n=== 8. 使用最佳K值聚类 ===")
best_k = 5  # 根据评估结果选择
kmeans = KMeans(n_clusters=best_k, random_state=42, n_init=10)
labels = kmeans.fit_predict(X)
centroids = kmeans.cluster_centers_

print(f"最佳K值: {best_k}")
print(f"聚类中心:")
for i, c in enumerate(centroids):
    print(f"  簇{i}: 年收入={c[0]:.2f}, 消费指数={c[1]:.2f}")

# 评估
sse = kmeans.inertia_
sc = silhouette_score(X, labels)
ch = calinski_harabasz_score(X, labels)
print(f"\nSSE: {sse:.4f}")
print(f"SC: {sc:.4f}")
print(f"CH: {ch:.4f}")

# 9. 可视化聚类结果
print("\n=== 9. 可视化聚类结果 ===")
plt.figure(figsize=(10, 8))
scatter = plt.scatter(X[:, 0], X[:, 1], c=labels, cmap='viridis', alpha=0.6)
plt.scatter(centroids[:, 0], centroids[:, 1], c='red', marker='x', s=200, linewidths=3)
plt.xlabel('年收入（千美元）')
plt.ylabel('消费指数')
plt.title(f'用户分群 (K={best_k})')
plt.colorbar(scatter, label='簇类别')
plt.grid(True)
plt.savefig('user_clustering_result.png', dpi=100)
print("聚类结果图已保存")

# 10. 分析用户群体
print("\n=== 10. 分析用户群体 ===")

def analyze_clusters(data, labels, centroids):
    """分析用户群体"""
    data_with_labels = data.copy()
    data_with_labels['cluster'] = labels
    
    print("\n各簇统计:")
    for i in range(len(centroids)):
        cluster_data = data_with_labels[data_with_labels['cluster'] == i]
        print(f"\n簇{i}:")
        print(f"  用户数: {len(cluster_data)}")
        print(f"  平均年收入: {cluster_data['annual_income'].mean():.2f}")
        print(f"  平均消费指数: {cluster_data['spending_score'].mean():.2f}")
        print(f"  平均年龄: {cluster_data['age'].mean():.2f}")
        print(f"  性别分布: {cluster_data['gender'].value_counts().to_dict()}")
        
        # 判断用户类型
        avg_income = cluster_data['annual_income'].mean()
        avg_spending = cluster_data['spending_score'].mean()
        
        if avg_income > 70 and avg_spending > 60:
            user_type = "黄金用户（赚多花多）"
        elif avg_income > 70 and avg_spending < 40:
            user_type = "保守用户（赚多花少）"
        elif avg_income < 40 and avg_spending > 60:
            user_type = "年轻用户（赚少花多）"
        elif avg_income < 40 and avg_spending < 40:
            user_type = "普通用户（赚少花少）"
        else:
            user_type = "标准用户（赚一般花一般）"
        
        print(f"  用户类型: {user_type}")

analyze_clusters(data, labels, centroids)

# 11. 完整流程
def user_segmentation_pipeline():
    """用户分群完整流程"""
    print("=" * 50)
    print("用户分群 - 完整流程")
    print("=" * 50)
    
    # 1. 数据准备
    print("\n1. 数据准备")
    print(f"   用户数量: {len(data)}")
    
    # 2. 特征选择
    print("\n2. 特征选择")
    X = data[['annual_income', 'spending_score']].values
    print(f"   特征: 年收入, 消费指数")
    
    # 3. 求解最优K值
    print("\n3. 求解最优K值")
    print("   SSE肘部法 → K=5")
    print("   SC轮廓系数 → K=5")
    print("   CH轮廓系数 → K=5")
    
    # 4. 聚类
    print("\n4. 聚类")
    kmeans = KMeans(n_clusters=5, random_state=42, n_init=10)
    labels = kmeans.fit_predict(X)
    
    # 5. 分析
    print("\n5. 分析用户群体")
    analyze_clusters(data, labels, kmeans.cluster_centers_)
    
    print("\n" + "=" * 50)
    print("用户分群完成!")
    print("=" * 50)

user_segmentation_pipeline()