from __future__ import annotations import hashlib import json import sys from pathlib import Path import numpy as np import pandas as pd REQUIRED_COLUMNS = [ "student_id", "age_years", "gender", "study_year", "chronotype", "weekday_sleep_hours", "weekend_sleep_hours", "sleep_quality_score", "bedtime_variability_minutes", "daytime_sleepiness_score", "evening_screen_hours", "weekly_study_hours", "attendance_percent", "prior_gpa", "exam_score", ] EXPECTED_RANGES = { "age_years": (18, 25), "study_year": (1, 4), "weekday_sleep_hours": (0, 24), "weekend_sleep_hours": (0, 24), "sleep_quality_score": (1, 10), "bedtime_variability_minutes": (5, 240), "daytime_sleepiness_score": (0, 24), "evening_screen_hours": (0.4, 6.5), "weekly_study_hours": (2, 35), "attendance_percent": (0, 100), "prior_gpa": (0, 4), "exam_score": (0, 100), } MODEL_PREDICTORS = [ "weekday_sleep_hours", "sleep_quality_score", "bedtime_variability_minutes", "prior_gpa", "weekly_study_hours", "attendance_percent", ] def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def join_ids(series: pd.Series) -> str: return ", ".join(series.astype(str).tolist()) or "无" def write_scatter_svg(data: pd.DataFrame, output: Path) -> None: plot = data[["weekday_sleep_hours", "exam_score"]].dropna() x = plot["weekday_sleep_hours"].to_numpy(float) y = plot["exam_score"].to_numpy(float) width, height = 900, 620 left, right, top, bottom = 90, 35, 55, 120 x_min, x_max = 4.0, 10.0 y_min, y_max = 40.0, 100.0 def sx(value: float) -> float: return left + (value - x_min) / (x_max - x_min) * (width - left - right) def sy(value: float) -> float: return height - bottom - (value - y_min) / (y_max - y_min) * (height - top - bottom) slope, intercept = np.polyfit(x, y, 1) parts = [ f'', ] ) output.write_text("\n".join(parts), encoding="utf-8") def main() -> None: if len(sys.argv) != 3: raise SystemExit("用法：分析代码.py <教学数据.csv> <输出目录>") input_path = Path(sys.argv[1]).resolve() output_dir = Path(sys.argv[2]).resolve() output_dir.mkdir(parents=True, exist_ok=True) raw_hash_before = sha256(input_path) raw = pd.read_csv(input_path) missing_columns = sorted(set(REQUIRED_COLUMNS) - set(raw.columns)) if missing_columns: raise ValueError(f"缺少变量：{missing_columns}") checks: list[dict[str, object]] = [] duplicate_ids = raw.loc[raw["student_id"].duplicated(keep=False), "student_id"] checks.append( { "检查类型": "重复编号", "变量": "student_id", "数量": int(duplicate_ids.size), "涉及记录": join_ids(duplicate_ids), "处理": "未发现时不处理；如发现则停止并交给研究者确认", } ) for column in REQUIRED_COLUMNS: count = int(raw[column].isna().sum()) if count: checks.append( { "检查类型": "缺失值", "变量": column, "数量": count, "涉及记录": join_ids(raw.loc[raw[column].isna(), "student_id"]), "处理": "原始文件保留空白；具体分析使用完整个案并报告样本量", } ) for column, (lower, upper) in EXPECTED_RANGES.items(): mask = raw[column].notna() & ~raw[column].between(lower, upper) if mask.any(): checks.append( { "检查类型": "超出变量范围", "变量": column, "数量": int(mask.sum()), "涉及记录": join_ids(raw.loc[mask, "student_id"]), "处理": "不改原始文件；按下方公开规则在分析副本中处理", } ) clean = raw.copy() minute_in_hour_mask = clean["weekday_sleep_hours"].between(60, 1440, inclusive="both") corrected_ids = clean.loc[minute_in_hour_mask, "student_id"].tolist() clean.loc[minute_in_hour_mask, "weekday_sleep_hours"] = clean.loc[minute_in_hour_mask, "weekday_sleep_hours"] / 60.0 invalid_exam_mask = ~clean["exam_score"].between(0, 100, inclusive="both") excluded_exam_ids = clean.loc[invalid_exam_mask, "student_id"].tolist() clean.loc[invalid_exam_mask, "exam_score"] = np.nan clean.to_csv(output_dir / "清洗后分析数据.csv", index=False, encoding="utf-8-sig", float_format="%.3f") quality = pd.DataFrame(checks) quality.to_csv(output_dir / "数据质量检查.csv", index=False, encoding="utf-8-sig") numeric_columns = list(EXPECTED_RANGES) desc_rows = [] for column in numeric_columns: series = clean[column] desc_rows.append( { "变量": column, "有效数": int(series.notna().sum()), "缺失数": int(series.isna().sum()), "均值": series.mean(), "标准差": series.std(ddof=1), "最小值": series.min(), "中位数": series.median(), "最大值": series.max(), } ) pd.DataFrame(desc_rows).to_csv(output_dir / "描述统计.csv", index=False, encoding="utf-8-sig", float_format="%.3f") correlation_columns = [ "weekday_sleep_hours", "weekend_sleep_hours", "sleep_quality_score", "bedtime_variability_minutes", "evening_screen_hours", "weekly_study_hours", "attendance_percent", "prior_gpa", "exam_score", ] correlations = clean[correlation_columns].corr(method="pearson") correlations.to_csv(output_dir / "相关矩阵.csv", encoding="utf-8-sig", float_format="%.3f") model_data = clean[["exam_score", *MODEL_PREDICTORS]].dropna() y = model_data["exam_score"].to_numpy(float) x_raw = model_data[MODEL_PREDICTORS].to_numpy(float) x_design = np.column_stack([np.ones(len(model_data)), x_raw]) beta, _, _, _ = np.linalg.lstsq(x_design, y, rcond=None) fitted = x_design @ beta ss_res = float(np.sum((y - fitted) ** 2)) ss_tot = float(np.sum((y - y.mean()) ** 2)) r_squared = 1.0 - ss_res / ss_tot x_z = (x_raw - x_raw.mean(axis=0)) / x_raw.std(axis=0, ddof=0) y_z = (y - y.mean()) / y.std(ddof=0) beta_z, _, _, _ = np.linalg.lstsq(x_z, y_z, rcond=None) coefficient_rows = [ { "变量": "截距", "非标准化系数": beta[0], "标准化系数": np.nan, "说明": "探索性线性回归；未计算显著性检验", } ] for index, predictor in enumerate(MODEL_PREDICTORS): coefficient_rows.append( { "变量": predictor, "非标准化系数": beta[index + 1], "标准化系数": beta_z[index], "说明": "控制表中其他变量后的条件关联，不能解释为因果", } ) coefficients = pd.DataFrame(coefficient_rows) coefficients["模型有效样本数"] = len(model_data) coefficients["模型R方"] = r_squared coefficients.to_csv(output_dir / "回归系数.csv", index=False, encoding="utf-8-sig", float_format="%.3f") write_scatter_svg(clean, output_dir / "睡眠时长与考试成绩.svg") summary = { "raw_rows": int(len(raw)), "raw_columns": int(len(raw.columns)), "duplicate_id_count": int(duplicate_ids.size), "missing_cells": int(raw.isna().sum().sum()), "unit_correction": {"variable": "weekday_sleep_hours", "records": corrected_ids, "rule": "60至1440之间的小时值按分钟除以60"}, "range_exclusion": {"variable": "exam_score", "records": excluded_exam_ids, "rule": "超出0至100的值在分析副本中设为缺失"}, "model_complete_cases": int(len(model_data)), "model_r_squared": round(r_squared, 4), "sleep_exam_correlation": round(float(correlations.loc["weekday_sleep_hours", "exam_score"]), 4), "note": "全部结果来自模拟教学数据，只作探索性演示。", } (output_dir / "分析摘要.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") assert raw_hash_before == sha256(input_path), "原始CSV在分析中发生变化" assert len(raw) == 300, "教学数据行数不是预期的300" assert corrected_ids == ["S073"], f"单位错误识别与材料说明不一致：{corrected_ids}" assert excluded_exam_ids == ["S214"], f"越界成绩识别与材料说明不一致：{excluded_exam_ids}" print(json.dumps(summary, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()