零门槛实操：利用RFM模型提升自然流量复购率

技术架构与前置准备

本文将基于 Python 和 MySQL 构建一套自动化的 RFM（Recency, Frequency, Monetary）模型系统。该系统通过分析用户的自然访问行为和订单数据，自动计算用户价值分层，并针对高复购潜力用户执行自动化触达策略。你需要准备以下环境：

1. 数据库环境准备

确保已安装 MySQL 5.7 或更高版本。我们将创建三张核心表：用户行为日志表（记录自然流量）、订单表（记录消费）、用户标签表（存储计算结果）。请在 MySQL 客户端执行以下 SQL：

```sql CREATE DATABASE IF NOT EXISTS traffic_analysis; USE traffic_analysis; -- 用户行为日志表 CREATE TABLE `user_logs` ( `id` INT AUTO_INCREMENT PRIMARY KEY, `user_id` INT NOT NULL, `source_type` VARCHAR(20) COMMENT '流量来源：organic为自然流量', `visit_time` DATETIME DEFAULT CURRENT_TIMESTAMP, INDEX (`user_id`, `visit_time`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; -- 订单表 CREATE TABLE `orders` ( `id` INT AUTO_INCREMENT PRIMARY KEY, `user_id` INT NOT NULL, `order_amount` DECIMAL(10, 2) NOT NULL, `pay_time` DATETIME DEFAULT CURRENT_TIMESTAMP, INDEX (`user_id`, `pay_time`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; -- 用户RFM标签表 CREATE TABLE `user_rfm_tags` ( `user_id` INT PRIMARY KEY, `r_score` INT COMMENT '最近一次消费得分', `f_score` INT COMMENT '消费频率得分', `m_score` INT COMMENT '消费金额得分', `user_level` VARCHAR(20) COMMENT '用户分层等级', `update_time` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; ```

Python 环境配置与依赖安装

在本地创建项目目录，并安装必要的 Python 库。我们需要 pymysql 用于数据库连接，pandas 用于数据处理，datetime 用于时间计算。打开终端执行：

```bash pip install pymysql pandas numpy ```

核心逻辑：RFM 模型数据计算

创建名为 rfm_calculator.py 的文件。该脚本将执行以下核心逻辑：从数据库提取自然流量用户的订单数据，计算 R、F、M 指标，进行 5 分制打分，并回写数据库。以下是完整的可直接运行的代码：

```python import pymysql import pandas as pd import numpy as np from datetime import datetime, timedelta 数据库配置（请根据实际情况修改） DB_CONFIG = { 'host': 'localhost', 'user': 'root', 'password': 'your_password', 'database': 'traffic_analysis', 'charset': 'utf8mb4' } def get_data(): """获取自然流量用户在过去90天的订单数据""" conn = pymysql.connect(DB_CONFIG) try: 筛选来源为 organic 的用户，并关联其订单数据 sql = """ SELECT DISTINCT o.user_id, o.pay_time, o.order_amount FROM orders o INNER JOIN user_logs l ON o.user_id = l.user_id WHERE l.source_type = 'organic' AND o.pay_time >= DATE_SUB(NOW(), INTERVAL 90 DAY) ORDER BY o.user_id, o.pay_time DESC """ df = pd.read_sql(sql, conn) return df finally: conn.close() def calculate_rfm(df): """计算R、F、M指标并进行打分""" if df.empty: return pd.DataFrame() 计算基础指标 R: 最近一次消费距离今天的天数 F: 消费次数 M: 消费总金额 snapshot_date = df['pay_time'].max() + timedelta(days=1) rfm = df.groupby('user_id').agg({ 'pay_time': lambda x: (snapshot_date - x.max()).days, R 'order_amount': ['count', 'sum'] F and M }).reset_index() rfm.columns = ['user_id', 'recency', 'frequency', 'monetary'] 打分逻辑 (5分制) R越小越好，F和M越大越好 rfm['r_score'] = pd.cut(rfm['recency'], bins=[0, 10, 30, 60, 90, 999], labels=[5, 4, 3, 2, 1], right=False).astype(int) rfm['f_score'] = pd.cut(rfm['frequency'], bins=[0, 1, 2, 3, 5, 100], labels=[1, 2, 3, 4, 5], right=False).astype(int) M金额打分根据分位数动态计算 m_bins = rfm['monetary'].quantile([0, 0.2, 0.4, 0.6, 0.8, 1]).tolist() rfm['m_score'] = pd.cut(rfm['monetary'], bins=m_bins, labels=[1, 2, 3, 4, 5], include_include_lowest=True).astype(int) 用户分层定义 def label_user(row): if row['r_score'] >= 4 and row['f_score'] >= 4: return '重要价值用户' elif row['r_score'] >= 4 and row['f_score'] <= 2: return '重要发展用户' elif row['r_score'] <= 2 and row['f_score'] >= 4: return '重要挽留用户' elif row['r_score'] <= 2 and row['f_score'] <= 2: return '流失用户' else: return '一般保持用户' rfm['user_level'] = rfm.apply(label_user, axis=1) return rfm[['user_id', 'r_score', 'f_score', 'm_score', 'user_level']] def save_to_db(rfm_df): """将计算结果回写数据库""" conn = pymysql.connect(DB_CONFIG) cursor = conn.cursor() 使用 REPLACE INTO 实现 Upsert 操作 sql = """ REPLACE INTO user_rfm_tags (user_id, r_score, f_score, m_score, user_level) VALUES (%s, %s, %s, %s, %s) """ data_to_insert = [] for _, row in rfm_df.iterrows(): data_to_insert.append((row['user_id'], row['r_score'], row['f_score'], row['m_score'], row['user_level'])) try: cursor.executemany(sql, data_to_insert) conn.commit() print(f"成功更新 {len(data_to_insert)} 条用户标签数据") except Exception as e: conn.rollback() print(f"数据写入失败: {e}") finally: cursor.close() conn.close() if __name__ == "__main__": 1. 提取数据 raw_data = get_data() 2. 计算RFM result_df = calculate_rfm(raw_data) 3. 保存结果 if not result_df.empty: save_to_db(result_df) else: print("未查询到自然流量用户数据") ```

自动化复购召回策略实现

有了用户分层后，我们需要针对“重要发展用户”（R高但F低，刚买过但买得少）和“重要挽留用户”（F高但R低，以前常买但最近没来）实施召回策略。创建 recall_strategy.py：

```python import pymysql import requests DB_CONFIG = { 'host': 'localhost', 'user': 'root', 'password': 'your_password', 'database': 'traffic_analysis', 'charset': 'utf8mb4' } def send_coupon(user_id, coupon_type): """ 模拟发送优惠券接口 在实际场景中，这里替换为你公司的CRM或短信发送API """ print(f"正在向用户 {user_id} 发送 {coupon_type}...") 示例：调用内部API url = "http://api.internal.com/send-coupon" payload = {"user_id": user_id, "template_id": coupon_type} requests.post(url, json=payload) return True def execute_recall_strategy(): conn = pymysql.connect(DB_CONFIG) cursor = conn.cursor() 获取需要召回的用户 1. 重要发展用户：推送凑单优惠，提升F值 2. 重要挽留用户：推送大额回归券，提升R值 sql = """ SELECT user_id, user_level FROM user_rfm_tags WHERE user_level IN ('重要发展用户', '重要挽留用户') """ try: cursor.execute(sql) users = cursor.fetchall() for user in users: user_id, level = user if level == '重要发展用户': 策略：发送满减券，刺激二次消费 send_coupon(user_id, "FULL_REDUCTION_COUPON_50") elif level == '重要挽留用户': 策略：发送无门槛回归券 send_coupon(user_id, "NO_THRESHOLD_RETURN_COUPON_20") except Exception as e: print(f"策略执行失败: {e}") finally: cursor.close() conn.close() if __name__ == "__main__": execute_recall_strategy() ```

定时任务自动化部署

为了让系统每天自动运行，无需人工干预，我们需要配置操作系统的定时任务。

Linux/Mac 系统配置 (Crontab)

在终端输入 crontab -e，添加以下两行配置。这将设置每天凌晨 1:00 计算模型，1:30 执行召回策略：

```bash 0 1 /usr/bin/python3 /path/to/your/project/rfm_calculator.py >> /var/log/rfm.log 2>&1 30 1 /usr/bin/python3 /path/to/your/project/recall_strategy.py >> /var/log/recall.log 2>&1 ```

Windows 系统配置 (任务计划程序)

1. 打开“任务计划程序”，点击“创建基本任务”。
2. 名称输入“自然流量复购模型”，触发器选择“每天”。
3. 操作选择“启动程序”，程序/脚本填入 python.exe 的绝对路径。
4. 参数添加填入脚本路径，例如：C:\Users\Admin\project\rfm_calculator.py。
5. 起始于填入项目所在目录，例如：C:\Users\Admin\project。

实操验证与数据监控

完成部署后，不要只看代码，必须验证数据准确性。你可以手动往数据库插入一条测试数据，然后运行脚本查看结果：

```sql -- 插入一个自然流量访问记录 INSERT INTO user_logs (user_id, source_type, visit_time) VALUES (9999, 'organic', NOW()); -- 插入该用户的一笔订单 INSERT INTO orders (user_id, order_amount, pay_time) VALUES (9999, 100.00, NOW()); ```

运行 python rfm_calculator.py，然后查询 SELECT FROM user_rfm_tags WHERE user_id = 9999;。如果看到该用户的 r_score 为 5 且 user_level 为“重要发展用户”，说明整个链路已跑通。此时，运行 python recall_strategy.py，控制台应打印出发送优惠券的日志，证明技术闭环已形成。