IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

如何编写用户分群筛选数据脚本

wen 实用脚本 32

本文目录导读：

如何编写用户分群筛选数据脚本

  1. 基于SQL的用户分群脚本
  2. Python脚本实现
  3. 配置文件驱动的分群
  4. 实时分群API
  5. 最佳实践建议

我来介绍几种常见的用户分群筛选数据脚本的编写方法。

基于SQL的用户分群脚本

基础SQL分群

-- 活跃用户分群
WITH user_segments AS (
    SELECT 
        user_id,
        CASE 
            WHEN last_login_date >= CURRENT_DATE - INTERVAL '7 days' THEN '高活跃用户'
            WHEN last_login_date >= CURRENT_DATE - INTERVAL '30 days' THEN '中等活跃用户'
            WHEN last_login_date >= CURRENT_DATE - INTERVAL '90 days' THEN '低活跃用户'
            ELSE '流失用户'
        END AS user_segment,
        -- 其他分群条件
        CASE 
            WHEN total_purchase > 10000 THEN '高价值用户'
            WHEN total_purchase > 1000 THEN '中等价值用户'
            ELSE '普通用户'
        END AS value_segment
    FROM users
    WHERE status = 'active'  -- 筛选条件
)
SELECT * FROM user_segments
WHERE user_segment = '高活跃用户'  -- 特定分群

多条件组合筛选

-- 复杂条件分群
SELECT 
    user_id,
    -- 根据多个维度组合分群
    CASE 
        WHEN purchase_frequency > 10 
             AND avg_order_value > 500 
             AND category_preference = '电子产品' 
        THEN '电子品类高价值用户'
        WHEN purchase_frequency > 5 
             AND recent_interaction = TRUE 
        THEN '潜力用户'
        WHEN is_new_user = TRUE 
             AND completed_profile = TRUE 
        THEN '待转化用户'
        ELSE '其他用户'
    END AS segment_name
FROM user_profile
WHERE created_at >= '2024-01-01'
  AND total_orders > 0

Python脚本实现

基础分群脚本

import pandas as pd
import datetime
from typing import List, Dict, Any
class UserSegmenter:
    def __init__(self, data: pd.DataFrame):
        self.data = data
    def segment_by_activity(self, df: pd.DataFrame) -> pd.DataFrame:
        """活跃度分群"""
        today = datetime.datetime.now()
        def get_activity_level(row):
            days_since_last_visit = (today - row['last_visit']).days
            if days_since_last_visit <= 7:
                return '活跃用户'
            elif days_since_last_visit <= 30:
                return '中等活跃用户'
            elif days_since_last_visit <= 90:
                return '沉淀用户'
            else:
                return '流失用户'
        df['activity_segment'] = df.apply(get_activity_level, axis=1)
        return df
    def segment_by_value(self, df: pd.DataFrame) -> pd.DataFrame:
        """价值分群"""
        def get_value_level(row):
            if row['total_spend'] >= 10000:
                return '高价值用户'
            elif row['total_spend'] >= 5000:
                return '中价值用户'
            else:
                return '低价值用户'
        df['value_segment'] = df.apply(get_value_level, axis=1)
        return df
    def combined_segment(self, df: pd.DataFrame) -> pd.DataFrame:
        """组合分群"""
        def get_segment(row):
            if row['activity_segment'] == '活跃用户' and row['value_segment'] == '高价值用户':
                return '核心用户'
            elif row['activity_segment'] == '活跃用户' and row['value_segment'] == '中价值用户':
                return '成长用户'
            elif row['activity_segment'] == '流失用户' and row['value_segment'] == '高价值用户':
                return '需挽回用户'
            else:
                return '普通用户'
        df['segment'] = df.apply(get_segment, axis=1)
        return df
    def filter_segment(self, segment_name: str, **criteria) -> pd.DataFrame:
        """筛选特定分群用户"""
        result = self.data.copy()
        # 应用分群逻辑
        result = self.segment_by_activity(result)
        result = self.segment_by_value(result)
        result = self.combined_segment(result)
        # 应用筛选条件
        if segment_name:
            result = result[result['segment'] == segment_name]
        # 应用额外条件
        for key, value in criteria.items():
            if key in result.columns:
                result = result[result[key] == value]
        return result
# 使用示例
# df = pd.read_csv('user_data.csv')
# segmenter = UserSegmenter(df)
# core_users = segmenter.filter_segment('核心用户', 
#                                       registered_date='2024-01-01',
#                                       country='CN')

规则引擎分群

from typing import Callable, Dict
from dataclasses import dataclass
@dataclass
class SegmentRule:
    name: str
    condition: Callable
    priority: int = 1
class RuleBasedSegmenter:
    def __init__(self):
        self.rules = []
    def add_rule(self, name: str, condition: Callable, priority: int = 1):
        """添加分群规则"""
        rule = SegmentRule(name=name, condition=condition, priority=priority)
        self.rules.append(rule)
        self.rules.sort(key=lambda x: x.priority, reverse=True)
    def segment_user(self, user_data: Dict) -> str:
        """对单个用户进行分群"""
        for rule in self.rules:
            if rule.condition(user_data):
                return rule.name
        return '未分类'
    def segment_users(self, users: List[Dict]) -> Dict[str, List]:
        """批量分群"""
        results = {'未分类': []}
        for user in users:
            segment = self.segment_user(user)
            if segment not in results:
                results[segment] = []
            results[segment].append(user)
        return results
# 使用示例
segmenter = RuleBasedSegmenter()
# 定义分群规则
segmenter.add_rule(
    name='高价值用户',
    condition=lambda u: u.get('total_orders', 0) > 100 and u.get('avg_order_value', 0) > 500,
    priority=3
)
segmenter.add_rule(
    name='新用户',
    condition=lambda u: u.get('days_since_register', 0) < 30,
    priority=2
)
segmenter.add_rule(
    name='活跃用户',
    condition=lambda u: u.get('last_login_days', 30) < 7,
    priority=1
)
# 批量分群
users = [
    {'total_orders': 150, 'avg_order_value': 600, 'days_since_register': 60, 'last_login_days': 3},
    {'total_orders': 50, 'avg_order_value': 200, 'days_since_register': 20, 'last_login_days': 2},
]
result = segmenter.segment_users(users)

配置文件驱动的分群

YAML配置文件

# segments_config.yaml
segments:
  - name: vip_users
    description: VIP用户
    conditions:
      - field: total_orders
        operator: '>='
        value: 100
      - field: total_spend
        operator: '>='
        value: 50000
    priority: 1
  - name: new_users
    description: 新注册用户
    conditions:
      - field: days_since_register
        operator: '<'
        value: 30
    priority: 2
  - name: active_users
    description: 活跃用户
    conditions:
      - field: last_login_days
        operator: '<='
        value: 7
    priority: 3

动态配置加载

import yaml
import operator
class ConfigDrivenSegmenter:
    def __init__(self, config_path: str):
        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,
            'in': lambda x, y: x in y,
            'between': lambda x, y: y[0] <= x <= y[1]
        }
    def check_condition(self, user: Dict, condition: Dict) -> bool:
        """检查单个条件"""
        field = condition['field']
        op = self.operators[condition['operator']]
        value = condition['value']
        if field not in user:
            return False
        return op(user[field], value)
    def segment_user(self, user: Dict) -> str:
        """对用户进行分群"""
        # 按优先级排序
        sorted_segments = sorted(
            self.config['segments'], 
            key=lambda x: x.get('priority', 0), 
            reverse=True
        )
        for segment in sorted_segments:
            conditions = segment['conditions']
            # 检查所有条件是否满足
            if all(self.check_condition(user, cond) for cond in conditions):
                return segment['name']
        return '未分类'

实时分群API

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
app = FastAPI()
class UserData(BaseModel):
    user_id: str
    total_orders: int
    total_spend: float
    last_login_days: int
    days_since_register: int
    category_preference: Optional[str] = None
class SegmentRequest(BaseModel):
    users: List[UserData]
    segment_name: Optional[str] = None
    filters: Optional[Dict] = None
@app.post("/api/segment")
async def segment_users(request: SegmentRequest):
    """用户分群API"""
    segmenter = RuleBasedSegmenter()
    # 配置规则...
    results = []
    for user in request.users:
        user_dict = user.dict()
        segment = segmenter.segment_user(user_dict)
        # 应用筛选
        if request.segment_name and segment != request.segment_name:
            continue
        results.append({
            'user_id': user.user_id,
            'segment': segment,
            'user_data': user_dict
        })
    return {'total': len(results), 'users': results}

最佳实践建议

  1. 性能优化：使用批处理、索引优化、并行计算
  2. 可维护性：配置驱动、模块化设计
  3. 准确性：数据质量检查、分群结果验证
  4. 扩展性：规则引擎、插件机制
  5. 监控告警：分群结果监控、异常检测

选择哪种实现方式取决于你的业务复杂度、数据规模和系统架构要求。

上一篇脚本怎样计算用户次日留存率

下一篇脚本如何动态更新用户标签

相关文章

抱歉，评论功能暂时关闭!