基于RFM模型的会员体系用户分层实操指南

一、RFM模型分层规则定义

在开始编码之前，必须明确定义分层的数学逻辑。RFM模型是用户分层中最经典且可落地的方案。我们需要根据用户的最近一次消费时间、消费频率和消费金额三个维度，将用户划分为不同的价值层级。

为了确保实操的严谨性，我们设定以下评分标准：

• R (Recency)：最近一次消费时间距今天数。天数越少，得分越高（5分为最近，1分为最久）。
• F (Frequency)：近90天内的消费次数。次数越多，得分越高。
• M (Monetary)：近90天内的消费总金额。金额越高，得分越高。

基于R、F、M三个维度的得分，我们将用户定义为以下5种核心类型。这种分类方式能直接指导后续的营销推送。

用户类型	判定规则（基于平均值的比较）	运营策略
重要价值会员	R高 & F高 & M高	优先提供服务，无门槛高权益
重要发展会员	R高 & F低 & M高	引导复购，提升频率
重要保持会员	R低 & F高 & M高	主动挽留，防止流失
一般发展会员	R高 & F低 & M低	新人券，尝试转化
流失/低价值会员	R低 & F低 & M低	暂不投入资源，仅在大促触达

二、数据库表结构设计

为了保证分层系统的独立性，我们不应直接修改业务主表，而是建立专门的宽表或结果表。假设你使用的是MySQL数据库，请执行以下SQL语句初始化环境。这些脚本包含了建表、插入模拟数据以及最终结果表的创建。

1. 创建用户基础信息表

```sql CREATE TABLE IF NOT EXISTS `t_user_info` ( `id` int(11) NOT NULL AUTO_INCREMENT, `user_id` bigint(20) NOT NULL COMMENT '用户ID', `register_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '注册时间', PRIMARY KEY (`id`), UNIQUE KEY `uk_user_id` (`user_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='用户基础表'; ```

2. 创建订单流水表

```sql CREATE TABLE IF NOT EXISTS `t_order_info` ( `id` int(11) NOT NULL AUTO_INCREMENT, `order_id` bigint(20) NOT NULL COMMENT '订单号', `user_id` bigint(20) NOT NULL COMMENT '用户ID', `amount` decimal(10, 2) NOT NULL COMMENT '订单金额', `pay_time` datetime NOT NULL COMMENT '支付时间', PRIMARY KEY (`id`), KEY `idx_user_pay` (`user_id`, `pay_time`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='订单流水表'; ```

3. 创建用户分层结果表

此表用于存储计算后的用户层级，业务系统直接读取此表即可获取用户的当前等级。

```sql CREATE TABLE IF NOT EXISTS `t_user_level_result` ( `id` int(11) NOT NULL AUTO_INCREMENT, `user_id` bigint(20) NOT NULL COMMENT '用户ID', `r_score` int(11) DEFAULT NULL COMMENT 'R得分', `f_score` int(11) DEFAULT NULL COMMENT 'F得分', `m_score` int(11) DEFAULT NULL COMMENT 'M得分', `user_level` varchar(32) DEFAULT NULL COMMENT '用户层级', `update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (`id`), UNIQUE KEY `uk_user_id` (`user_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='用户分层结果表'; ```

三、Python自动化打标脚本实现

虽然SQL可以实现计算，但涉及复杂的逻辑判断和分位数计算时，Python的Pandas库效率更高且更易维护。以下是一个完整的、可直接运行的Python脚本。你需要先安装依赖库：

pip install pandas pymysql sqlalchemy

将以下代码保存为 user_segmentation.py。请务必修改代码顶部的数据库连接配置。

```python import pandas as pd import pymysql from sqlalchemy import create_engine from datetime import datetime, timedelta ================= 配置区域 ================= 请替换为你实际的数据库连接信息 DB_CONFIG = { 'host': '127.0.0.1', 'port': 3306, 'user': 'root', 'password': 'your_password', 'database': 'your_db_name', 'charset': 'utf8mb4' } 创建数据库连接引擎 engine = create_engine(f"mysql+pymysql://{DB_CONFIG['user']}:{DB_CONFIG['password']}@{DB_CONFIG['host']}:{DB_CONFIG['port']}/{DB_CONFIG['database']}") def get_data(): """从数据库提取原始数据，计算R、F、M原始值""" 计算截止时间：例如取最近90天的数据 end_date = datetime.now() start_date = end_date - timedelta(days=90) sql = f""" SELECT user_id, DATEDIFF(NOW(), MAX(pay_time)) as recency_days, -- R：最近一次消费距今天数 COUNT(order_id) as frequency_count, -- F：消费次数 SUM(amount) as monetary_sum -- M：消费总额 FROM t_order_info WHERE pay_time >= '{start_date.strftime('%Y-%m-%d')}' GROUP BY user_id """ df = pd.read_sql(sql, engine) return df def score_rfm(df): """计算R、F、M的得分 (1-5分)""" R得分：天数越小越好，使用qcut将数据分为5份，labels=[5,4,3,2,1]表示天数越短得分越高 duplicates='drop' 防止数据量少导致边界值报错 try: df['r_score'] = pd.qcut(df['recency_days'], 5, labels=[5, 4, 3, 2, 1], duplicates='drop') except ValueError: df['r_score'] = 3 如果数据量不足以分箱，给平均分 F得分：次数越多越好 try: df['f_score'] = pd.qcut(df['frequency_count'].rank(method='first'), 5, labels=[1, 2, 3, 4, 5], duplicates='drop') except ValueError: df['f_score'] = 3 M得分：金额越高越好 try: df['m_score'] = pd.qcut(df['monetary_sum'].rank(method='first'), 5, labels=[1, 2, 3, 4, 5], duplicates='drop') except ValueError: df['m_score'] = 3 将得分转为整数类型 df['r_score'] = df['r_score'].astype(int) df['f_score'] = df['f_score'].astype(int) df['m_score'] = df['m_score'].astype(int) return df def define_level(row): """根据R、F、M得分定义用户层级""" 计算平均分作为阈值 r_avg = 3 f_avg = 3 m_avg = 3 r = row['r_score'] f = row['f_score'] m = row['m_score'] if r >= r_avg and f >= f_avg and m >= m_avg: return "重要价值会员" elif r >= r_avg and f < f_avg and m >= m_avg: return "重要发展会员" elif r < r_avg and f >= f_avg and m >= m_avg: return "重要保持会员" elif r >= r_avg and f < f_avg and m < m_avg: return "一般发展会员" else: return "流失/低价值会员" def save_to_db(df): """将结果保存回数据库""" 选择需要的列 result_df = df[['user_id', 'r_score', 'f_score', 'm_score', 'user_level']] 使用 REPLACE INTO 策略，如果user_id存在则更新，不存在则插入 先清空旧数据（可选，或者根据业务策略保留历史） pd.read_sql("TRUNCATE TABLE t_user_level_result", engine) result_df.to_sql('t_user_level_result', engine, if_exists='replace', index=False, method='multi') print(f"成功写入/更新 {len(result_df)} 条用户分层数据") def main(): print("开始执行用户分层任务...") 1. 获取数据 df = get_data() if df.empty: print("未查询到订单数据，任务结束。") return 2. 计算得分 df = score_rfm(df) 3. 定义层级 df['user_level'] = df.apply(define_level, axis=1) 4. 保存结果 save_to_db(df) print("任务执行完成。") if __name__ == '__main__': main() ```

四、业务系统集成与查询

脚本运行后，数据已经存储在 t_user_level_result 表中。业务系统（如Java、Go或PHP后端）无需重新计算逻辑，只需通过用户ID进行关联查询即可。

查询示例SQL：

```sql SELECT u.user_id, u.register_time, r.user_level, r.r_score, r.f_score, r.m_score, CASE r.user_level WHEN '重要价值会员' THEN '发放8折优惠券' WHEN '重要保持会员' THEN '发送召回短信' ELSE '发送常规推送' END as suggested_action FROM t_user_info u LEFT JOIN t_user_level_result r ON u.user_id = r.user_id WHERE u.user_id = 100001; ```

五、定时任务调度配置

用户分层不是一次性任务，需要每天或每周更新。在生产环境中，我们推荐使用 Linux 的 Crontab 来调度上述 Python 脚本。

1. 打开终端，输入以下命令编辑 crontab：

crontab -e

2. 添加一行任务配置，设置为每天凌晨2点30分执行一次：

```bash 30 2 /usr/bin/python3 /your/path/to/user_segmentation.py >> /var/log/user_segmentation.log 2>&1 ```

注意事项：

• 请将 /usr/bin/python3 替换为你服务器上 Python 的实际路径（可通过 which python3 查看）。
• 请将 /your/path/to/ 替换为脚本实际存放的绝对路径。
• >> /var/log/user_segmentation.log 2>&1 表示将脚本输出的日志打印到文件中，方便排查错误。