🐼 Lab 2: 教务系统 Pandas 数据分析实战

课程：数据库技术与应用：AI时代的基石数据技能
授课教师：王健楠 (计算机科学与技术系)
学期：2026年（春季）

🎯 实验目的与学习目标

完成本实验后，你应能够：

1. 建立数据分析的整体认知：理解数据如何从原始记录出发，经过数据流水线（pipeline）处理，最终形成可解释、可用于管理决策的指标与报告。
2. 熟练掌握 Pandas 的核心数据处理技能：过滤、去重、排序、分组聚合、透视分析与表连接，并能在真实业务场景中组合运用。

📖 实验背景

欢迎来到 Lab 2！本次实验中，你将担任清华大学教务处数据中心（Academic Data Center）的实习数据分析师。

教务处每天都会产生海量的选课与成绩数据。为了更好地支持教学评估、学业预警以及自动化成绩管理，你的任务是使用 Python 的数据科学利器 Pandas，对原始的教务数据库导出文件进行清洗、特征工程、多维聚合与综合透视分析。

准备好用数据驱动校园管理了吗？

---

评分规则： 共 100 分（8 道编程题）。 作答要求： 请在包含 # YOUR CODE HERE 的代码块中编写完整逻辑，并删除 raise NotImplementedError()。运行后续的测试块，测试块通过表示该步骤满足基础要求。

📊 第 0 步：数据加载与环境准备

本实验已随 notebook 一并提供 data/ 目录下的两张核心表：选课成绩明细表 (scores.csv) 与 课程信息表 (course_info.csv)。

请直接运行下面的代码块，将文件加载为 Pandas DataFrame。

# 说明：课程资料已提供 data/scores.csv 与 data/course_info.csv
import pandas as pd
import numpy as np
from pathlib import Path


data_dir = Path("data")
scores_path = data_dir / "scores.csv"
course_info_path = data_dir / "course_info.csv"

if not scores_path.exists() or not course_info_path.exists():
    raise FileNotFoundError("未找到 data/scores.csv 或 data/course_info.csv，请确认课程资料中的 data 目录已完整放在 notebook 同级目录。")

scores = pd.read_csv(scores_path, dtype={"sid": str})
course_info = pd.read_csv(course_info_path)

print("✅ 数据加载成功！")
print(f"   scores 文件: {scores_path.resolve()}")
print(f"   course_info 文件: {course_info_path.resolve()}")
print(f"   scores (成绩明细表): {scores.shape[0]} 条记录, {scores.shape[1]} 个字段  |  字段名: {list(scores.columns)}")
print(f"   course_info (课程信息表): {course_info.shape[0]} 门课程, {course_info.shape[1]} 个字段  |  字段名: {list(course_info.columns)}")
display(scores.head(8))

---

🧑‍💻 数据分析任务大厅

🧹 任务 1: 基础数据清洗与异常处理 (12 分)

在正式分析前，必须对底层数据进行清洗。由于存在缺考和重修的情况，我们需要提炼出一份“有效最高分成绩单”。

请编写代码完成以下流程：

1. 过滤缺考： 创建 valid_scores，从 scores 表中移除所有 score == 0 的行。
2. 异常侦测： 创建 dup_check，找出 valid_scores 中同一学生选修了同一门课超过一次的记录（即同一 sid + course 出现多次）。请将所有相关的重复行都保留以供人工核查。
3. 归一化处理： 创建 clean_scores。对于同一学生的同一门课有多条有效记录时，系统规定只保留最高分的那一条记录。最终保证 clean_scores 中每个 (sid, course) 组合只出现一次。
4. 计算清洗量： 声明一个整数变量 n_removed，计算从 valid_scores 到 clean_scores 去重时被移除的行数。

# YOUR CODE HERE
raise NotImplementedError()

print(f"有效成绩表 valid_scores: {len(valid_scores)} 行")
print("侦测到的重复选课记录 dup_check (前20行):")
display(dup_check.head(20))
print(f"最终干净数据 clean_scores: {len(clean_scores)} 行")
print(f"去重清洗移除: {n_removed} 行")

assert isinstance(valid_scores, pd.DataFrame)
assert isinstance(dup_check, pd.DataFrame)
assert isinstance(clean_scores, pd.DataFrame)
assert isinstance(n_removed, (int, np.integer))

assert (valid_scores["score"] > 0).all(), "valid_scores 不应包含 score=0 的异常记录"
assert clean_scores.duplicated(subset=["sid", "course"]).sum() == 0, "clean_scores 中不应存在重复 (sid, course)"
assert n_removed == len(valid_scores) - len(clean_scores), "n_removed 计算不正确"

if len(dup_check) > 0:
    assert dup_check.duplicated(subset=["sid", "course"], keep=False).all(), "dup_check 应只包含重复选课记录"

print("✅ Q1 基础数据清洗与去重测试通过！")

🏷️ 任务 2: 学业预警系统与特征工程 (12 分)

教务处需要构建一套“学业预警模型”。请基于 Q1 产出的干净数据 clean_scores（请复制一份进行操作，不要修改原数据：df = clean_scores.copy()），添加业务所需的特征列：

1. 等级评定： 添加 grade 列，映射规则如下：
   • A (>=90), B (>=80), C (>=70), D (>=60), F (<60)
2. 核心课标记： 添加 is_cs_core 列（布尔值 True/False）。当开课院系 dept == "计算机系" 且 课程名属于 ["数据结构", "操作系统", "数据库原理"] 时标记为 True。
3. 学业预警名单： 创建列表 risk_students。请筛选出至少有一门课 grade 为 C 或 D 或 F 的学生姓名。列表需去重，并按 Python 默认字符串升序排序。
4. 等级大盘分布： 创建 Series grade_dist。统计整个数据集中各个 grade（A/B/C/D/F）的出现次数。要求索引严格按照 ["A", "B", "C", "D", "F"] 排序（即使某个等级数量为0，也需包含并在结果中补 0）。

df = clean_scores.copy()

# YOUR CODE HERE
raise NotImplementedError()

print("添加特征后的数据概览:")
display(df.head(5))
print(f"🚨 学业预警名单人数: {len(risk_students)}")
print(f"🚨 学业预警名单前20个: {risk_students[:20]}")
print("📊 全校成绩等级分布 grade_dist:")
print(grade_dist)

assert "grade" in df.columns and "is_cs_core" in df.columns
assert isinstance(risk_students, list)
assert isinstance(grade_dist, pd.Series)

assert set(df["grade"].dropna().unique()).issubset({"A", "B", "C", "D", "F"}), "grade 取值不在 A/B/C/D/F 内"
assert df["is_cs_core"].dropna().isin([True, False]).all(), "is_cs_core 必须是布尔值"
assert risk_students == sorted(set(risk_students)), "risk_students 应去重并排序"
assert list(grade_dist.index) == ["A", "B", "C", "D", "F"], "grade_dist 索引顺序应为 A/B/C/D/F"
assert int(grade_dist.sum()) == len(df), "grade_dist 总数应等于 df 行数"

print("✅ Q2 特征工程与预警名单测试通过！")

📈 任务 3: 多维度学情聚合分析 (12 分)

继续基于 clean_scores，你需要为各级教学管理者提供多维度的数据报表：

1. 生成学生学期画像： 创建 DataFrame student_report。按学生学号（sid）分组，利用命名聚合 (Named Aggregation) 一次性计算出：
   • name: 学生姓名（每个 sid 保留任意一个一致姓名即可，如 first）
   • num_courses: 选课门数（计数）
   • avg_score: 平均分（平均值，保留原始精度即可）
   • max_score: 最高分
   • min_score: 最低分
   • 同时，请在聚合结果后，自行添加一列 score_range（极差 = 最高分 - 最低分），用于衡量学生成绩的稳定性。
2. 交叉对比分析： 创建 Series dept_gender_avg。按 (dept, gender) 进行多级分组，计算平均分。
3. 各科“状元”榜单： 创建 Series top_per_course。对每门课程，找出得分最高的学生姓名。索引为课程名，值为学生姓名。若最高分并列，任选其中一名即可。

# YOUR CODE HERE
raise NotImplementedError()

print("🧑‍🎓 学生个人学情报告（按 avg_score 降序前20行）:")
display(student_report.sort_values("avg_score", ascending=False).head(20))
print("🏫 院系×性别 平均分对比:")
print(dept_gender_avg)
print("👑 各科榜首学生（前20门课）:")
print(top_per_course.head(20))

assert isinstance(student_report, pd.DataFrame)
assert "score_range" in student_report.columns
assert isinstance(dept_gender_avg, pd.Series)
assert isinstance(top_per_course, pd.Series)

assert student_report.index.is_unique, "student_report 索引应唯一"
assert (student_report["score_range"] == (student_report["max_score"] - student_report["min_score"]).values).all(), "score_range 计算不正确"
assert dept_gender_avg.index.nlevels == 2, "dept_gender_avg 应为二级索引 (dept, gender)"
assert top_per_course.index.is_unique, "top_per_course 每门课应只对应一位榜首"

print("✅ Q3 多维聚合分析逻辑测试通过！")

🧮 任务 4: 核心指标计算 —— 加权 GPA (12 分)

评奖评优季即将到来，你需要计算每位学生的标准加权 GPA。 注意：目前 clean_scores 只有成绩，没有学分数据，你需要将它与 course_info 表进行关联连接（Join）。

1. 建立事实宽表： 使用 pd.merge() 将 clean_scores 与 course_info 按 course 字段进行内连接，赋值给新的 DataFrame merged。
2. 折算基础绩点： 在 merged 中添加 gp 列（单科绩点），严格遵循以下四分制换算规则：
   • score >= 95 对应 4.0 ; score >= 90 对应 3.7 ; score >= 85 对应 3.3 ; score >= 80 对应 3.0
   • score >= 75 对应 2.7 ; score >= 70 对应 2.3 ; score >= 60 对应 1.0 ; score < 60 对应 0.0
3. 计算加权基数： 添加 weighted_gp 列（该课程的加权绩点 = gp * credits）。
4. 计算最终 GPA： 创建 Series gpa_table。按学生学号（sid）分组，计算总加权 GPA（公式：sum(weighted_gp) / sum(credits)）。请确保结果按 GPA 从高到低 降序排列。
5. 提取尖子生： 创建列表 gpa_top3。按 gpa_table 的当前排序结果，提取前 3 名学生学号。

# YOUR CODE HERE
raise NotImplementedError()

print("🏆 全校 GPA 排名榜单（前20名）:")
display(gpa_table.head(20))
print(f"🏅 Top 3 拔尖人才: {gpa_top3}")

assert isinstance(merged, pd.DataFrame)
assert "gp" in merged.columns and "weighted_gp" in merged.columns
assert isinstance(gpa_table, pd.Series)
assert isinstance(gpa_top3, list)

allowed_gp = {0.0, 1.0, 2.3, 2.7, 3.0, 3.3, 3.7, 4.0}
assert set(merged["gp"].unique()).issubset(allowed_gp), "gp 取值不符合换算规则"
assert np.allclose(merged["weighted_gp"], merged["gp"] * merged["credits"]), "weighted_gp 计算错误"
assert gpa_table.is_monotonic_decreasing, "gpa_table 应按降序排列"

if len(gpa_table) >= 3:
    assert len(gpa_top3) == 3, "gpa_top3 应包含3名学生"
    assert gpa_top3 == gpa_table.head(3).index.tolist(), "gpa_top3 与 gpa_table 前3名不一致"

print("✅ Q4 表连接与 GPA 加权计算测试通过！")

🔍 任务 5: 教学资源调度全景透视 (12 分)

排课中心想了解各个院系对不同课程的选修倾向，我们需要借用数据透视表（Pivot Table）来构建一个二维的“院系-课程”矩阵。 请基于 clean_scores 完成以下交叉分析：

1. 均分透视： 创建 DataFrame pivot_dept。通过数据透视表，设定行索引为 dept，列索引为 course，统计值为 score 的平均值。没有选课交叉的单元格保持默认的 NaN 即可。
2. 选课热度透视： 创建 DataFrame dept_course_count。设定行索引为 dept，列索引为 course，统计值为 score 的计数（即选课人数）。要求将缺失值填充为 0。
3. 课程强势院系定位： 创建 Series best_dept_per_course。基于上面的均分透视表 pivot_dept，按列（每门课程）找出平均分最高的院系名称。
4. 交叉覆盖度评估： 声明一个整数变量 cross_dept_course。统计在全校范围内，总共有多少个 (dept, course) 组合存在实质选课记录（即 dept_course_count > 0 的非零单元格数量）。

# YOUR CODE HERE
raise NotImplementedError()

print("🗺️ 院系×课程 平均分二维透视 (前6行×前12列):")
display(pivot_dept.iloc[:6, :12])
print("💪 各课程表现最强院系（前20门课）:")
print(best_dept_per_course.head(20))
print(f"网状交叉覆盖度 (有效组合数): {cross_dept_course}")

assert isinstance(pivot_dept, pd.DataFrame)
assert isinstance(dept_course_count, pd.DataFrame)
assert isinstance(best_dept_per_course, pd.Series)
assert isinstance(cross_dept_course, (int, np.integer))

assert dept_course_count.isna().sum().sum() == 0, "dept_course_count 不应包含 NaN"
assert (dept_course_count >= 0).all().all(), "dept_course_count 应为非负"
assert set(best_dept_per_course.index) == set(pivot_dept.columns), "best_dept_per_course 索引应覆盖所有课程"
assert cross_dept_course == int((dept_course_count > 0).sum().sum()), "cross_dept_course 计算错误"

print("✅ Q5 数据透视与交叉覆盖度计算通过！")

🏆 任务 6: 院系核心竞争力评估流水线 (15 分)

这是一项高度综合的任务！教务处长要求提供一份纯粹反映“硬核学术实力”的院系必修课竞争力报告。

请将你的分析逻辑封装为一条严谨的数据处理流水线：

1. 构建全局视野： 将 clean_scores 与 course_info 进行内连接合并，赋值给 DataFrame full。
2. 剥离选修噪音： 在 full 表中，过滤数据使得只保留 ctype == "必修" 的核心课程记录，将过滤后的表赋值给 core。
3. 多维竞争力聚合： 创建 DataFrame dept_summary，按 dept 分组，计算三大核心指标：
   • core_avg: 该院系必修课的总平均分。
   • core_90_rate: 优良率（该院系必修课记录中，成绩 >= 90 的占比，取值范围 0~1）。
   • n_students: 该院系参与必修课的不重复学生人数（nunique）。
4. 实力榜单排序： 强制将 dept_summary 按核心指标 core_avg 进行降序排列。
5. 输出冠军院系： 将必修课平均分登顶的院系名称提取为字符串，赋值给变量 strongest_dept。

# YOUR CODE HERE
raise NotImplementedError()

print("📊 院系必修课学业竞争力报告:")
display(dept_summary)
print(f"🌟 本年度必修课最强硬核院系: {strongest_dept}")

assert isinstance(full, pd.DataFrame)
assert isinstance(core, pd.DataFrame)
assert isinstance(dept_summary, pd.DataFrame)
assert set(dept_summary.columns) >= {"core_avg", "core_90_rate", "n_students"}
assert isinstance(strongest_dept, str)

assert (core["ctype"] == "必修").all(), "core 应只包含必修课记录"
assert dept_summary["core_avg"].is_monotonic_decreasing, "dept_summary 应按 core_avg 降序"
assert dept_summary["core_90_rate"].between(0, 1).all(), "core_90_rate 应位于 [0,1]"
assert strongest_dept == dept_summary.index[0], "strongest_dept 应等于榜首院系"

print("✅ Q6 院系核心竞争力综合评估流水线测试通过！")

🔄 任务 7: 深度数据挖掘 —— 重修轨迹动态追踪 (13 分)

教研组长提出了一个尖锐的问题：“同学们重修一门课，成绩真的都提升了吗？” 你需要对底层存在历史冗余的原始数据进行深度挖掘。

注意：本次任务必须基于最原始的、未去重的 scores 数据表进行分析！

1. 获取有效全集： 从原始 scores 中筛选出真正参加了考试的记录（score > 0），赋值给 DataFrame valid。
2. 锁定重修人群： 创建 DataFrame repeat_students。从 valid 中挖掘出同一学生同一门课出现了 2 次及以上的特殊 (sid, course) 组合，并将这些组合涉及的所有原始记录行全部提取出来。
3. 时序变化量计算： 对上一步的 repeat_students，按 (sid, course, term) 排序后，计算每个分组前后的成绩变化量 diff（即：最晚一次成绩减去最早一次成绩）。本数据中 term 采用 YYYY春/秋 格式，可直接按字符串升序作为时间顺序。
   • 请输出结果 DataFrame improvements。它必须包含每组的 name, course, first_score, last_score, diff 这五列。
4. 得出最终结论： 声明一个布尔变量 all_improved。判定是否在分析样本中，所有重修学生的最终成绩都比初始成绩有提升（即 diff > 0）。

# YOUR CODE HERE
raise NotImplementedError()

print("📈 重修学生成绩变化轨迹（前20行）:")
display(improvements.head(20))
print(f"结论判定：所有重修学生成绩均实现提升？ -> {all_improved}")

assert isinstance(valid, pd.DataFrame)
assert isinstance(repeat_students, pd.DataFrame)
assert isinstance(improvements, pd.DataFrame)
assert set(improvements.columns) >= {"name", "course", "first_score", "last_score", "diff"}
assert isinstance(all_improved, bool)

assert (valid["score"] > 0).all(), "valid 应只包含 score>0 的记录"
if len(repeat_students) > 0:
    assert repeat_students.duplicated(subset=["sid", "course"], keep=False).all(), "repeat_students 仅应包含重修组合"
assert (improvements["diff"] == improvements["last_score"] - improvements["first_score"]).all(), "diff 计算错误"
assert all_improved == bool((improvements["diff"] > 0).all()), "all_improved 计算错误"

print("✅ Q7 重修轨迹深度挖掘逻辑测试通过！")

🖨️ 任务 8: 生产环境部署 —— 成绩单自动化生成 API (12 分)

实习的最后阶段，工程部门需要你将之前零散的数据处理逻辑，沉淀为一个稳定可复用的内部接口。

请编写一个 Python 函数 generate_transcript(sid)，当传入任意学号时，它能自动完成全链路的数据获取、关联计算与格式化输出。

API 规格说明：

• 输入参数 sid：字符串类型的目标学号。
• 内部逻辑：
  1. 必须基于你清洗好的 clean_scores 来检索该学生记录。
  2. 与全局变量 course_info 进行合并以补全课程元数据（学分等）。
  3. 按照 Q4 严格的四分制换算规则，动态计算每门课的单科绩点 gp 以及 weighted_gp。
• 返回值 (Return)：返回一个组装好的 Python 字典 (Dict)，必须包含以下严格命名的键：
  • "name": 学生姓名（字符串）
  • "dept": 院系名称（字符串）
  • "total_credits": 该生已修总学分（整数 int）
  • "gpa": 该生的标准加权 GPA（浮点数，请利用 round(x, 3) 保留 3 位小数）
  • "transcript": 一份精美的明细 DataFrame，仅包含 ["course", "credits", "score", "gp"] 四列，且必须强制按 score 的分值从高到低降序排列展示。

在代码块末尾，我们已经为你编写了针对学号 2023010102 (李娜) 和 2023020203 (周琳) 的调用测试。能够完美打印，即代表系统级联调成功！

def generate_transcript(sid):
    # YOUR CODE HERE
    raise NotImplementedError()

# ================== 自动化系统联调测试 ==================
t1 = generate_transcript("2023010102")
t2 = generate_transcript("2023020203")

for t in [t1, t2]:
    print("=== 🎓 学生学业档案 ===")
    print(f"姓名: {t['name']}  |  院系: {t['dept']}")
    print(f"已修总学分: {t['total_credits']}  |  加权 GPA: {t['gpa']}")
    print("【成绩明细单】")
    display(t["transcript"].head(20))
    print("=" * 40)

assert callable(generate_transcript)
assert isinstance(t1, dict) and isinstance(t2, dict)
assert set(t1.keys()) >= {"name", "dept", "total_credits", "gpa", "transcript"}
assert isinstance(t1["transcript"], pd.DataFrame)
assert list(t1["transcript"].columns) == ["course", "credits", "score", "gp"], "transcript 列应为 [course, credits, score, gp]"
assert t1["transcript"]["score"].is_monotonic_decreasing, "transcript 应按 score 降序"
assert isinstance(t1["total_credits"], (int, np.integer)) and t1["total_credits"] >= 0
assert isinstance(t1["gpa"], (float, np.floating))
print("✅ Q8 自动化成绩单 API 系统集成测试通过！")

---

🎉 实习圆满结束！

恭喜你完成了教务处数据中心的全部实战挑战！你已经充分证明了自己熟练运用 Pandas 驾驭复杂业务数据的能力。这份硬核的数据分析经验，必将成为你未来职业生涯的宝贵基石。

提交前最终检查：

• 建议点击顶部菜单栏的 Kernel -> Restart & Run All，确保代码环境从头到尾运行顺畅无报错。
• 完成全部 8 个任务后，确认测试均通过。
• 确认无误后，请保存并上传该 .ipynb 文件！