Python 学习行为分析：学生作答数据的特征工程与模型训练
一、平台存了 500 万条做题记录，却不知道学生为什么会放弃
一个在线教育平台每天产生几十万条学生作答记录——什么时候开始做题、答了哪道题、花了多少秒、是对是错、错了之后有没有看解析。这些数据孤零零地躺在 MySQL 里，没人知道怎么用。产品经理问："为什么这个学生连续 7 天登录，第 8 天突然不来了？"数据分析师给不出答案——因为答案不在某一条记录里，而在"一系列行为模式"中。
学习行为分析的挑战是：原始数据（答题日志）和业务洞察（流失原因、薄弱点）之间有巨大的语义鸿沟。这个鸿沟需要用特征工程来桥接——将原始的"时间-事件序列"转化为模型可理解的特征向量。
二、学习行为特征工程的完整 Pipeline
从原始日志到模型特征，需要经历三个层次的抽象：
每提升一个抽象层，特征的预测能力增强，但可解释性下降。对于学习行为分析，三层特征全保留，让模型自己学权重——事实证明，低层统计特征和高层时序特征对预测同等重要。
三、Python 实现：学习行为特征提取
import pandas as pd
import numpy as np
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report
import warnings
warnings.filterwarnings('ignore')
@dataclass
class AnswerLog:
"""单条作答记录"""
student_id: str
question_id: str
knowledge_id: str # 考察的知识点
is_correct: bool
time_spent: int # 作答耗时（秒）
timestamp: pd.Timestamp
viewed_solution: bool # 是否查看解析
skipped: bool # 是否跳过
class BehaviorFeatureExtractor:
"""学习行为特征提取器"""
def __init__(self, session_gap_minutes: int = 30):
self.session_gap = pd.Timedelta(minutes=session_gap_minutes)
self.scaler = StandardScaler()
def _sessionize(self, logs: pd.DataFrame) -> pd.DataFrame:
"""阶段1：会话切分（>30分钟无操作视为新会话）"""
logs = logs.sort_values(['student_id', 'timestamp'])
logs['prev_timestamp'] = logs.groupby(
'student_id'
)['timestamp'].shift(1)
logs['gap'] = logs['timestamp'] – logs['prev_timestamp']
# 新会话标记
logs['is_new_session'] = (
logs['gap'] > self.session_gap
) | logs['prev_timestamp'].isna()
logs['session_id'] = logs.groupby(
'student_id'
)['is_new_session'].cumsum()
return logs
def extract_basic_features(
self, logs: pd.DataFrame
) -> pd.DataFrame:
"""阶段2：基础统计特征"""
features = logs.groupby('student_id').agg(
total_questions=('question_id', 'count'),
correct_rate=('is_correct', 'mean'),
avg_time_spent=('time_spent', 'mean'),
std_time_spent=('time_spent', 'std'),
max_time_spent=('time_spent', 'max'),
view_solution_rate=('viewed_solution', 'mean'),
skip_rate=('skipped', 'mean'),
total_sessions=('session_id', 'nunique'),
unique_knowledge=('knowledge_id', 'nunique'),
).reset_index()
# 填空缺失值
features['std_time_spent'] = features[
'std_time_spent'
].fillna(0)
return features
def extract_temporal_features(
self, logs: pd.DataFrame,
window_days: List[int] = [1, 3, 7, 14]
) -> pd.DataFrame:
"""阶段3：时序特征 —— 滚动窗口统计"""
logs = logs.copy()
logs['date'] = logs['timestamp'].dt.date
now = logs['timestamp'].max()
all_features = []
for student_id, group in logs.groupby('student_id'):
row = {'student_id': student_id}
for window in window_days:
cutoff = now – pd.Timedelta(days=window)
recent = group[group['timestamp'] >= cutoff]
row[f'questions_{window}d'] = len(recent)
row[f'correct_rate_{window}d'] = (
recent['is_correct'].mean()
if len(recent) > 0 else 0
)
row[f'avg_time_{window}d'] = (
recent['time_spent'].mean()
if len(recent) > 0 else 0
)
row[f'sessions_{window}d'] = (
recent['session_id'].nunique()
if len(recent) > 0 else 0
)
# 趋势特征：最近3天 vs 前3-7天
recent_3d = group[
group['timestamp'] >= now – pd.Timedelta(days=3)
]
older_3d = group[
(group['timestamp'] >= now – pd.Timedelta(days=7)) &
(group['timestamp'] < now – pd.Timedelta(days=3))
]
if len(recent_3d) > 0 and len(older_3d) > 0:
row['correct_rate_trend'] = (
recent_3d['is_correct'].mean() –
older_3d['is_correct'].mean()
)
row['time_trend'] = (
recent_3d['time_spent'].mean() –
older_3d['time_spent'].mean()
)
else:
row['correct_rate_trend'] = 0
row['time_trend'] = 0
# 连续错误/正确模式
last_10 = group.tail(10)['is_correct'].values
row['consecutive_errors'] = self._count_consecutive(
last_10, False
)
row['consecutive_corrects'] = self._count_consecutive(
last_10, True
)
all_features.append(row)
return pd.DataFrame(all_features)
@staticmethod
def _count_consecutive(series: np.ndarray, target: bool) -> int:
"""计算最近连续的 target 数量"""
count = 0
for val in reversed(series):
if val == target:
count += 1
else:
break
return count
def prepare_training_data(
self, logs: pd.DataFrame, labels: pd.DataFrame
) -> Tuple[np.ndarray, np.ndarray, List[str]]:
"""准备训练数据"""
logs = self._sessionize(logs)
basic_feats = self.extract_basic_features(logs)
temporal_feats = self.extract_temporal_features(logs)
# 特征融合
merged = basic_feats.merge(temporal_feats, on='student_id')
merged = merged.merge(labels, on='student_id')
feature_cols = [c for c in merged.columns
if c not in ['student_id', 'label']]
X = merged[feature_cols].fillna(0).values
X = self.scaler.fit_transform(X)
y = merged['label'].values
return X, y, feature_cols
def train_risk_model(
self, X: np.ndarray, y: np.ndarray,
feature_names: List[str]
) -> Tuple[RandomForestClassifier, Dict]:
"""训练流失风险预测模型"""
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=42
)
model = RandomForestClassifier(
n_estimators=100,
max_depth=8,
min_samples_leaf=10,
random_state=42,
n_jobs=-1,
)
model.fit(X_train, y_train)
y_pred = model.predict(X_test)
report = classification_report(
y_test, y_pred, output_dict=True
)
# 输出 Top-5 重要特征
importances = sorted(
zip(feature_names, model.feature_importances_),
key=lambda x: x[1], reverse=True
)
print("Top-5 重要特征:")
for name, imp in importances[:5]:
print(f" {name}: {imp:.4f}")
return model, report
# 示例使用
def example_usage():
"""演示完整的特征工程 + 模型训练流程"""
# 模拟数据
np.random.seed(42)
n_students = 1000
n_logs = 50000
students = [f'S{i:04d}' for i in range(n_students)]
logs_data = []
for _ in range(n_logs):
sid = np.random.choice(students)
logs_data.append({
'student_id': sid,
'question_id': f'Q{np.random.randint(1, 5001)}',
'knowledge_id': f'K{np.random.randint(1, 101)}',
'is_correct': np.random.random() > 0.35,
'time_spent': int(np.random.exponential(60)),
'timestamp': pd.Timestamp('2026-06-01') + pd.Timedelta(
days=np.random.randint(0, 30)
),
'viewed_solution': np.random.random() > 0.6,
'skipped': np.random.random() > 0.85,
})
logs_df = pd.DataFrame(logs_data)
labels_df = pd.DataFrame({
'student_id': students,
'label': np.random.choice([0, 1], n_students, p=[0.8, 0.2]),
})
extractor = BehaviorFeatureExtractor()
X, y, feature_names = extractor.prepare_training_data(
logs_df, labels_df
)
model, report = extractor.train_risk_model(X, y, feature_names)
print(f"\\n模型准确率: {report['accuracy']:.3f}")
return model, extractor
四、边界分析与 Trade-offs
特征爆炸问题：滑动窗口 × 统计指标 = 特征数爆炸（7 个窗口 × 8 个指标 = 56 个特征）。但实际上大部分特征之间存在高相关性（如 questions_1d 和 questions_3d），需要通过特征选择（卡方检验或基于树的特征重要性）降维。经验是保留 Top-15 个特征，模型效果与全特征相当。
冷启动学生的处理：新学生没有时序特征（所有滚动窗口都是 0），模型会把他们统一预测为"低风险"——因为缺少活动信号。应该在输出预测时标注"数据不足，置信度低"，同时基于静态人口学特征（年级、注册渠道）做辅助预测。
样本不均衡的典型问题：流失学生通常只占 5%-15%，模型会倾向于预测"不流失"。用 SMOTE 过采样或调整类别权重（class_weight='balanced'）可以有效缓解。但要注意：SMOTE 生成的人工样本会降低模型在真实数据上的校准度。
特征因果 vs 相关：模型可能发现"跳过题目多 = 流失风险高"，但这是相关性而不是因果——学生在决定放弃之前，行为已经发生变化。真正的 actionable insight 是：当检测到"连续跳过 3 题"信号时，推送干预（如降低难度或推荐视频），而不是等到学生已经流失才触发挽留。
五、总结
学习行为分析的核心是"从事件流到特征向量"的转化过程。分三个阶段提取——基础统计（快照）、时序窗口（趋势）、序列模式（状态变化）——是实践证明有效的特征工程范式。代码上要注意三点：会话切分的阈值调优（30 分钟是通用经验，但不同学段可能需要调整）、特征空值处理（新学生缺少的历史特征用 0 填充，但要标记 low_confidence）、以及模型输出的校准（不能只给概率，要附带置信度和关键特征贡献）。
