本文目录导读：
我来详细介绍如何编写用户行为挖掘脚本,从基础概念到具体实现。
用户行为挖掘概述
用户行为挖掘是从用户操作日志中提取有价值信息的过程,包括：
- 页面浏览路径
- 功能使用频率
- 停留时间分析
- 转化漏斗
- 用户分群
基础脚本框架
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()
这个脚本框架可以根据具体业务场景进行扩展和定制,关键在于理解用户行为背后的业务逻辑。
