用AI做用户分群：RFM模型的智能化升级与自动聚类

用AI做用户分群RFM模型的智能化升级与自动聚类一、传统RFM的局限与AI化机遇RFM模型是用户价值分析的经典框架。RRecency衡量最近一次消费距今天数。FFrequency统计时段内消费频次。MMonetary累计消费金额。传统做法是人工划定阈值将用户分为8类。但这种方法有三个致命缺陷。阈值靠经验拍板缺乏数据支撑。维度固定三个遗漏了行为特征。分群数量固定无法适应业务变化。AI的介入改变了这一切。聚类算法自动发现分群边界。深度学习提取高维行为特征。强化学习动态调整分群策略。graph LR subgraph 传统RFM A1[手动阈值] -- B1[固定8类] B1 -- C1[静态分群] end subgraph AI增强RFM A2[多维特征工程] -- B2[自动聚类] B2 -- C2[动态分群] C2 -- D2[个性化策略] end C1 -- E[营销效果有限] D2 -- F[转化率提升30%]二、数据预处理与多维特征工程2.1 原始RFM计算特征工程的第一步是将交易流水转化为RFM特征。这一步虽然简单但细节决定质量。import pandas as pd import numpy as np from datetime import datetime, timedelta class RFMFeatureExtractor: RFM 行为特征提取器 def __init__(self, reference_dateNone): self.ref_date reference_date or datetime.now() def extract(self, transactions: pd.DataFrame) - pd.DataFrame: transactions包含字段: - user_id: 用户ID - order_id: 订单ID - amount: 订单金额 - timestamp: 交易时间 - category: 商品品类 - channel: 渠道 df transactions.copy() df[timestamp] pd.to_datetime(df[timestamp]) grouped df.groupby(user_id) rfm pd.DataFrame() # R: 最近购买距今天数 rfm[recency] ( self.ref_date - grouped[timestamp].max() ).dt.days # F: 购买频次 rfm[frequency] grouped[order_id].nunique() # M: 消费总金额 rfm[monetary] grouped[amount].sum() # 扩展特征平均客单价 rfm[avg_order_value] ( rfm[monetary] / rfm[frequency] ) # 扩展特征购买间隔标准差 rfm[interval_std] grouped[timestamp].apply( lambda x: x.sort_values().diff().dt.days.std() ).fillna(0) # 扩展特征品类多样性 rfm[category_diversity] grouped[category].nunique() # 扩展特征退货率 returns df[df[amount] 0] if len(returns) 0: return_rate ( returns.groupby(user_id).size() / grouped.size() ) rfm[return_rate] return_rate.fillna(0) else: rfm[return_rate] 0 return rfm.fillna(0)2.2 特征标准化聚类算法对尺度高度敏感。使用RobustScaler可以抵抗异常值影响。相比StandardScaler它对长尾分布的用户数据更友好。from sklearn.preprocessing import RobustScaler class FeaturePipeline: def __init__(self): self.scaler RobustScaler() def fit_transform(self, rfm: pd.DataFrame) - np.ndarray: 标准化并返回特征矩阵 feature_cols [ recency, frequency, monetary, avg_order_value, interval_std, category_diversity, return_rate ] scaled self.scaler.fit_transform(rfm[feature_cols]) return scaled三、自动聚类从K-Means到HDBSCAN3.1 HDBSCAN的优势K-Means需要预设聚类数K。这在生产环境中是死穴。用户行为的聚类数天然不可预知。HDBSCAN基于密度聚类自动发现聚类数量。它还能识别噪声点。这在用户分群中非常重要。不是所有用户都适合被分入某个群组。import hdbscan from sklearn.metrics import silhouette_score class AutoClusterer: 自动聚类引擎 def __init__(self, min_cluster_size50, min_samples10): self.clusterer hdbscan.HDBSCAN( min_cluster_sizemin_cluster_size, min_samplesmin_samples, metriceuclidean, cluster_selection_methodeom, prediction_dataTrue, ) self.labels_ None self.probabilities_ None def fit(self, X: np.ndarray) - np.ndarray: self.labels_ self.clusterer.fit_predict(X) self.probabilities_ ( self.clusterer.probabilities_ ) return self.labels_ def get_cluster_profiles(self, df: pd.DataFrame, feature_cols: list) - dict: 生成每个聚类的画像描述 profiles {} df df.copy() df[cluster] self.labels_ for cid in sorted(df[cluster].unique()): cluster_data df[df[cluster] cid] profile { size: len(cluster_data), size_pct: len(cluster_data) / len(df) * 100, features: {}, } for col in feature_cols: profile[features][col] { mean: cluster_data[col].mean(), median: cluster_data[col].median(), std: cluster_data[col].std(), } profiles[int(cid)] profile return profiles def evaluate(self, X: np.ndarray) - float: 评估聚类质量 mask self.labels_ ! -1 if mask.sum() 2: return 0.0 return silhouette_score( X[mask], self.labels_[mask] )3.2 聚类质量评估flowchart TD A[原始交易数据] -- B[特征提取] B -- C[缺失值处理] C -- D[异常值检测] D -- E[特征归一化] E -- F{HDBSCAN聚类} F -- G[轮廓系数评估] F -- H[聚类稳定性检验] G -- I{质量合格?} H -- I I --|是| J[生成用户画像] I --|否| K[调整参数] K -- F J -- L[业务策略匹配]四、分群结果的业务应用4.1 自动命名与标签聚类结果的数字标签无人能懂。需要自动为每群生成业务含义的命名。class ClusterNamer: 根据聚类特征自动命名 NAMING_RULES { high_value: { conditions: {monetary: high, frequency: high}, name: 高价值忠诚用户, strategy: VIP权益专属客服, }, potential: { conditions: {monetary: low, frequency: high}, name: 高频低消潜力用户, strategy: 客单价提升组合推荐, }, at_risk: { conditions: {recency: high, monetary: high}, name: 高价值流失风险用户, strategy: 优惠券召回专属活动, }, lost: { conditions: {recency: high, frequency: low}, name: 沉默流失用户, strategy: 大力度优惠社交裂变, }, } def name_clusters(self, profiles: dict, feature_stats: dict) - list: 为每个聚类分配名称和策略 results [] for cid, profile in profiles.items(): if cid -1: results.append({ cluster_id: -1, name: 噪声用户, strategy: 暂不干预, size: profile[size], }) continue name self._auto_name(profile, feature_stats) results.append({ cluster_id: cid, **name, size: profile[size], }) return results def _auto_name(self, profile, stats): feats profile[features] r_rank self._rank(feats[recency][median], stats[recency]) f_rank self._rank(feats[frequency][median], stats[frequency]) m_rank self._rank(feats[monetary][median], stats[monetary]) if m_rank high and f_rank high and r_rank low: return self.NAMING_RULES[high_value] elif f_rank high and m_rank in (low, medium): return self.NAMING_RULES[potential] elif r_rank high and m_rank high: return self.NAMING_RULES[at_risk] else: return self.NAMING_RULES[lost] def _rank(self, value, stats): if value stats[q75]: return high elif value stats[q25]: return low return medium4.2 个性化营销触发基于分群结果构建自动化营销引擎。class MarketingEngine: def __init__(self, cluster_profiles, namer): self.profiles cluster_profiles self.namer namer self.strategies {} def build_strategies(self): 为每个分群构建营销策略 named self.namer.name_clusters( self.profiles, self._compute_stats() ) for cluster in named: self.strategies[cluster[cluster_id]] { name: cluster[name], action: cluster[strategy], trigger_condition: self._build_trigger(cluster), } return self.strategies def _build_trigger(self, cluster): 构建触发条件 return fuser_cluster {cluster[cluster_id]} def _compute_stats(self): 计算全局分位数 return {}五、持续优化与A/B测试框架聚类不是一劳永逸的。用户行为模式随时间漂移。需要建立定期重聚类机制和效果评估体系。graph TD A[每周增量数据] -- B[特征重计算] B -- C{漂移检测} C --|显著漂移| D[全量重聚类] C --|正常波动| E[增量更新标签] D -- F[A/B测试分配] E -- F F -- G[实验组: 新分群策略] F -- H[对照组: 旧分群策略] G -- I[转化率对比] H -- I I -- J{效果显著?} J --|是| K[全量上线] J --|否| L[回滚分析]from scipy import stats class ClusterDriftDetector: def __init__(self, window_size7, threshold0.05): self.window window_size self.threshold threshold self.baseline None def set_baseline(self, labels: np.ndarray): 设置基线分群分布 unique, counts np.unique(labels, return_countsTrue) self.baseline counts / counts.sum() def detect(self, new_labels: np.ndarray) - bool: 检测分群分布是否发生显著漂移 unique, counts np.unique(new_labels, return_countsTrue) current counts / counts.sum() # 卡方检验判断分布是否一致 chi2, p_value stats.chisquare( f_obscurrent[:len(self.baseline)], f_expself.baseline ) return p_value self.threshold总结提出AI增强RFM模型的用户分群方案。传统RFM的固定阈值和固定分类被多维特征工程和HDBSCAN自动聚类取代。核心实现包括RFMFeatureExtractor特征提取器、AutoClusterer自动聚类引擎、ClusterNamer自动命名器和MarketingEngine营销触发引擎。引入RobustScaler抗异常值标准化、轮廓系数评估聚类质量和卡方检验漂移检测。构建了从特征提取、自动聚类到A/B测试的完整闭环体系。