本文目录导读：
我来为你详细介绍用户分群逻辑脚本的编写方法,包含多种实现方式和最佳实践。
基础分群脚本结构
Python版本
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
class UserSegmentation:
def __init__(self, user_data):
"""
初始化用户分群器
Args:
user_data: DataFrame，包含用户行为数据
"""
self.data = user_data
def segment_by_rfm(self, recency_days=30, frequency_count=5, monetary_value=1000):
"""
RFM模型分群
Args:
recency_days: 最近一次购买天数阈值
frequency_count: 购买频率阈值
monetary_value: 消费金额阈值
"""
# 计算RFM指标
today = datetime.now()
rfm = self.data.groupby('user_id').agg({
'order_date': lambda x: (today - x.max()).days, # 近度
'order_id': 'count', # 频率
'amount': 'sum' # 金额
}).rename(columns={
'order_date': 'recency',
'order_id': 'frequency',
'amount': 'monetary'
})
# 分群逻辑
conditions = [
(rfm['recency'] <= recency_days) &
(rfm['frequency'] >= frequency_count) &
(rfm['monetary'] >= monetary_value),
(rfm['recency'] <= recency_days) &
(rfm['frequency'] >= frequency_count),
(rfm['monetary'] >= monetary_value * 2),
]
choices = ['高价值用户', '活跃用户', '大额用户']
rfm['segment'] = np.select(conditions, choices, default='普通用户')
return rfm
def segment_by_behavior(self):
"""
基于行为模式分群
"""
segments = []
for user_id, group in self.data.groupby('user_id'):
# 计算用户行为特征
total_visits = len(group)
total_purchases = group['is_purchase'].sum() if 'is_purchase' in group else 0
conversion_rate = total_purchases / total_visits if total_visits > 0 else 0
# 分群逻辑
if conversion_rate > 0.3 and total_visits > 10:
segments.append('忠实用户')
elif conversion_rate > 0.1:
segments.append('潜在用户')
elif total_visits > 5 and conversion_rate == 0:
segments.append('浏览用户')
else:
segments.append('新用户')
return segments
SQL实现方式
-- 用户分群查询示例
WITH user_metrics AS (
SELECT
user_id,
COUNT(DISTINCT order_id) as order_count,
SUM(amount) as total_amount,
MAX(order_date) as last_order_date,
DATEDIFF(CURRENT_DATE, MAX(order_date)) as days_since_last_order
FROM orders
GROUP BY user_id
),
user_segments AS (
SELECT
user_id,
CASE
WHEN order_count >= 10 AND total_amount >= 10000
THEN 'VIP用户'
WHEN order_count >= 5 AND days_since_last_order <= 30
THEN '活跃高价值用户'
WHEN days_since_last_order <= 30
THEN '近期活跃用户'
WHEN days_since_last_order BETWEEN 31 AND 90
THEN '沉默用户'
WHEN days_since_last_order > 90
THEN '流失风险用户'
ELSE '新用户'
END as user_segment
FROM user_metrics
)
SELECT * FROM user_segments;
复杂分群规则引擎
class RuleBasedSegmentation:
def __init__(self):
self.rules = []
def add_rule(self, rule_name, condition_func, segment_name):
"""
添加分群规则
Args:
rule_name: 规则名称
condition_func: 条件判断函数
segment_name: 分群名称
"""
self.rules.append({
'name': rule_name,
'condition': condition_func,
'segment': segment_name
})
def evaluate(self, user_data):
"""
评估用户属于哪个群组
"""
results = {}
for user_id, user_info in user_data.items():
matched_segments = []
for rule in self.rules:
try:
if rule['condition'](user_info):
matched_segments.append(rule['segment'])
except Exception as e:
print(f"规则 {rule['name']} 评估失败: {e}")
# 优先级处理 - 选择第一个匹配的规则
results[user_id] = matched_segments[0] if matched_segments else '未分类'
return results
# 使用示例
segmentation = RuleBasedSegmentation()
# 定义规则函数
def is_vip(user):
return (user['total_orders'] >= 20 and
user['total_spent'] >= 50000 and
user['last_order_days'] <= 90)
def is_active_user(user):
return (user['last_order_days'] <= 30 and
user['total_orders'] >= 5)
def is_churn_risk(user):
return (user['last_order_days'] > 90 and
user['last_order_days'] <= 180)
# 添加规则
segmentation.add_rule('vip_rule', is_vip, 'VIP用户')
segmentation.add_rule('active_rule', is_active_user, '活跃用户')
segmentation.add_rule('churn_rule', is_churn_risk, '流失风险用户')
实时分群系统
from collections import defaultdict
import redis
import json
class RealTimeSegmentation:
def __init__(self, redis_client=None):
self.redis = redis_client or redis.Redis()
self.segment_cache = defaultdict(set)
def process_event(self, user_id, event_type, event_data):
"""
实时处理用户事件并更新分群
Args:
user_id: 用户ID
event_type: 事件类型
event_data: 事件数据
"""
# 更新用户画像
user_profile_key = f"user:{user_id}:profile"
# 从Redis获取用户画像
profile = self.redis.get(user_profile_key)
if profile:
profile = json.loads(profile)
else:
profile = {'events': [], 'segments': []}
# 记录事件
profile['events'].append({
'type': event_type,
'data': event_data,
'timestamp': time.time()
})
# 计算新分群
new_segments = self._calculate_segments(profile)
# 如果分群发生变化,更新缓存
old_segments = set(profile.get('segments', []))
if set(new_segments) != old_segments:
profile['segments'] = new_segments
self.redis.set(user_profile_key, json.dumps(profile))
# 更新分群缓存
for segment in new_segments:
self.segment_cache[segment].add(user_id)
for segment in old_segments - set(new_segments):
self.segment_cache[segment].discard(user_id)
return new_segments
def _calculate_segments(self, profile):
"""
计算用户所属分群
"""
segments = []
# 基于事件类型分群
events = profile['events']
recent_events = [e for e in events if e['timestamp'] > time.time() - 86400] # 24小时
# 规则示例
if any(e['type'] == 'purchase' for e in recent_events):
segments.append('今日购买用户')
if len([e for e in recent_events if e['type'] == 'view']) > 5:
segments.append('高频浏览用户')
# 检查是否在特定群组
if self._is_high_value(profile):
segments.append('高价值用户')
return segments
配置文件驱动分群
# segmentation_config.yaml
segments:
- name: "VIP用户"
priority: 1
conditions:
total_orders:
operator: ">="
value: 20
total_spent:
operator: ">="
value: 50000
last_order_days:
operator: "<="
value: 90
- name: "高活跃用户"
priority: 2
conditions:
monthly_visits:
operator: ">="
value: 10
conversion_rate:
operator: ">="
value: 0.2
- name: "流失风险用户"
priority: 3
conditions:
last_order_days:
operator: ">"
value: 90
days_since_last_login:
operator: ">"
value: 30
import yaml
import operator
class ConfigDrivenSegmentation:
def __init__(self, config_path):
with open(config_path, 'r') as f:
self.config = yaml.safe_load(f)
self.operators = {
'>=': operator.ge,
'<=': operator.le,
'>': operator.gt,
'<': operator.lt,
'==': operator.eq,
'!=': operator.ne
}
def segment_user(self, user_data):
"""
根据配置分群用户
"""
segments = self.config['segments']
# 按优先级排序
segments.sort(key=lambda x: x['priority'])
for segment in segments:
if self._check_conditions(segment['conditions'], user_data):
return segment['name']
return '未分类'
def _check_conditions(self, conditions, user_data):
"""
检查所有条件是否满足
"""
for field, condition in conditions.items():
user_value = user_data.get(field)
op = self.operators[condition['operator']]
if not op(user_value, condition['value']):
return False
return True
批量分群脚本
def batch_segment_users(user_records, batch_size=1000):
"""
批量处理用户分群
Args:
user_records: 用户记录列表
batch_size: 批处理大小
"""
segmenter = ConfigDrivenSegmentation('segmentation_config.yaml')
results = []
# 分批处理
for i in range(0, len(user_records), batch_size):
batch = user_records[i:i+batch_size]
for user in batch:
segment = segmenter.segment_user(user)
results.append({
'user_id': user['user_id'],
'segment': segment,
'processed_at': datetime.now()
})
# 进度报告
progress = min((i + batch_size) / len(user_records) * 100, 100)
print(f"处理进度: {progress:.2f}%")
return results
最佳实践建议
- 数据质量检查: 分群前验证数据完整性
- 性能优化: 使用索引、分批处理大数据集
- 测试验证: 编写单元测试验证分群逻辑
- 监控告警: 设置分群结果的异常监控
- 迭代更新: 定期根据业务需求调整分群规则
- 文档记录: 详细记录每个分群的定义和规则
选择哪种实现方式取决于你的具体需求：
- 简单规则: 使用SQL或Python直接实现
- 复杂规则: 使用规则引擎
- 实时处理: 使用流处理框架
- 灵活配置: 使用配置文件驱动
