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(minutessession_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(dayswindow) recent group[group[timestamp] cutoff] row[fquestions_{window}d] len(recent) row[fcorrect_rate_{window}d] ( recent[is_correct].mean() if len(recent) 0 else 0 ) row[favg_time_{window}d] ( recent[time_spent].mean() if len(recent) 0 else 0 ) row[fsessions_{window}d] ( recent[session_id].nunique() if len(recent) 0 else 0 ) # 趋势特征最近3天 vs 前3-7天 recent_3d group[ group[timestamp] now - pd.Timedelta(days3) ] older_3d group[ (group[timestamp] now - pd.Timedelta(days7)) (group[timestamp] now - pd.Timedelta(days3)) ] 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, onstudent_id) merged merged.merge(labels, onstudent_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_size0.2, random_state42 ) model RandomForestClassifier( n_estimators100, max_depth8, min_samples_leaf10, random_state42, n_jobs-1, ) model.fit(X_train, y_train) y_pred model.predict(X_test) report classification_report( y_test, y_pred, output_dictTrue ) # 输出 Top-5 重要特征 importances sorted( zip(feature_names, model.feature_importances_), keylambda x: x[1], reverseTrue ) 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 [fS{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: fQ{np.random.randint(1, 5001)}, knowledge_id: fK{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( daysnp.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_weightbalanced可以有效缓解。但要注意SMOTE 生成的人工样本会降低模型在真实数据上的校准度。特征因果 vs 相关模型可能发现跳过题目多 流失风险高但这是相关性而不是因果——学生在决定放弃之前行为已经发生变化。真正的 actionable insight 是当检测到连续跳过 3 题信号时推送干预如降低难度或推荐视频而不是等到学生已经流失才触发挽留。五、总结学习行为分析的核心是从事件流到特征向量的转化过程。分三个阶段提取——基础统计快照、时序窗口趋势、序列模式状态变化——是实践证明有效的特征工程范式。代码上要注意三点会话切分的阈值调优30 分钟是通用经验但不同学段可能需要调整、特征空值处理新学生缺少的历史特征用 0 填充但要标记 low_confidence、以及模型输出的校准不能只给概率要附带置信度和关键特征贡献。