283 lines
10 KiB
Python
283 lines
10 KiB
Python
from __future__ import annotations
|
||
|
||
from collections import defaultdict
|
||
|
||
from .excel_parse import class_sort_key
|
||
from .kalman import (
|
||
fit_qr,
|
||
hamilton,
|
||
holdout_hits,
|
||
kalman_filter,
|
||
monte_carlo,
|
||
ranks_desc,
|
||
zscores,
|
||
)
|
||
|
||
|
||
TRACK_DEFAULTS = {
|
||
"物理": [{"id": "t88", "name": "特控线上", "k": 88}],
|
||
"历史": [
|
||
{"id": "top5", "name": "尖子生", "k": 5},
|
||
{"id": "top12", "name": "600分以上", "k": 12},
|
||
{"id": "top52", "name": "特控线上", "k": 52},
|
||
],
|
||
}
|
||
|
||
|
||
def infer_track_group(track: str) -> str:
|
||
t = track or ""
|
||
if "历史" in t:
|
||
return "历史"
|
||
if "物理" in t:
|
||
return "物理"
|
||
return t or "未分方向"
|
||
|
||
|
||
def default_targets(groups: list[str]) -> dict[str, list[dict]]:
|
||
out = {}
|
||
for g in groups:
|
||
if g in TRACK_DEFAULTS:
|
||
out[g] = [dict(x) for x in TRACK_DEFAULTS[g]]
|
||
else:
|
||
out[g] = [{"id": "top20", "name": "年级前20%", "k": 20}]
|
||
return out
|
||
|
||
|
||
def select_students(parsed: dict, settings: dict) -> list[dict]:
|
||
include_art = settings.get("include_art", False)
|
||
include_leave = settings.get("include_leave", False)
|
||
include_mingyuan = settings.get("include_mingyuan", True)
|
||
include_other = settings.get("include_other", True)
|
||
exams = settings.get("exams") or parsed["exams"]
|
||
selected = []
|
||
for s in parsed["students"]:
|
||
if s["art"] and not include_art:
|
||
continue
|
||
if s["leave"] and not include_leave:
|
||
continue
|
||
st = s["status"] or "在籍"
|
||
if st == "明远" and not include_mingyuan:
|
||
continue
|
||
if st not in ("在籍", "明远", "") and not include_other:
|
||
continue
|
||
scores = {e: s["scores"].get(e) for e in exams}
|
||
if all(v is None for v in scores.values()):
|
||
continue
|
||
row = dict(s)
|
||
row["scores"] = scores
|
||
row["group"] = infer_track_group(s["track"])
|
||
selected.append(row)
|
||
return selected
|
||
|
||
|
||
def _forecast_horizon(q: float, p: float, horizon: int) -> float:
|
||
return p + max(1, horizon) * q
|
||
|
||
|
||
def analyze(parsed: dict, settings: dict) -> dict:
|
||
exams = settings.get("exams") or parsed["exams"]
|
||
if not exams:
|
||
raise ValueError("没有可用的考试列")
|
||
horizon = int(settings.get("horizon", 1))
|
||
n_sim = int(settings.get("n_sim", 600))
|
||
selected = select_students(parsed, settings)
|
||
groups = sorted({s["group"] for s in selected})
|
||
targets = settings.get("targets") or default_targets(groups)
|
||
|
||
# fit Q,R per group on z-scores
|
||
group_models = {}
|
||
for g in groups:
|
||
members = [s for s in selected if s["group"] == g]
|
||
z_by_exam = {}
|
||
for e in exams:
|
||
z_by_exam[e] = zscores([s["scores"].get(e) for s in members])
|
||
for i, s in enumerate(members):
|
||
s["z"] = [z_by_exam[e][i] for e in exams]
|
||
s["ranks"] = []
|
||
for e in exams:
|
||
rk = ranks_desc([s["scores"].get(e) for s in members])
|
||
for i, s in enumerate(members):
|
||
s["ranks"].append(rk[i])
|
||
series = [s["z"] for s in members]
|
||
q, r, rmse = fit_qr(series)
|
||
group_models[g] = {"q": q, "r": r, "rmse": rmse, "n": len(members)}
|
||
for s in members:
|
||
fr = kalman_filter(s["z"], q, r)
|
||
s["kf_x"] = fr.x
|
||
s["kf_p"] = fr.p
|
||
s["kf_xs"] = fr.xs
|
||
s["kf_ps"] = fr.ps
|
||
if fr.x is None:
|
||
s["pred_mean"] = None
|
||
s["pred_var"] = None
|
||
else:
|
||
s["pred_mean"] = fr.x
|
||
s["pred_var"] = _forecast_horizon(q, fr.p if fr.p is not None else r, horizon)
|
||
|
||
results = []
|
||
for g in groups:
|
||
members = [s for s in selected if s["group"] == g]
|
||
g_targets = targets.get(g) or default_targets([g])[g]
|
||
model = group_models[g]
|
||
eligible = [s for s in members if s["pred_mean"] is not None]
|
||
last_ranks = [s["ranks"][-1] if s["ranks"] else None for s in members]
|
||
holdouts = []
|
||
if len(exams) >= 3:
|
||
series = [s["z"] for s in members]
|
||
true_ranks = [s["ranks"][-1] if s["ranks"] else None for s in members]
|
||
last_train_ranks = [s["ranks"][-2] if len(s["ranks"]) >= 2 else None for s in members]
|
||
for t in g_targets:
|
||
k = min(int(t["k"]), max(1, sum(1 for r in true_ranks if r is not None)))
|
||
h = holdout_hits(
|
||
series,
|
||
last_train_ranks,
|
||
true_ranks,
|
||
model["q"],
|
||
model["r"],
|
||
k,
|
||
)
|
||
h["target_id"] = t["id"]
|
||
h["target_name"] = t["name"]
|
||
h["k"] = k
|
||
holdouts.append(h)
|
||
|
||
target_blocks = []
|
||
for t in g_targets:
|
||
k = int(t["k"])
|
||
k = max(1, min(k, len(eligible)))
|
||
means = [s["pred_mean"] for s in eligible]
|
||
variances = [s["pred_var"] for s in eligible]
|
||
class_ids = [s["class_name"] for s in eligible]
|
||
probs, class_q = monte_carlo(means, variances, class_ids, k, n_sim=n_sim)
|
||
for s, p in zip(eligible, probs):
|
||
s.setdefault("probs", {})[t["id"]] = p
|
||
# current on-line: last exam rank <= k
|
||
class_now = defaultdict(int)
|
||
class_n = defaultdict(int)
|
||
class_exp = defaultdict(float)
|
||
class_mingyuan_exp = defaultdict(float)
|
||
for s in members:
|
||
class_n[s["class_name"]] += 1
|
||
for s in eligible:
|
||
class_exp[s["class_name"]] += s["probs"][t["id"]]
|
||
if s.get("status") == "明远":
|
||
class_mingyuan_exp[s["class_name"]] += s["probs"][t["id"]]
|
||
if s["ranks"] and s["ranks"][-1] is not None and s["ranks"][-1] <= k:
|
||
class_now[s["class_name"]] += 1
|
||
class_names = sorted(class_n.keys(), key=class_sort_key)
|
||
expected = [class_exp[c] for c in class_names]
|
||
assigned = hamilton(expected, k)
|
||
rows = []
|
||
for c, a in zip(class_names, assigned):
|
||
qtl = class_q.get(c, {"p20": 0, "p50": 0, "p80": 0, "mean": 0})
|
||
rows.append(
|
||
{
|
||
"class_name": c,
|
||
"n": class_n[c],
|
||
"current": class_now[c],
|
||
"expected": round(class_exp[c], 2),
|
||
"assigned": a,
|
||
"floor": int(qtl["p20"]),
|
||
"median": int(qtl["p50"]),
|
||
"stretch": int(qtl["p80"]),
|
||
"delta": a - class_now[c],
|
||
"mingyuan_expected": round(class_mingyuan_exp[c], 2),
|
||
}
|
||
)
|
||
critical = []
|
||
for s in eligible:
|
||
p = s["probs"][t["id"]]
|
||
if 0.2 <= p <= 0.8:
|
||
critical.append(
|
||
{
|
||
"sid": s["sid"],
|
||
"name": s["name"],
|
||
"class_name": s["class_name"],
|
||
"p": round(p, 3),
|
||
"last_rank": s["ranks"][-1] if s["ranks"] else None,
|
||
"status": s.get("status") or "",
|
||
}
|
||
)
|
||
critical.sort(key=lambda x: -x["p"])
|
||
target_blocks.append(
|
||
{
|
||
"id": t["id"],
|
||
"name": t["name"],
|
||
"k": k,
|
||
"classes": rows,
|
||
"critical_n": len(critical),
|
||
"critical": critical[:80],
|
||
"sum_expected": round(sum(expected), 2),
|
||
}
|
||
)
|
||
|
||
results.append(
|
||
{
|
||
"group": g,
|
||
"n": len(members),
|
||
"model": {
|
||
"q": model["q"],
|
||
"r": model["r"],
|
||
"rmse": round(model["rmse"], 3) if model["rmse"] is not None else None,
|
||
"horizon": horizon,
|
||
"n_sim": n_sim,
|
||
},
|
||
"holdout": holdouts,
|
||
"targets": target_blocks,
|
||
}
|
||
)
|
||
|
||
student_views = []
|
||
for s in selected:
|
||
student_views.append(
|
||
{
|
||
"sid": s["sid"],
|
||
"name": s["name"],
|
||
"class_name": s["class_name"],
|
||
"group": s["group"],
|
||
"track": s["track"],
|
||
"subjects": s["subjects"],
|
||
"status": s.get("status") or "",
|
||
"move": s.get("move") or "",
|
||
"art": s["art"],
|
||
"leave": s["leave"],
|
||
"scores": s["scores"],
|
||
"z": s.get("z"),
|
||
"ranks": s.get("ranks"),
|
||
"kf_xs": s.get("kf_xs"),
|
||
"kf_ps": s.get("kf_ps"),
|
||
"pred_mean": s.get("pred_mean"),
|
||
"pred_var": s.get("pred_var"),
|
||
"probs": {k: round(v, 4) for k, v in (s.get("probs") or {}).items()},
|
||
}
|
||
)
|
||
|
||
return {
|
||
"exams": exams,
|
||
"horizon": horizon,
|
||
"n_selected": len(selected),
|
||
"n_total": len(parsed["students"]),
|
||
"groups": results,
|
||
"students": student_views,
|
||
}
|
||
|
||
|
||
def class_comment(row: dict, target_name: str) -> str:
|
||
assigned = row["assigned"]
|
||
current = row["current"]
|
||
expected = row["expected"]
|
||
floor = row["floor"]
|
||
stretch = row["stretch"]
|
||
name = row["class_name"]
|
||
bits = [
|
||
f"{name}班{target_name}下达 {assigned} 人（滤波期望 {expected:.1f}）。"
|
||
]
|
||
if assigned > current:
|
||
bits.append(f"比最近一次在线人数多 {assigned - current} 人，班级里有上升轨迹的临界生，需要把潜力兑现。")
|
||
elif assigned < current:
|
||
bits.append(f"比最近一次在线人数少 {current - assigned} 人，说明部分当前在线学生轨迹不稳，重点是保住临界生。")
|
||
else:
|
||
bits.append("与最近一次在线人数持平，按现有轨迹属于维持性目标。")
|
||
bits.append(f"模拟分布大约在 {floor}–{stretch} 人之间，超越下达人数才算跑赢预估。")
|
||
return "".join(bits)
|