如何编写用户分群逻辑脚本

本文目录导读：

1. 基础分群脚本结构
2. SQL实现方式
3. 复杂分群规则引擎
4. 实时分群系统
5. 配置文件驱动分群
6. 批量分群脚本
7. 最佳实践建议

我来为你详细介绍用户分群逻辑脚本的编写方法,包含多种实现方式和最佳实践。

基础分群脚本结构

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

最佳实践建议

1. 数据质量检查: 分群前验证数据完整性
2. 性能优化: 使用索引、分批处理大数据集
3. 测试验证: 编写单元测试验证分群逻辑
4. 监控告警: 设置分群结果的异常监控
5. 迭代更新: 定期根据业务需求调整分群规则
6. 文档记录: 详细记录每个分群的定义和规则

选择哪种实现方式取决于你的具体需求：

• 简单规则: 使用SQL或Python直接实现
• 复杂规则: 使用规则引擎
• 实时处理: 使用流处理框架
• 灵活配置: 使用配置文件驱动