lofyerandfactory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com>
71db82393a
Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com>
138 lines
5.2 KiB
Python
138 lines
5.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
测试脚本：完整Pipeline - 数据加载→赋分→PCA→标准化→水平判定→统计分析
|
|
"""
|
|
import sys
|
|
from pathlib import Path
|
|
import json
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "backend"))
|
|
|
|
from app.engines.data_engine import DataEngine
|
|
from app.engines.scoring_engine import ScoringEngine
|
|
from app.engines.stats_engine import StatsEngine
|
|
from app.config import DIMENSION_FRAMEWORK, LEVEL_THRESHOLDS
|
|
import logging
|
|
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
|
|
|
|
|
|
def main():
|
|
print("=" * 80)
|
|
print("完整Pipeline测试：Excel → 赋分 → PCA → 标准化 → 水平判定")
|
|
print("=" * 80)
|
|
|
|
# Step 1: 加载数据
|
|
print("\n[Step 1] 加载Excel数据...")
|
|
data_engine = DataEngine()
|
|
data_engine.load_all()
|
|
|
|
# Step 2: 赋分
|
|
print("\n[Step 2] 对所有学校进行赋分...")
|
|
scoring_engine = ScoringEngine(data_engine)
|
|
raw_scores = scoring_engine.score_all_schools()
|
|
|
|
# 打印赋分结果概要
|
|
print("\n📊 赋分结果概要:")
|
|
for school in data_engine.schools:
|
|
scores = raw_scores[school]
|
|
print(f"\n 🏫 {school}:")
|
|
for dim, vals in scores.items():
|
|
avg = sum(vals) / len(vals) if vals else 0
|
|
print(f" {dim}: {len(vals)}个指标, 均值={avg:.2f}, 值={[round(v,1) for v in vals[:5]]}{'...' if len(vals) > 5 else ''}")
|
|
|
|
# Step 3: 统计分析
|
|
print("\n[Step 3] PCA合成 + 标准化...")
|
|
stats_engine = StatsEngine()
|
|
sub_scores = stats_engine.compute_dimension_scores(raw_scores)
|
|
print(f"\n三级维度标准化得分 (均值50, 标准差10):")
|
|
print(sub_scores.round(2).to_string())
|
|
|
|
# Step 4: 二级维度聚合
|
|
print("\n[Step 4] 二级维度聚合...")
|
|
dim_scores = stats_engine.compute_dimension_aggregates(sub_scores)
|
|
print(f"\n二级维度得分:")
|
|
print(dim_scores.round(2).to_string())
|
|
|
|
# Step 5: 水平判定
|
|
print("\n[Step 5] 水平判定...")
|
|
levels = stats_engine.compute_levels(sub_scores)
|
|
print(f"\n各学校各维度水平:")
|
|
print(levels.to_string())
|
|
|
|
# Step 6: 聚类分析
|
|
print("\n[Step 6] 聚类分析...")
|
|
dim_cols = [c for c in dim_scores.columns if c in DIMENSION_FRAMEWORK]
|
|
clusters = stats_engine.cluster_analysis(dim_scores[dim_cols])
|
|
print(f"\n学校聚类结果:")
|
|
for school, cluster in clusters["school_clusters"].items():
|
|
print(f" {school}: {cluster}")
|
|
|
|
# Step 7: 相关性
|
|
print("\n[Step 7] 维度间相关性:")
|
|
corr = stats_engine.correlation_analysis(dim_scores)
|
|
print(corr.round(3).to_string())
|
|
|
|
# Step 8: 生成一所学校的完整报告数据
|
|
test_school = "延安中学"
|
|
print(f"\n[Step 8] 生成 {test_school} 的完整报告数据包...")
|
|
report_data = stats_engine.compute_school_report_data(test_school, sub_scores, dim_scores)
|
|
|
|
print(f"\n📋 {test_school} 报告数据概要:")
|
|
print(f" 总体得分: {report_data['overall']['score']}")
|
|
print(f" 区内排名: {report_data['overall']['rank_in_district']}/{report_data['overall']['total_schools']}")
|
|
print(f" 聚类类型: {report_data['overall']['cluster']}")
|
|
|
|
print(f"\n 二级维度得分:")
|
|
for dim, data in report_data['dimensions'].items():
|
|
print(f" {dim}: {data['score']} (区均值{data['district_avg']}, 差异{data['diff_district']:+.2f})")
|
|
|
|
print(f"\n 三级维度水平:")
|
|
for sub_dim, data in report_data['sub_dimensions'].items():
|
|
print(f" {sub_dim}: 得分{data['score']}, 水平{data['level']}, {data['level_description'][:30]}")
|
|
|
|
# 保存报告数据为JSON
|
|
output_dir = Path(__file__).parent.parent / "output"
|
|
output_dir.mkdir(exist_ok=True)
|
|
|
|
# 清理numpy类型以便JSON序列化
|
|
def clean_for_json(obj):
|
|
import numpy as np
|
|
if isinstance(obj, dict):
|
|
return {k: clean_for_json(v) for k, v in obj.items()}
|
|
elif isinstance(obj, list):
|
|
return [clean_for_json(v) for v in obj]
|
|
elif isinstance(obj, (np.integer,)):
|
|
return int(obj)
|
|
elif isinstance(obj, (np.floating,)):
|
|
return round(float(obj), 4)
|
|
elif isinstance(obj, np.ndarray):
|
|
return obj.tolist()
|
|
elif isinstance(obj, float):
|
|
return round(obj, 4)
|
|
return obj
|
|
|
|
report_clean = clean_for_json(report_data)
|
|
with open(output_dir / f"{test_school}_report_data.json", "w", encoding="utf-8") as f:
|
|
json.dump(report_clean, f, ensure_ascii=False, indent=2)
|
|
print(f"\n✅ 报告数据已保存到 output/{test_school}_report_data.json")
|
|
|
|
# 保存所有学校的得分汇总
|
|
all_scores = {
|
|
"sub_dimension_scores": clean_for_json(sub_scores.to_dict()),
|
|
"dimension_scores": clean_for_json(dim_scores.to_dict()),
|
|
"levels": clean_for_json(levels.to_dict()),
|
|
"clusters": clean_for_json(clusters),
|
|
}
|
|
with open(output_dir / "all_schools_scores.json", "w", encoding="utf-8") as f:
|
|
json.dump(all_scores, f, ensure_ascii=False, indent=2)
|
|
print(f"✅ 全校得分汇总已保存到 output/all_schools_scores.json")
|
|
|
|
print("\n" + "=" * 80)
|
|
print("✅ 完整Pipeline测试完成!")
|
|
print("=" * 80)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|