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>
375 lines
15 KiB
Python
375 lines
15 KiB
Python
"""
|
|
二期统计引擎 (era2)
|
|
支持全市基准标准化 + 同类学校均值计算
|
|
"""
|
|
import pandas as pd
|
|
import numpy as np
|
|
from typing import Dict, List, Optional, Tuple
|
|
from sklearn.decomposition import PCA
|
|
from sklearn.preprocessing import StandardScaler
|
|
from sklearn.cluster import KMeans
|
|
from scipy import stats
|
|
import logging
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
from config_era2 import (
|
|
PCA_MEAN, PCA_STD, LEVEL_THRESHOLDS, DIMENSION_FRAMEWORK, SUBJECTS,
|
|
SCHOOL_TYPE_MAP,
|
|
)
|
|
try:
|
|
from config_era2 import CLUSTER_CONFIG
|
|
except ImportError:
|
|
CLUSTER_CONFIG = {}
|
|
try:
|
|
from config_era2 import LEVEL_DESCRIPTIONS
|
|
except ImportError:
|
|
LEVEL_DESCRIPTIONS = {}
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class StatsEngineEra2:
|
|
"""二期统计引擎"""
|
|
|
|
def __init__(self):
|
|
self._dimension_scores: Optional[pd.DataFrame] = None
|
|
self._sub_dimension_scores: Optional[pd.DataFrame] = None
|
|
|
|
def standardize_scores(self, raw_scores: np.ndarray) -> np.ndarray:
|
|
if len(raw_scores) < 2:
|
|
return np.full_like(raw_scores, PCA_MEAN, dtype=float)
|
|
mean = np.nanmean(raw_scores)
|
|
std = np.nanstd(raw_scores, ddof=1)
|
|
if std == 0 or np.isnan(std):
|
|
return np.full_like(raw_scores, PCA_MEAN, dtype=float)
|
|
return (raw_scores - mean) / std * PCA_STD + PCA_MEAN
|
|
|
|
def pca_compose(self, data_matrix: pd.DataFrame) -> np.ndarray:
|
|
filled = data_matrix.fillna(data_matrix.mean())
|
|
if filled.shape[1] == 0:
|
|
return np.full(filled.shape[0], PCA_MEAN)
|
|
if filled.shape[1] == 1:
|
|
return self.standardize_scores(filled.iloc[:, 0].values)
|
|
scaler = StandardScaler()
|
|
scaled = scaler.fit_transform(filled)
|
|
n_components = min(1, filled.shape[1], filled.shape[0])
|
|
pca = PCA(n_components=n_components)
|
|
scores = pca.fit_transform(scaled)[:, 0]
|
|
loadings = pca.components_[0]
|
|
if np.sum(loadings) < 0:
|
|
scores = -scores
|
|
return self.standardize_scores(scores)
|
|
|
|
def determine_level(self, score: float, dimension: str) -> int:
|
|
thresholds = LEVEL_THRESHOLDS.get(dimension, {})
|
|
if not thresholds:
|
|
if score > 55: return 4
|
|
elif score > 50: return 3
|
|
elif score > 45: return 2
|
|
else: return 1
|
|
if score > thresholds["level4"]: return 4
|
|
elif score > thresholds["level3"]: return 3
|
|
elif score > thresholds["level2"]: return 2
|
|
else: return 1
|
|
|
|
def get_level_description(self, dimension: str, level: int) -> str:
|
|
descriptions = LEVEL_DESCRIPTIONS.get(dimension, {})
|
|
return descriptions.get(level, f"水平{level}")
|
|
|
|
def compute_dimension_scores(self, school_raw_scores: Dict[str, Dict[str, List[float]]]) -> pd.DataFrame:
|
|
"""原始赋分 → nanmean → z-score标准化（旧方法，兼容保留）"""
|
|
schools = list(school_raw_scores.keys())
|
|
all_sub_dims = []
|
|
for dim, info in DIMENSION_FRAMEWORK.items():
|
|
all_sub_dims.extend(info["sub_dimensions"])
|
|
|
|
raw_matrix = {}
|
|
for sub_dim in all_sub_dims:
|
|
values = []
|
|
for school in schools:
|
|
scores = school_raw_scores.get(school, {}).get(sub_dim, [])
|
|
values.append(np.nanmean(scores) if scores else np.nan)
|
|
raw_matrix[sub_dim] = values
|
|
|
|
raw_df = pd.DataFrame(raw_matrix, index=schools)
|
|
|
|
result = pd.DataFrame(index=schools)
|
|
for sub_dim in all_sub_dims:
|
|
if sub_dim in raw_df.columns:
|
|
result[sub_dim] = self.standardize_scores(raw_df[sub_dim].values)
|
|
else:
|
|
result[sub_dim] = PCA_MEAN
|
|
|
|
self._sub_dimension_scores = result
|
|
return result
|
|
|
|
def compute_dimension_scores_pca(self, pca_sub_scores: pd.DataFrame) -> pd.DataFrame:
|
|
"""
|
|
接受PCA引擎的输出（已经是子维度得分的DataFrame），直接使用。
|
|
PCA引擎内部已完成 Z标准化 → PCA → ×10+50 的全流程。
|
|
|
|
Args:
|
|
pca_sub_scores: PcaScoringEngineEra2.compute_all() 的输出
|
|
DataFrame, index=学校, columns=子维度名
|
|
Returns:
|
|
与 compute_dimension_scores 相同格式的 DataFrame
|
|
"""
|
|
all_sub_dims = []
|
|
for dim, info in DIMENSION_FRAMEWORK.items():
|
|
all_sub_dims.extend(info["sub_dimensions"])
|
|
|
|
result = pd.DataFrame(index=pca_sub_scores.index)
|
|
for sub_dim in all_sub_dims:
|
|
if sub_dim in pca_sub_scores.columns:
|
|
result[sub_dim] = pca_sub_scores[sub_dim]
|
|
else:
|
|
result[sub_dim] = PCA_MEAN
|
|
|
|
self._sub_dimension_scores = result
|
|
return result
|
|
|
|
def compute_dimension_aggregates(self, sub_scores: pd.DataFrame) -> pd.DataFrame:
|
|
result = pd.DataFrame(index=sub_scores.index)
|
|
for dim, info in DIMENSION_FRAMEWORK.items():
|
|
sub_dims = [s for s in info["sub_dimensions"] if s in sub_scores.columns]
|
|
if sub_dims:
|
|
result[dim] = sub_scores[sub_dims].mean(axis=1)
|
|
else:
|
|
result[dim] = PCA_MEAN
|
|
result["总体得分"] = result[list(DIMENSION_FRAMEWORK.keys())].mean(axis=1)
|
|
self._dimension_scores = result
|
|
return result
|
|
|
|
def compute_levels(self, sub_scores: pd.DataFrame) -> pd.DataFrame:
|
|
levels = pd.DataFrame(index=sub_scores.index)
|
|
for col in sub_scores.columns:
|
|
levels[col] = sub_scores[col].apply(lambda x: self.determine_level(x, col))
|
|
return levels
|
|
|
|
def cluster_analysis(self, scores: pd.DataFrame, n_clusters: int = 2,
|
|
dimension_name: Optional[str] = None) -> Dict:
|
|
"""
|
|
K-means 聚类分析。
|
|
|
|
Args:
|
|
scores: 学校×子维度 的得分 DataFrame
|
|
n_clusters: 聚类数（默认2，可通过 CLUSTER_CONFIG 覆盖）
|
|
dimension_name: 维度名称，用于从 CLUSTER_CONFIG 查询聚类数
|
|
"""
|
|
# 从配置覆盖聚类数
|
|
if dimension_name and dimension_name in CLUSTER_CONFIG:
|
|
n_clusters = CLUSTER_CONFIG[dimension_name]
|
|
|
|
scaler = StandardScaler()
|
|
scaled = scaler.fit_transform(scores.fillna(PCA_MEAN))
|
|
n_clusters = min(n_clusters, len(scores))
|
|
if n_clusters < 2:
|
|
return {"labels": [0] * len(scores), "centers": scores.values.tolist()}
|
|
kmeans = KMeans(n_clusters=n_clusters, random_state=42, n_init=10)
|
|
labels = kmeans.fit_predict(scaled)
|
|
|
|
cluster_means = {}
|
|
for c in range(n_clusters):
|
|
mask = labels == c
|
|
cluster_means[c] = scores[mask].mean().to_dict()
|
|
|
|
avg_per_cluster = {c: np.mean(list(v.values())) for c, v in cluster_means.items()}
|
|
sorted_clusters = sorted(avg_per_cluster.items(), key=lambda x: x[1], reverse=True)
|
|
|
|
# 命名策略：2类 → 较好/待提升，3类 → 较好/中等/待提升
|
|
cluster_names = {}
|
|
if n_clusters == 2:
|
|
name_list = ["较好", "待提升"]
|
|
elif n_clusters == 3:
|
|
name_list = ["较好", "中等", "待提升"]
|
|
else:
|
|
name_list = [f"第{i+1}类" for i in range(n_clusters)]
|
|
|
|
for rank, (c, _) in enumerate(sorted_clusters):
|
|
cluster_names[c] = name_list[rank] if rank < len(name_list) else f"第{rank+1}类"
|
|
|
|
return {
|
|
"labels": labels.tolist(),
|
|
"n_clusters": n_clusters,
|
|
"school_clusters": {
|
|
school: cluster_names[labels[i]]
|
|
for i, school in enumerate(scores.index)
|
|
},
|
|
"cluster_means": cluster_means,
|
|
"cluster_names": cluster_names,
|
|
}
|
|
|
|
def t_test_vs_mean(self, school_scores: np.ndarray, ref_mean: float) -> Dict:
|
|
scores = school_scores[~np.isnan(school_scores)]
|
|
if len(scores) < 2:
|
|
return {"t": np.nan, "p": np.nan, "significant": False, "n": len(scores)}
|
|
t_stat, p_value = stats.ttest_1samp(scores, ref_mean)
|
|
return {
|
|
"t": round(float(t_stat), 3),
|
|
"p": round(float(p_value), 4),
|
|
"significant": float(p_value) < 0.05,
|
|
"n": len(scores),
|
|
}
|
|
|
|
def correlation_analysis(self, dim_scores: pd.DataFrame) -> pd.DataFrame:
|
|
dim_cols = [c for c in dim_scores.columns if c in DIMENSION_FRAMEWORK]
|
|
return dim_scores[dim_cols].corr()
|
|
|
|
def compute_school_report_data(self, school: str,
|
|
sub_scores: pd.DataFrame,
|
|
dim_scores: pd.DataFrame,
|
|
district_schools: Optional[List[str]] = None) -> Dict:
|
|
"""
|
|
生成单校报告数据。
|
|
|
|
sub_scores / dim_scores: 全市所有学校的标准化分数（全市基准）
|
|
district_schools: 该学校所在区的学校列表（用于计算区均值和区内排名）
|
|
如果为None，则用sub_scores中所有学校
|
|
"""
|
|
all_schools = list(sub_scores.index)
|
|
if school not in all_schools:
|
|
raise ValueError(f"学校 '{school}' 不在数据中")
|
|
|
|
# 确定区内学校列表
|
|
if district_schools is None:
|
|
district_schools = all_schools
|
|
district_schools = [s for s in district_schools if s in all_schools]
|
|
|
|
# 区内分数切片
|
|
dist_sub = sub_scores.loc[district_schools]
|
|
dist_dim = dim_scores.loc[district_schools]
|
|
|
|
# 区均值
|
|
district_avg_sub = dist_sub.mean()
|
|
district_avg_dim = dist_dim.mean()
|
|
|
|
# 同类学校均值（从全市SCHOOL_TYPE_MAP中找同类型学校）
|
|
school_info_data = SCHOOL_TYPE_MAP.get(school, {})
|
|
school_type = school_info_data.get("type", "")
|
|
same_type_schools = [
|
|
s for s in all_schools
|
|
if SCHOOL_TYPE_MAP.get(s, {}).get("type") == school_type and school_type
|
|
]
|
|
if len(same_type_schools) >= 2:
|
|
same_type_avg_sub = sub_scores.loc[same_type_schools].mean()
|
|
same_type_avg_dim = dim_scores.loc[same_type_schools].mean()
|
|
else:
|
|
same_type_avg_sub = district_avg_sub
|
|
same_type_avg_dim = district_avg_dim
|
|
|
|
# 构建 school_info（从 SCHOOL_TYPE_MAP 获取）
|
|
school_info = {}
|
|
if school_info_data:
|
|
school_info = {
|
|
"type": school_info_data.get("type", ""),
|
|
"code": school_info_data.get("code", ""),
|
|
"nature": school_info_data.get("nature", ""),
|
|
"feature": school_info_data.get("area", ""), # 所处地区作为feature
|
|
}
|
|
|
|
levels = self.compute_levels(sub_scores)
|
|
|
|
# 聚类只在区内做
|
|
dim_cols = [c for c in dim_scores.columns if c in DIMENSION_FRAMEWORK]
|
|
overall_cluster = self.cluster_analysis(dist_dim[dim_cols], dimension_name="总体")
|
|
|
|
dim_clusters = {}
|
|
for dim, info in DIMENSION_FRAMEWORK.items():
|
|
sub_dims = [s for s in info["sub_dimensions"] if s in sub_scores.columns]
|
|
if sub_dims:
|
|
dim_clusters[dim] = self.cluster_analysis(dist_sub[sub_dims], dimension_name=dim)
|
|
|
|
correlation = self.correlation_analysis(dist_dim)
|
|
|
|
# 区内排名
|
|
school_score = float(dist_dim.loc[school, "总体得分"])
|
|
rank_in_district = int((dist_dim["总体得分"] >= school_score).sum())
|
|
|
|
# 全市排名（不分类型，所有参与监测的学校）
|
|
rank_in_city = int((dim_scores["总体得分"] >= school_score).sum())
|
|
total_schools_in_city = len(all_schools)
|
|
|
|
# 全市同类排名
|
|
if len(same_type_schools) >= 2:
|
|
st_dim = dim_scores.loc[same_type_schools]
|
|
rank_in_same_type = int((st_dim["总体得分"] >= school_score).sum())
|
|
total_same_type = len(same_type_schools)
|
|
else:
|
|
rank_in_same_type = rank_in_district
|
|
total_same_type = len(district_schools)
|
|
|
|
report = {
|
|
"school": school,
|
|
"school_info": school_info,
|
|
"overall": {
|
|
"score": round(school_score, 2),
|
|
"district_avg": round(float(district_avg_dim["总体得分"]), 2),
|
|
"same_type_avg": round(float(same_type_avg_dim.get("总体得分", PCA_MEAN)), 2),
|
|
"rank_in_district": rank_in_district,
|
|
"total_schools": len(district_schools),
|
|
"rank_in_city": rank_in_city,
|
|
"total_schools_in_city": total_schools_in_city,
|
|
"cluster": overall_cluster["school_clusters"].get(school, ""),
|
|
"school_type": school_type,
|
|
"same_type_count": total_same_type,
|
|
"rank_in_same_type": rank_in_same_type,
|
|
},
|
|
"dimensions": {},
|
|
"sub_dimensions": {},
|
|
"correlation": correlation.to_dict(),
|
|
"all_schools_dim_scores": dist_dim.to_dict(),
|
|
"all_schools_sub_scores": dist_sub.to_dict(),
|
|
}
|
|
|
|
for dim in DIMENSION_FRAMEWORK:
|
|
score = float(dist_dim.loc[school, dim])
|
|
d_avg = float(district_avg_dim[dim])
|
|
st_avg = float(same_type_avg_dim.get(dim, PCA_MEAN))
|
|
|
|
sub_dims = DIMENSION_FRAMEWORK[dim]["sub_dimensions"]
|
|
sub_vals = np.array([float(sub_scores.loc[school, s]) for s in sub_dims if s in sub_scores.columns])
|
|
t_test = self.t_test_vs_mean(sub_vals, d_avg)
|
|
|
|
report["dimensions"][dim] = {
|
|
"score": round(score, 2),
|
|
"district_avg": round(d_avg, 2),
|
|
"same_type_avg": round(st_avg, 2),
|
|
"diff_district": round(score - d_avg, 2),
|
|
"rank_in_district": int((dist_dim[dim] >= score).sum()),
|
|
"rank_in_city": int((dim_scores[dim] >= score).sum()),
|
|
"t_test_vs_district": t_test,
|
|
"cluster": dim_clusters.get(dim, {}).get("school_clusters", {}).get(school, ""),
|
|
}
|
|
|
|
for dim, info in DIMENSION_FRAMEWORK.items():
|
|
for sub_dim in info["sub_dimensions"]:
|
|
if sub_dim not in sub_scores.columns:
|
|
continue
|
|
score = float(sub_scores.loc[school, sub_dim])
|
|
d_avg = float(district_avg_sub[sub_dim])
|
|
level = int(levels.loc[school, sub_dim])
|
|
# 区内排名
|
|
rank = int((dist_sub[sub_dim] >= score).sum())
|
|
# 全市排名
|
|
rank_city = int((sub_scores[sub_dim] >= score).sum())
|
|
# 区内水平分布
|
|
dist_levels = levels.loc[district_schools]
|
|
dim_levels = dist_levels[sub_dim]
|
|
level_dist = {f"水平{i}": int((dim_levels == i).sum()) for i in range(1, 5)}
|
|
|
|
report["sub_dimensions"][sub_dim] = {
|
|
"parent_dimension": dim,
|
|
"score": round(score, 2),
|
|
"district_avg": round(d_avg, 2),
|
|
"diff_district": round(score - d_avg, 2),
|
|
"rank_in_district": rank,
|
|
"rank_in_city": rank_city,
|
|
"level": level,
|
|
"level_description": self.get_level_description(sub_dim, level),
|
|
"level_distribution": level_dist,
|
|
}
|
|
|
|
return report
|