一、环境准备与依赖安装
本指南将基于Python语言构建一套轻量级的私域流量复购触发系统。该系统通过RFM模型（最近一次消费、消费频率、消费金额）自动筛选高价值流失用户，并通过企业微信机器人发送定向复购提醒。请确保你的服务器或本地电脑已安装Python 3.8及以上版本。
我们需要安装处理数据的核心库Pandas以及发起HTTP请求的Requests库。打开终端或命令行工具，直接执行以下命令进行安装：
```bash
pip install pandas numpy requests
```
安装完成后，创建一个项目目录，例如private_domain_system，并在其中新建一个名为config.py的文件。我们将把所有的配置项集中管理，避免在代码中硬编码，方便后续维护。
二、构建模拟私域用户数据库
为了演示实操效果，我们首先生成一份模拟的用户订单数据。在实际生产环境中，你可以直接从MySQL或PostgreSQL中导出CSV文件，格式需保持一致。新建文件init_data.py，并写入以下代码：
```python
import pandas as pd
import random
from datetime import datetime, timedelta
生成模拟数据的函数
def generate_mock_data(num_users=100):
data = []
base_date = datetime.now()
for i in range(1, num_users + 1):
user_id = f"USER_{i:04d}"
模拟每个用户有1-10笔订单
order_count = random.randint(1, 10)
for _ in range(order_count):
随机生成过去90天内的订单
days_ago = random.randint(0, 90)
order_date = (base_date - timedelta(days=days_ago)).strftime("%Y-%m-%d")
随机生成金额 50-500元
amount = round(random.uniform(50, 500), 2)
data.append({
"user_id": user_id,
"order_date": order_date,
"amount": amount
})
df = pd.DataFrame(data)
df.to_csv("user_orders.csv", index=False)
print("模拟数据已生成至 user_orders.csv")
if __name__ == "__main__":
generate_mock_data()
```
运行该脚本：python init_data.py。此时目录下会生成user_orders.csv文件，包含user_id（用户ID）、order_date（订单日期）、amount（订单金额）三列，这是我们进行复购分析的基础数据源。
三、RFM模型核心算法实现
RFM模型是衡量用户价值和用户创利能力的重要工具。我们需要编写代码计算每个用户的R、F、M值，并进行评分。新建文件rfm_analysis.py，首先实现数据处理与评分逻辑。
核心逻辑如下：
- R (Recency)： 最近一次消费时间距今的天数。天数越少，分数越高。
- F (Frequency)： 消费频率。订单数越多，分数越高。
- M (Monetary)： 消费金额。总金额越高，分数越高。
我们将采用5分制进行打分。在rfm_analysis.py中添加以下代码：
```python
import pandas as pd
import numpy as np
from datetime import datetime
def calculate_rfm_scores(file_path):
读取数据
df = pd.read_csv(file_path)
df['order_date'] = pd.to_datetime(df['order_date'])
计算R, F, M基础值
current_date = datetime.now()
rfm = df.groupby('user_id').agg({
'order_date': lambda x: (current_date - x.max()).days, R: 距今天数
'user_id': 'count', F: 订单数
'amount': 'sum' M: 总金额
}).rename(columns={
'order_date': 'recency',
'user_id': 'frequency',
'amount': 'monetary'
})
定义评分函数 (1-5分)
def r_score(x):
if x <= 7: return 5
elif x <= 15: return 4
elif x <= 30: return 3
elif x <= 60: return 2
else: return 1
def f_score(x):
if x >= 10: return 5&lgt;
elif x >= 6: return 4
elif x >= 3: return 3
elif x >= 2: return 2
else: return 1
def m_score(x):
if x >= 1000: return 5
elif x >= 500: return 4
elif x >= 200: return 3
elif x >= 100: return 2
else: return 1
应用评分
rfm['R_Score'] = rfm['recency'].apply(r_score)
rfm['F_Score'] = rfm['frequency'].apply(f_score)
rfm['M_Score'] = rfm['monetary'].apply(m_score)
计算综合RFM值 (字符串拼接，便于分层)
rfm['RFM_Segment'] = rfm['R_Score'].astype(str) + rfm['F_Score'].astype(str) + rfm['M_Score'].astype(str)
return rfm
if __name__ == "__main__":
rfm_data = calculate_rfm_scores("user_orders.csv")
print(rfm_data.head())
```
四、复购触发规则与消息推送
计算完RFM分数后，我们需要定义“重要流失客户”的筛选规则，这是提升复购率的关键人群。通常，我们将R分低（很久没买）、但F和M分高（以前买得多、买得贵）的用户作为重点召回对象。
我们将集成企业微信群机器人的Webhook接口进行消息推送。你需要获取企业微信机器人的Webhook地址。在企业微信群聊中，添加群机器人，复制生成的Webhook URL，并将其填入config.py：
```python
config.py
WECHAT_WEBHOOK_URL = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY_HERE"
```
接下来，在rfm_analysis.py中增加筛选和发送消息的函数：
```python
import requests
import config
def send_wechat_message(user_id, rfm_data):
"""
发送复购提醒消息到企业微信
"""
构建消息内容
content = f"""
【高价值流失客户预警】
用户ID: {user_id}
最近消费: {rfm_data['recency']}天前
消费频次: {rfm_data['frequency']}次
总消费金额: {rfm_data['monetary']}元
建议：立即发送定向优惠券进行召回！
"""
data = {
"msgtype": "text",
"text": {
"content": content
}
}
try:
response = requests.post(config.WECHAT_WEBHOOK_URL, json=data)
if response.json()['errcode'] == 0:
print(f"用户 {user_id} 预警消息发送成功")
else:
print(f"用户 {user_id} 消息发送失败: {response.text}")
except Exception as e:
print(f"发送异常: {e}")
def trigger_repurchase_logic(rfm_df):
"""
执行复购触发逻辑
规则：R_Score <= 2 (超过30天未购) 且 M_Score >= 4 (高价值)
"""
筛选目标用户
target_users = rfm_df[(rfm_df['R_Score'] <= 2) & (rfm_df['M_Score'] >= 4)]
if target_users.empty:
print("当前没有符合条件的高价值流失用户。")
return
print(f"检测到 {len(target_users)} 名高价值流失用户，开始触发召回...")
遍历并发送消息
for user_id, row in target_users.iterrows():
send_wechat_message(user_id, row)
```
五、全流程自动化部署
我们需要将上述步骤串联起来，并设置定时任务。将rfm_analysis.py的main部分修改如下：
```python
if __name__ == "__main__":
1. 计算RFM
print("开始计算RFM模型...")
rfm_result = calculate_rfm_scores("user_orders.csv")
2. 执行触发逻辑
print("开始执行复购触发逻辑...")
trigger_repurchase_logic(rfm_result)
print("复购扫描任务完成。")
```
现在，你可以通过Linux的Crontab或Windows的任务计划程序来每天自动运行此脚本。假设你的脚本位于/home/user/private_domain_system/rfm_analysis.py，编辑crontab：
```bash
crontab -e
```
添加一行，设定为每天早上9点运行：
```bash
0 9 /usr/bin/python3 /home/user/private_domain_system/rfm_analysis.py >> /home/user/private_domain_system/run.log 2>&1
```
这条命令的含义是：每天09:00执行Python脚本，并将标准输出和错误输出重定向到run.log文件中，方便排查故障。确保user_orders.csv中的数据每天通过ETL作业更新，或者你的脚本直接连接生产数据库读取实时数据，这样整个闭环就打通了。
