客户价值模型-RFM
1.RFM模型引入
比如电商网站要做一次营销活动，需要针对不同价值的客户群体进行分群，对于高价值的用户推荐手表，珠宝等高端商品，对于低价值用户推荐打折促销的廉价商品，当然还有以下这些问题都是需要考虑的：
1.谁是我的最佳客户？
2.谁即将要成为流失客户？
3.谁将有潜力成为有价值的客户
4.哪些客户能够留存？
5.哪些客户会对你目前对活动有所反应？
那么最终的问题是如何对客户进行分群，即如何建立客户的价值模型呢？
在传统企业和电商众多的客户细分模型中，RFM模型是被广泛提到和使用的。
RFM模型是衡量当前用户价值和客户潜在价值的重要工具和手段。
==RFM是Rencency（最近一次消费），Frequency（消费频率）、Monetary（消费金额），三个指标首字母组合，==如图所示：
一般情况下RFM模型可以说明下列几个事实：
1.最近购买的时间越近，用户对产品促销互动越大
2.客户购买的频率越高，客户就品牌的满意度就越大
3.货币价值将高消费客户和低消费客户区分开来 。
如图所示，根据RFM模型，就可以统计在某一段时间内，用户最近的消费间隔，消费次数和消费金额，再根据使用k-means算法对用户进行聚类分群
注意一点，不仅仅可以局限于这三个数据字段，还可以根据业务需求，加入其他字段，进行调整模型。
我们可以根据RFM模型计算出所有用户的RFM值形成一个二维表:
userid, R值, F值, M值
1 -11-01 5 10000
2 -10-01 4 800
对于以上数据的量纲不一致(单位不统一),所以要对数据进行归一化
如何归一化?—需要自定义归一化的规则!即建立一个评分标准?
如何建立评分标准?----根据运营/产品的经验,做一个标准
R: 1-3天=5分，4-6天=4分，7-9天=3分，10-15天=2分，大于16天=1分
F: ≥200=5分，150-199=4分，100-149=3分，50-99=2分，1-49=1分
M: ≥20w=5分，10-19w=4分，5-9w=3分，1-4w=2分，<1w=1分
根据上面的打分规则就可以对数据进行自定义的归一化,得到如下类似结果:
userid, R值, F值, M值
1 5 1 2
2 1 1 1
那么现在数据已经归一化了,如何对数据进行分类?–肯定不能简单的将数据直接丢到三维坐标系,因为坐标系的原点不好确定,且三维坐标系只能分为8类
所以应该使用算法进行分类(聚类)–让算法自动学习用户之间的相似度,然后相似度高的用户,自动聚成一类,最后完成聚类的划分
计算流程
首先对所有用户的最近一次消费时间/总共消费次数/总共消费金额进行统计 再进行归一化(运营/产品提供的打分规则) 再使用算法进行聚类(K-Means) 根据聚类结果给用户打Tag(标签)
2.1.1.RFM详解
2.1.1.1.R值：最近一次消费（Recency）
消费指的是客户在店铺消费最近一次和上一次的时间间隔，理论上R值越小的客户是价值越高的客户，即对店铺的回购几次最有可能产生回应。目前网购便利，顾客已经有了更多的购买选择和更低的购买成本，去除地域的限制因素，客户非常容易流失，因此CRM操盘手想要提高回购率和留存率，需要时刻警惕R值。
如下图，某零食网店用户最近一次消费R值分布图：
户R值呈规律性的“波浪形”分布，时间越长，波浪越小；
2、最近一年内用户占比50%（真的很巧）；
数据分析：这个数据根据向行业内专业人员请教，已经是比较理想了的。说明每引入2个客户，就有一位用户在持续购买。说明店铺复购做的比较好。
2.1.1.2.F值：消费频率（Frequency）
消费频率是客户在固定时间内的购买次数（一般是1年）。但是如果实操中实际店铺由于受品类宽度的原因，比如卖3C产品，耐用品等即使是忠实粉丝用户也很难在1年内购买多次。所以，有些店铺在运营RFM模型时，会把F值的时间范围去掉，替换成累计购买次数。
如下图，某零食网店用户购买频次图（如1个客户在1天内购买多笔订单，则自动合并为1笔订单）：
1、购买1次（新客户）占比为65.5%，产生重复购买（老客户）的占比为34.4%；
2、购买3次及以上（成熟客户）的占比为17%，购买5次及以上（忠实客户）的占比为6%。
数据分析：影响复购的核心因素是商品，因此复购不适合做跨类目比较。比如食品类目和美妆类目：食品是属于“半标品”，产品的标品化程度越高，客户背叛的难度就越小，越难形成忠实用户；但是相对美妆，食品又属于易耗品，消耗周期短，购买频率高，相对容易产生重复购买，因此跨类目复购并不具有可比性。
2.1.1.3.M值：消费金额（Monetary）
M值是RFM模型中相对于R值和F值最难使用，但最具有价值的指标。大家熟知的“二八定律”（又名“帕雷托法则”）曾作出过这样的解释：公司80%的收入来自于20%的用户。
这个数据我在自己所从事的公司总都得到过验证！可能有些店铺不会那么精确，一般也会在30%客户贡献70%收入，或者40%贡献60%收入。
理论上M值和F值是一样的，都带有时间范围，指的是一段时间（通常是1年）内的消费金额，在工作中我认为对于一般店铺的类目而言，产品的价格带都是比较单一的，比如：同一品牌美妆类，价格浮动范围基本在某个特定消费群的可接受范围内，加上单一品类购买频次不高，所以对于一般店铺而言，M值对客户细分的作用相对较弱。
所以我认为用店铺的累计购买金额和平均客单价替代传统的M值能更好的体现客户消费金额的差异。
教大家一个特别简单的累积金额划分方法：将1/2的客单价作为累积消费金额的分段，比如客单价是300元，则按照150元进行累计消费金额分段，得出十个分段。
现以国内某知名化妆品店铺举例，店铺平均客单为160元，因此以80元作为间隔将累积消费金额分段，从表中可以很明显发现，累计消费160元以下用户占比为65.5%（近2/3），贡献的店铺收入比例只占31.6%（近1/3），具体如下：
2.1.2.基于RFM模型的实践应用
主要有两种方法来分析RFM模型的结果：用基于RFM模型的划分标准来进行客户细分，用基于RFM模型的客户评分来进行客户细分。
2.1.2.1.基于RFM模型进行客户细分
CRM实操时可以选择RFM模型中的1-3个指标进行客户细分，如下表所示。切记细分指标需要在自己可操控的合理范围内，并非越多越好，一旦用户细分群组过多，一来会给自己的营销方案执行带来较大的难度，而来可能会遗漏用户群或者对同个用户造成多次打扰。
最终选择多少个指标有两个参考标准：店铺的客户基数，店铺的商品和客户结构。
店铺的客户基数：在店铺客户一定的情况下选择的维度越多，细分出来每一组的用户越少。对于店铺基数不大（5万以下客户数）的店铺而言，选择1-2个维度进行细分即可。对于客户超过50万的大卖家而言可以选择2-3个指标。
店铺的商品和客户结构：如果在店铺的商品层次比较单一，客单价差异幅度不大的情况下，购买频次（F值）和消费金额（M值）高度相关的情况下，可以只选择比较容易操作的购买频次（F值）代替消费金额（M值）。对于刚刚开店还没形成客户粘性的店铺，则可以放弃购买频次（F值），直接用最后一次消费（R值）或者消费金额（M值）。
2.1.2.2.通过RFM模型评分后输出目标用户
除了直接用RFM模型对用户进行分组之外，还有一种常见的方法是利用RFM模型的三个属性对客户进行打分，通过打分确定每个用户的质量，最终筛选出自己的目标用户。
RFM模型评分主要有三个部分：
1、确定RFM三个指标的分段和每个分段的分值；
2、计算每个客户RFM三个指标的得分；
3、计算每个客户的总得分，并且根据总得分筛选出优质的客户
比如，实操的过程中一般每个指标分为3-5段，其中R值可以根据开店以来的时间和产品的回购周期来判定，F值根据现有店铺的平均购买频次，M值可参考上文客单价的分段指标。
举个例子：
确认RFM的分段和对应分段的分值之后，就可以按照用户情况对应进行打分。
这个时候可能有人会对此产生质疑，我如何验证这个给予的分值就是合理的呢？一般使用经验值或用算法模型进行验证。
2、KMeans聚类算法详解–慢点讲
2.2.1.算法原理
http://shabal.in/visuals/kmeans/3.html
/blog/visualizing-k-means-clustering/
计算步骤
选择 K 个点作为初始聚类中心
计算其他的点到中心点的距离, 进行聚类, 使用欧式距离
重新计算每个聚类的中心点, 再次聚类
直到中心点不再变化, 或者达到迭代次数
2.2.2.快速体验
2.2.2.1.数据集
IRIS数据集由Fisher在1936年整理的一个经典数据集，在统计学习和机器学习领域都经常被用作示例。
数据集内包含 3 类共 150 条记录，每类各 50 个数据，每条数据包含4个特征，都是浮点数，单位为厘米
Sepal.Length（花萼长度）
Sepal.Width（花萼宽度）
Petal.Length（花瓣长度）
Petal.Width（花瓣宽度））
目标值为鸢尾花的分类:
Iris Setosa（山鸢尾）1
Iris Versicolour（杂色鸢尾）2
Iris Virginica（维吉尼亚鸢尾）3
https://scikit-/stable/modules/generated/sklearn.datasets.load_iris.html#sklearn.datasets.load_iris
https://scikit-/stable/auto_examples/cluster/plot_cluster_iris.html#sphx-glr-auto-examples-cluster-plot-cluster-iris-py
2.2.2.2.代码演示:
import org.apache.spark.ml.clustering.{KMeans, KMeansModel}import org.apache.spark.ml.feature.{MinMaxScaler, MinMaxScalerModel}import org.apache.spark.sql.{DataFrame, SparkSession}object Iris {def main(args: Array[String]): Unit = {//创建sparksessionval spark: SparkSession = SparkSession.builder().master("local").appName("Iris").getOrCreate()//读取数据val irisDF: DataFrame = spark.read.format("libsvm").load("file:///F:\\传智播客\\传智专修学院\\第二学期\\34\\08-项目2\\用户画像2\\第三阶段\\挖掘型标签\\数据集\\iris_kmeans.txt")irisDF.show(false)//数据归一化//把数据映射到0～1范围之内处理，更加便捷快速//MinMaxScaler 把有量纲表达式变成无量纲表达式，便于不同单位或量级的指标能够进行比较和加权。//x' = (x - X_min) / (X_max - X_min)val model: MinMaxScalerModel = new MinMaxScaler().setInputCol("features").setOutputCol("featuresOut").fit(irisDF)val scalerDatas: DataFrame = model.transform(irisDF)scalerDatas.show(false)//训练val KMM: KMeansModel = new KMeans().setK(3).setMaxIter(10).setSeed(10).setFeaturesCol("featuresOut").setPredictionCol("result").fit(scalerDatas)val FD: DataFrame = KMM.transform(scalerDatas)FD.show(false)FD.groupBy("label","result").count().show(false)}}
随机种子:
public static void main(String[] args){Random rnd = new Random();rnd.setSeed(10);//用于设置种子。rnd.nextInt();// 用于产生随机数。rnd.nextInt(10); // 产生（0-9）数字。System.out.println(rnd.nextInt());}
RFM标签计算:
import cn.itcast.up.base.BaseModel2import org.apache.spark.ml.clustering.{KMeans, KMeansModel}import org.apache.spark.ml.feature.VectorAssemblerimport org.apache.spark.sql._import scala.collection.immutable/*** Author itcast* Desc 客户价值模型-RFM:* Rencency:最近一次消费,最后一次订单距今时间* Frequency:消费频率,订单总数量* Monetary:消费金额,订单总金额*/object RFMModel extends BaseModel2{def main(args: Array[String]): Unit = {exec()}/*** 获取标签id(即模型id,该方法应该在编写不同模型时进行实现)* @return*/override def getTagID(): Int = 37/*** 开始计算* inType=HBase##zkHosts=192.168.10.20##zkPort=2181##* hbaseTable=tbl_orders##family=detail##selectFields=memberId,orderSn,orderAmount,finishTime* @param fiveDF MySQL中的5级规则 id,rule* @param hbaseDF 根据selectFields查询出来的HBase中的数据* @return userid,tagIds*/override def compute(fiveDF: DataFrame, hbaseDF: DataFrame): DataFrame = {//fiveDF.show()//fiveDF.printSchema()//hbaseDF.show(10)//hbaseDF.printSchema()/*+---+----+| id|rule|+---+----+| 38| 1|| 39| 2|| 40| 3|| 41| 4|| 42| 5|| 43| 6|| 44| 7|+---+----+root|-- id: long (nullable = false)|-- rule: string (nullable = true)+---------+-------------------+-----------+----------+| memberId| orderSn|orderAmount|finishTime|+---------+-------------------+-----------+----------+| 13823431| ts_792756751164275| 2479.45|1564415022|| 4035167| D14090106121770839| 2449.00|1565687310|| 4035291| D14090112394810659| 1099.42|1564681801|| 4035041| fx_787749561729045| 1999.00|1565799378|| 13823285| D1409214435903| 2488.00|1565062072|| 4034219| D1409215620305| 3449.00|1563601306||138230939|top_810791455519102| 1649.00|1565509622|| 4035083| D1409211884409| 7.00|1565731851||138230935| D1409212313538| 1299.00|1565382991|| 13823231| D1409212378713|499.00|1565677650|+---------+-------------------+-----------+----------+only showing top 10 rowsroot|-- memberId: string (nullable = true)|-- orderSn: string (nullable = true)|-- orderAmount: string (nullable = true)|-- finishTime: string (nullable = true)*/import spark.implicits._import scala.collection.JavaConversions._import org.apache.spark.sql.functions._//0.定义常量字符串,避免后续拼写错误val recencyStr = "recency"val frequencyStr = "frequency"val monetaryStr = "monetary"val featureStr = "feature"val predictStr = "predict"//1.按用户id进行聚合获取客户RFM//客户价值模型-RFM://Rencency:最近一次消费,最后一次订单距今时间//Frequency:消费频率,订单总数量//Monetary:消费金额,订单总金额///liam08/article/details/79663018val recencyAggColumn: Column = functions.datediff(date_sub(current_timestamp(),60), from_unixtime(max('finishTime))) as recencyStrval frequencyAggColumn: Column = functions.count('orderSn) as frequencyStrval monetaryAggColumn: Column = functions.sum('orderAmount) as monetaryStrval RFMResult = hbaseDF.groupBy('memberId).agg(recencyAggColumn, frequencyAggColumn, monetaryAggColumn)//RFMResult.show(10)//RFMResult.printSchema()//RFMResult.groupBy(recencyStr).count().show()/*+---------+-------+---------+------------------+| memberId|recency|frequency|monetary|+---------+-------+---------+------------------+| 13822725|12|116| 179298.34|| 13823083|12|132| 233524.17||138230919|12|125|240061.56999999998|| 13823681|12|108|169746.1|| 4033473|12|142| 251930.92|| 13822841|12|113| 205931.91|| 13823153|12|133| 250698.57|| 13823431|12|122| 180858.22|| 4033348|12|145|240173.78999999998|| 4033483|12|110|157811.09999999998|+---------+-------+---------+------------------+only showing top 10 rowsroot|-- memberId: string (nullable = true)|-- recency: integer (nullable = true)|-- frequency: long (nullable = false)|-- monetary: double (nullable = true)*///2.为RFM打分//R: 1-3天=5分，4-6天=4分，7-9天=3分，10-15天=2分，大于16天=1分//F: ≥200=5分，150-199=4分，100-149=3分，50-99=2分，1-49=1分//M: ≥20w=5分，10-19w=4分，5-9w=3分，1-4w=2分，<1w=1分val recencyScore: Column = functions.when((col(recencyStr) >= 1) && (col(recencyStr) <= 3), 5).when((col(recencyStr) >= 4) && (col(recencyStr) <= 6), 4).when((col(recencyStr) >= 7) && (col(recencyStr) <= 9), 3).when((col(recencyStr) >= 10) && (col(recencyStr) <= 15), 2).when(col(recencyStr) >= 16, 1).as(recencyStr)val frequencyScore: Column = functions.when(col(frequencyStr) >= 200, 5).when((col(frequencyStr) >= 150) && (col(frequencyStr) <= 199), 4).when((col(frequencyStr) >= 100) && (col(frequencyStr) <= 149), 3).when((col(frequencyStr) >= 50) && (col(frequencyStr) <= 99), 2).when((col(frequencyStr) >= 1) && (col(frequencyStr) <= 49), 1).as(frequencyStr)val monetaryScore: Column = functions.when(col(monetaryStr) >= 200000, 5).when(col(monetaryStr).between(100000, 199999), 4).when(col(monetaryStr).between(50000, 99999), 3).when(col(monetaryStr).between(10000, 49999), 2).when(col(monetaryStr) <= 9999, 1).as(monetaryStr)val RFMScoreResult: DataFrame = RFMResult.select('memberId, recencyScore, frequencyScore, monetaryScore)//RFMScoreResult.show(10)/*+---------+-------+---------+--------+| memberId|recency|frequency|monetary|+---------+-------+---------+--------+| 13822725|2| 3| 4|| 13823083|2| 3| 5||138230919|2| 3| 5|| 13823681|2| 3| 4|| 4033473|2| 3| 5|| 13822841|2| 3| 5|| 13823153|2| 3| 5|| 13823431|2| 3| 4|| 4033348|2| 3| 5|| 4033483|2| 3| 4|| 4033575|2| 3| 5|| 4034191|2| 3| 5|+---------+-------+---------+--------+only showing top 200 rows*///3.聚类//为方便后续模型进行特征输入，需要部分列的数据转换为特征向量，并统一命名，VectorAssembler类就可以完成这一任务。//VectorAssembler是一个transformer，将多列数据转化为单列的向量列val vectorDF: DataFrame = new VectorAssembler().setInputCols(Array(recencyStr, frequencyStr, monetaryStr)).setOutputCol(featureStr).transform(RFMScoreResult)//vectorDF.show(10)/*+---------+-------+---------+--------+-------------+| memberId|recency|frequency|monetary|feature|+---------+-------+---------+--------+-------------+| 13822725|2| 3| 4|[2.0,3.0,4.0]|| 13823083|2| 3| 5|[2.0,3.0,5.0]||138230919|2| 3| 5|[2.0,3.0,5.0]|| 13823681|2| 3| 4|[2.0,3.0,4.0]|| 4033473|2| 3| 5|[2.0,3.0,5.0]|| 13822841|2| 3| 5|[2.0,3.0,5.0]|| 13823153|2| 3| 5|[2.0,3.0,5.0]|| 13823431|2| 3| 4|[2.0,3.0,4.0]|| 4033348|2| 3| 5|[2.0,3.0,5.0]|| 4033483|2| 3| 4|[2.0,3.0,4.0]|+---------+-------+---------+--------+-------------+*/val kMeans: KMeans = new KMeans().setK(7).setSeed(10) //可重复的随机种子.setMaxIter(2)//最大迭代次数.setFeaturesCol(featureStr) //特征列.setPredictionCol(predictStr)//预测结果列//4.训练模型val model: KMeansModel = kMeans.fit(vectorDF)model.save("/model/RFMModel/")//val model = KMeansModel.load("/model/RFMModel/")//5.预测val result: DataFrame = model.transform(vectorDF)//result.show(10)/*+---------+-------+---------+--------+-------------+-------+| memberId|recency|frequency|monetary|feature|predict|+---------+-------+---------+--------+-------------+-------+| 13822725|2| 3| 4|[2.0,3.0,4.0]|1|| 13823083|2| 3| 5|[2.0,3.0,5.0]|0||138230919|2| 3| 5|[2.0,3.0,5.0]|0|| 13823681|2| 3| 4|[2.0,3.0,4.0]|1|| 4033473|2| 3| 5|[2.0,3.0,5.0]|0|| 13822841|2| 3| 5|[2.0,3.0,5.0]|0|| 13823153|2| 3| 5|[2.0,3.0,5.0]|0|| 13823431|2| 3| 4|[2.0,3.0,4.0]|1|| 4033348|2| 3| 5|[2.0,3.0,5.0]|0|| 4033483|2| 3| 4|[2.0,3.0,4.0]|1|+---------+-------+---------+--------+-------------+-------+*///6.测试时看下聚类效果val ds: Dataset[Row] = result.groupBy(predictStr).agg(max(col(recencyStr) + col(frequencyStr) + col(monetaryStr)), min(col(recencyStr) + col(frequencyStr) + col(monetaryStr))).sort(col(predictStr).asc)//ds.show()/*+-------+---------------------------------------+---------------------------------------+|predict|max(((recency + frequency) + monetary))|min(((recency + frequency) + monetary))|+-------+---------------------------------------+---------------------------------------+|0| 10| 9||1| 9| 8||2| 4| 3||3| 6| 6||4| 7| 7||5| 12| 11||6| 5| 5|+-------+---------------------------------------+---------------------------------------+*///问题: 每一个簇的ID是无序的,但是我们将分类簇和rule进行对应的时候,需要有序//7.按质心排序,质心大,该类用户价值大//[(质心id, 质心值)]val center: immutable.IndexedSeq[(Int, Double)] = for(i <- model.clusterCenters.indices) yield (i, model.clusterCenters(i).toArray.sum)val sortedCenter: immutable.IndexedSeq[(Int, Double)] = center.sortBy(_._2).reverse//sortedCenter.foreach(println)/*(5,11.038461538461538)(0,10.0)(1,9.0)(6,8.0)(3,5.333333333333333)(2,4.0)(4,3.0)*///[(质心id, rule值)]val centerIdAndRule: immutable.IndexedSeq[(Int, Int)] = for(i <- sortedCenter.indices) yield (sortedCenter(i)._1, i+1)//centerIdAndRule.foreach(println)/*(5,1)(0,2)(1,3)(6,4)(3,5)(2,6)(4,7)*/val predictRuleDF: DataFrame = centerIdAndRule.toDF(predictStr,"rule")//predictRuleDF.show()/*+-------+----+|predict|rule|+-------+----+|5| 1||0| 2||1| 3||6| 4||3| 5||2| 6||4| 7|+-------+----+*///8.将rule和5级规则进行匹配val ruleTagDF: DataFrame = fiveDF.as[(Long,String)].map(t=>(t._2,t._1)).toDF("rule","tag")ruleTagDF.show()/*+----+---+|rule|tag|+----+---+| 1| 38|| 2| 39|| 3| 40|| 4| 41|| 5| 42|| 6| 43|| 7| 44|+----+---+*/val predictRuleTagDF: DataFrame = predictRuleDF.join(ruleTagDF,"rule")predictRuleTagDF.show()/*+----+-------+---+|rule|predict|tag|+----+-------+---+| 1|5| 38|| 2|0| 39|| 3|1| 40|| 4|6| 41|| 5|3| 42|| 6|2| 43|| 7|4| 44|+----+-------+---+*///Map[predict, tag]val map: Map[Long, Long] = predictRuleTagDF.as[(String,Long,Long)].map(t=>(t._2,t._3)).collect().toMapval predict2tag = functions.udf((predict:Long)=>{val tag = map(predict)tag})//8.predict->tagval newDF: DataFrame = result.select($"memberId".as("userId"),predict2tag($"predict").as("tagIds"))newDF.show(10)/*+---------+------+| userId|tagIds|+---------+------+| 13822725| 40|| 13823083| 39||138230919| 39|| 13823681| 40|| 4033473| 39|| 13822841| 39|| 13823153| 39|| 13823431| 40|| 4033348| 39|| 4033483| 40|+---------+------+*/newDF}}
