如何编写用户行为挖掘脚本

本文目录导读：

1. 用户行为挖掘概述
2. 基础脚本框架
3. 核心分析功能
4. 高级分析功能
5. 完整使用示例
6. 最佳实践建议

我来详细介绍如何编写用户行为挖掘脚本,从基础概念到具体实现。

用户行为挖掘概述

用户行为挖掘是从用户操作日志中提取有价值信息的过程,包括：

• 页面浏览路径
• 功能使用频率
• 停留时间分析
• 转化漏斗
• 用户分群

基础脚本框架

1 数据收集脚本

import pandas as pd
import json
from datetime import datetime, timedelta
import numpy as np
class UserBehaviorMiner:
    def __init__(self, log_file_path):
        self.log_file = log_file_path
        self.logs = []
    def load_logs(self):
        """加载用户行为日志"""
        with open(self.log_file, 'r', encoding='utf-8') as f:
            for line in f:
                try:
                    log_entry = json.loads(line.strip())
                    self.logs.append(log_entry)
                except json.JSONDecodeError:
                    continue
        return pd.DataFrame(self.logs)
    def preprocess_data(self, df):
        """数据预处理"""
        # 转换时间戳
        if 'timestamp' in df.columns:
            df['timestamp'] = pd.to_datetime(df['timestamp'])
        # 处理缺失值
        df = df.fillna({'user_id': 'unknown', 'action': 'undefined'})
        # 添加日期维度
        df['date'] = df['timestamp'].dt.date
        df['hour'] = df['timestamp'].dt.hour
        return df

2 用户会话分割

def segment_sessions(self, df, timeout_minutes=30):
    """将用户行为分割成会话"""
    df = df.sort_values(['user_id', 'timestamp'])
    # 计算时间差
    df['time_diff'] = df.groupby('user_id')['timestamp'].diff()
    # 超过超时时间视为新会话
    df['is_new_session'] = (
        df['time_diff'].isna() | 
        (df['time_diff'] > pd.Timedelta(minutes=timeout_minutes))
    )
    # 生成会话ID
    df['session_id'] = df.groupby('user_id')['is_new_session'].cumsum()
    return df

核心分析功能

1 行为路径分析

def analyze_behavior_paths(self, df):
    """分析用户行为路径"""
    # 获取用户行为序列
    behavior_paths = df.groupby('session_id').agg({
        'action': lambda x: ' -> '.join(x),
        'timestamp': ['min', 'max'],
        'user_id': 'first'
    }).reset_index()
    # 统计路径频率
    path_frequency = behavior_paths['action'].value_counts().head(20)
    # 计算平均路径长度
    behavior_paths['path_length'] = behavior_paths['action'].str.count(' -> ') + 1
    return {
        'top_paths': path_frequency.to_dict(),
        'avg_path_length': behavior_paths['path_length'].mean(),
        'sessions_count': len(behavior_paths)
    }

2 停留时间分析

def analyze_dwell_time(self, df):
    """分析页面停留时间"""
    df = df.sort_values(['user_id', 'timestamp'])
    # 计算每个页面的停留时间
    df['next_action_time'] = df.groupby('user_id')['timestamp'].shift(-1)
    df['dwell_seconds'] = (
        df['next_action_time'] - df['timestamp']
    ).dt.total_seconds()
    # 过滤异常值（如超过1小时）
    df = df[df['dwell_seconds'].between(1, 3600)]
    # 按页面统计
    page_stats = df.groupby('page').agg({
        'dwell_seconds': ['mean', 'median', 'std'],
        'user_id': 'nunique'
    }).round(2)
    return page_stats

3 转化漏斗分析

def analyze_conversion_funnel(self, df, funnel_steps):
    """
    分析转化漏斗
    funnel_steps: ['visit_home', 'search_product', 'add_to_cart', 'checkout']
    """
    funnel_data = {}
    for i, step in enumerate(funnel_steps):
        step_users = df[df['action'] == step]['user_id'].unique()
        if i == 0:
            funnel_data[step] = {
                'users': len(step_users),
                'conversion_rate': 100.0
            }
        else:
            previous_users = df[df['action'].isin(funnel_steps[:i])]['user_id'].unique()
            funnel_data[step] = {
                'users': len(step_users),
                'conversion_rate': round(len(step_users) / len(previous_users) * 100, 2)
            }
    # 总体转化率
    total_conversion = round(
        funnel_data[funnel_steps[-1]]['users'] / 
        funnel_data[funnel_steps[0]]['users'] * 100, 
        2
    )
    return {
        'funnel_steps': funnel_steps,
        'funnel_data': funnel_data,
        'total_conversion_rate': total_conversion
    }

高级分析功能

1 用户分群

def user_segmentation(self, df):
    """用户分群分析"""
    # 计算用户指标
    user_metrics = df.groupby('user_id').agg({
        'action': 'count',
        'session_id': 'nunique',
        'timestamp': ['min', 'max']
    })
    user_metrics.columns = ['total_actions', 'total_sessions', 'first_visit', 'last_visit']
    # 计算活跃天数
    user_dates = df.groupby('user_id')['date'].nunique()
    user_metrics['active_days'] = user_dates
    # 用户分类
    conditions = [
        (user_metrics['total_sessions'] >= 10) & (user_metrics['active_days'] >= 5),
        (user_metrics['total_sessions'] >= 3) & (user_metrics['active_days'] >= 2),
        (user_metrics['total_sessions'] < 3)
    ]
    choices = ['忠实用户', '普通用户', '沉睡用户']
    user_metrics['user_segment'] = np.select(conditions, choices, default='新用户')
    # 统计各分类数量
    segment_stats = user_metrics['user_segment'].value_counts()
    return segment_stats

2 实时行为检测

class RealTimeBehaviorMonitor:
    def __init__(self, window_minutes=5):
        self.window_minutes = window_minutes
        self.recent_events = []
    def process_event(self, event):
        """处理实时事件"""
        current_time = datetime.now()
        # 添加当前事件
        self.recent_events.append({
            'user_id': event.get('user_id'),
            'action': event.get('action'),
            'timestamp': current_time
        })
        # 清理过期事件
        self.recent_events = [
            e for e in self.recent_events 
            if (current_time - e['timestamp']).total_seconds() < self.window_minutes * 60
        ]
        return self.detect_anomalies()
    def detect_anomalies(self):
        """检测异常行为"""
        user_frequency = {}
        for event in self.recent_events:
            user = event['user_id']
            user_frequency[user] = user_frequency.get(user, 0) + 1
        alerts = []
        for user, freq in user_frequency.items():
            if freq > self.window_minutes * 10:  # 每分钟超过10次
                alerts.append({
                    'user_id': user,
                    'frequency': freq,
                    'alert_type': '高频访问'
                })
        return alerts

完整使用示例

# 主程序入口
if __name__ == "__main__":
    # 初始化挖掘器
    miner = UserBehaviorMiner("user_logs.json")
    # 加载和处理数据
    df = miner.load_logs()
    df = miner.preprocess_data(df)
    df = miner.segment_sessions(df)
    # 执行分析
    print("=== 行为路径分析 ===")
    path_analysis = miner.analyze_behavior_paths(df)
    print(f"热门路径: {path_analysis['top_paths']}")
    print(f"平均路径长度: {path_analysis['avg_path_length']:.2f}")
    print("\n=== 停留时间分析 ===")
    dwell_stats = miner.analyze_dwell_time(df)
    print(dwell_stats.head())
    print("\n=== 转化漏斗分析 ===")
    funnel_steps = ['view_page', 'login', 'search', 'purchase']
    funnel = miner.analyze_conversion_funnel(df, funnel_steps)
    print(f"总体转化率: {funnel['total_conversion_rate']}%")
    print("\n=== 用户分群 ===")
    segments = miner.user_segmentation(df)
    print(segments)
    # 实时监控示例
    monitor = RealTimeBehaviorMonitor()
    sample_event = {
        'user_id': 'user123',
        'action': 'page_view',
        'page': '/home'
    }
    anomalies = monitor.process_event(sample_event)
    if anomalies:
        print(f"检测到异常: {anomalies}")

最佳实践建议

数据质量保证

• 数据验证：检查日志格式、时间戳合法性
• 去重处理：处理重复记录
• 异常过滤：排除机器人访问

性能优化

# 使用分块处理大数据
chunk_size = 10000
for chunk in pd.read_json('large_logs.json', lines=True, chunksize=chunk_size):
    process_chunk(chunk)

可视化输出

import matplotlib.pyplot as plt
import seaborn as sns
def visualize_funnel(funnel_data):
    """可视化转化漏斗"""
    steps = list(funnel_data['funnel_data'].keys())
    values = [v['users'] for v in funnel_data['funnel_data'].values()]
    plt.figure(figsize=(10, 6))
    plt.barh(steps, values)
    plt.xlabel('用户数')
    plt.title('转化漏斗分析')
    plt.show()

这个脚本框架可以根据具体业务场景进行扩展和定制,关键在于理解用户行为背后的业务逻辑。