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

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

一、传统RFM的局限与AI化机遇

RFM模型是用户价值分析的经典框架。R（Recency）衡量最近一次消费距今天数。F（Frequency）统计时段内消费频次。M（Monetary）累计消费金额。传统做法是人工划定阈值，将用户分为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_date=None): 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到HDBSCAN

3.1 HDBSCAN的优势

K-Means需要预设聚类数K。这在生产环境中是死穴。用户行为的聚类数天然不可预知。HDBSCAN基于密度聚类，自动发现聚类数量。它还能识别噪声点。这在用户分群中非常重要。不是所有用户都适合被分入某个群组。

import hdbscan from sklearn.metrics import silhouette_score class AutoClusterer: """自动聚类引擎""" def __init__(self, min_cluster_size=50, min_samples=10): self.clusterer = hdbscan.HDBSCAN( min_cluster_size=min_cluster_size, min_samples=min_samples, metric='euclidean', cluster_selection_method='eom', prediction_data=True, ) 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 'medium'

4.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 f"user_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_size=7, threshold=0.05): self.window = window_size self.threshold = threshold self.baseline = None def set_baseline(self, labels: np.ndarray): """设置基线分群分布""" unique, counts = np.unique(labels, return_counts=True) self.baseline = counts / counts.sum() def detect(self, new_labels: np.ndarray) -> bool: """检测分群分布是否发生显著漂移""" unique, counts = np.unique(new_labels, return_counts=True) current = counts / counts.sum() # 卡方检验判断分布是否一致 chi2, p_value = stats.chisquare( f_obs=current[:len(self.baseline)], f_exp=self.baseline ) return p_value < self.threshold

总结：提出AI增强RFM模型的用户分群方案。传统RFM的固定阈值和固定分类被多维特征工程和HDBSCAN自动聚类取代。核心实现包括RFMFeatureExtractor特征提取器、AutoClusterer自动聚类引擎、ClusterNamer自动命名器和MarketingEngine营销触发引擎。引入RobustScaler抗异常值标准化、轮廓系数评估聚类质量和卡方检验漂移检测。构建了从特征提取、自动聚类到A/B测试的完整闭环体系。