diff --git a/README.md b/README.md index b446c7e..768fabf 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ -# 航标 · 班级达线指标预估 +# 指标测算 · 相邻层次法 -B/S 系统：管理员上传历次大考总分，系统用卡尔曼滤波估计每个学生的稳定水平，模拟后续考试的年级排名，把达线概率加总后下达各班指标。下达人数是“按现有轨迹本来会到的人数”，班级工作是超越它。 +B/S 系统：上传历次大考总分，按方向内名次把学生分到固定宽度层次；每次只比较相邻两次考试的“上一层次 → 本次到线”命中人数；汇总后做拟合概率，再用最后一次考试各班层次人数摊派尖子线和特控线指标。 ## 启动 @@ -11,29 +11,37 @@ pip install -r requirements.txt python3 -m app ``` -浏览器打开 http://服务器IP:8765 （本机也可用 http://127.0.0.1:8765） -服务默认监听 `0.0.0.0:8765`。可用环境变量覆盖：`HOST=127.0.0.1 PORT=8765 ./run.sh` +浏览器打开 http://127.0.0.1:8765 。服务默认监听 `0.0.0.0:8765`，可用 `HOST`、`PORT` 覆盖。 -默认账号：`admin` / `hangbiao2026`（请在 `app/config.py` 修改） +默认账号：`admin` / `hangbiao2026`（可在 `app/config.py` 修改）。 ## 成绩表要求 - Excel `.xlsx` -- 至少包含：学号或姓名、班级、多次考试总分 -- 建议包含：方向（物理/历史）、选科、学籍、异动 -- 物理/历史分表、艺体表、异动表都可以；系统会识别并在第二步让你勾选口径 +- 至少包含：学号/姓名、班级、多次考试总分 +- 建议包含：方向、选科、学籍、异动 +- 物理/历史分表可直接识别；同分按同名次，后续名次递增 +- 未排除的缺考学生按 0 分参加同方向排名 -同目录的 `2024级指标分析成绩基础数据_5次考试.xlsx` 可在首页一键演示。 +同目录 `basedata.xlsx` 上传后可作为按新算法复算的示例。 -## 四步分析 +## Web 设置 -1. 数据体检：读到哪些人、哪些考试、分数尺度 -2. 口径与目标：算谁、年级总盘子、预估步长 -3. 滤波与回测：Q/R 含义，以及用历史考试验证准不准 -4. 下达指标：班级期望 / 下达 / 保底 / 冲刺，点班级看学生，点学生看轨迹 +- 勾选是否排除异动学生（异动/备注非空，或来源为异动/艺体表） +- 修改每个方向的尖子线人数、特控线人数、每层人数 +- 选择参与计算哪些考试；只比较勾选后的相邻两次 +- 第三步查看每次相邻命中概率、汇总概率和拟合概率；最后一次考试既参与相邻转移计算，也用于最终班级摊派 +- 第四步查看班级摊派，并可导出总表 -## 指标怎么理解 +## 总表结构 -- **下达指标**：滤波期望取整，合计等于年级目标人数 -- **保底 / 冲刺**：模拟分布大约 20% / 80% 分位 -- **相对最近一次**：预估相对最近一场已在线人数的增减，用来找上升生和不稳的在线生 +导出的 Excel 包含： + +1. 物理方向人员 +2. 历史方向人员 +3. 物理类概率分析 +4. 历史类概率分析 +5. 物理指标数 +6. 历史指标数 + +结构与 `basedata.xlsx` 对应；概率和指标为本系统按当前 Web 口径重新计算后的结果。 diff --git a/app/config.py b/app/config.py index 6a12aef..c66a94e 100644 --- a/app/config.py +++ b/app/config.py @@ -4,7 +4,7 @@ BASE_DIR = Path(__file__).resolve().parent.parent DATA_DIR = BASE_DIR / "data" UPLOAD_DIR = DATA_DIR / "uploads" DB_PATH = DATA_DIR / "hangbiao.db" -SAMPLE_XLSX = BASE_DIR / "2024级指标分析成绩基础数据_5次考试.xlsx" +SAMPLE_XLSX = BASE_DIR / "basedata.xlsx" SECRET_KEY = "hangbiao-dev-secret-change-me" ADMIN_USER = "admin" diff --git a/app/excel_parse.py b/app/excel_parse.py index 9171b70..8435aa0 100644 --- a/app/excel_parse.py +++ b/app/excel_parse.py @@ -22,6 +22,14 @@ EXAM_HINT = re.compile( re.I, ) SKIP_SHEETS = ("说明", "readme", "目录") + + +def _is_skip_sheet(name: str) -> bool: + n = name.lower() + if any(x in n for x in SKIP_SHEETS): + return True + return "概率分析" in n or "指标数" in n + ART_RE = re.compile(r"艺术|体育|艺体") LEAVE_RE = re.compile(r"请假|出国|休学|国际|不在校|港澳|港籍.*假") @@ -76,6 +84,8 @@ def parse_workbook(path: Path | BytesIO) -> dict: for ws in wb.worksheets: kind = _sheet_kind(ws.title) + if _is_skip_sheet(ws.title): + continue rows = ws.iter_rows(values_only=True) try: header_row = next(rows) @@ -201,7 +211,7 @@ def summarize_parse(parsed: dict) -> dict: tracks = {} classes = {} status = {} - art = leave = 0 + art = leave = moves = 0 exam_present = {e: 0 for e in exams} for s in students: tracks[s["track"] or "未分方向"] = tracks.get(s["track"] or "未分方向", 0) + 1 @@ -211,6 +221,8 @@ def summarize_parse(parsed: dict) -> dict: art += 1 if s["leave"]: leave += 1 + if (s.get("move") or "").strip() or s.get("sheet_kind") in ("art", "leave"): + moves += 1 for e in exams: if s["scores"].get(e) is not None: exam_present[e] += 1 @@ -236,6 +248,7 @@ def summarize_parse(parsed: dict) -> dict: "status": status, "art": art, "leave": leave, + "moves": moves, "exam_present": exam_present, "score_scale": score_scale, "sheets": parsed["sheets"], diff --git a/app/export_xlsx.py b/app/export_xlsx.py index 2a3c921..8a2bee4 100644 --- a/app/export_xlsx.py +++ b/app/export_xlsx.py @@ -3,80 +3,208 @@ from __future__ import annotations from io import BytesIO from openpyxl import Workbook -from openpyxl.styles import Alignment, Font, PatternFill, Border, Side +from openpyxl.styles import Alignment, Border, Font, PatternFill, Side +from openpyxl.utils import get_column_letter + + +def _style_header(ws, row: int, ncols: int): + fill = PatternFill("solid", fgColor="12233D") + font = Font(color="F4EFE6", bold=True) + for c in range(1, ncols + 1): + cell = ws.cell(row, c) + cell.fill = fill + cell.font = font + cell.alignment = Alignment(wrap_text=True, vertical="center") + + +def _borders(ws, min_row: int = 1, max_row: int | None = None, max_col: int | None = None): + thin = Side(style="thin", color="D9D1C3") + border = Border(left=thin, right=thin, top=thin, bottom=thin) + max_row = max_row or ws.max_row + max_col = max_col or ws.max_column + for row in ws.iter_rows(min_row=min_row, max_row=max_row, min_col=1, max_col=max_col): + for cell in row: + if cell.value is not None: + cell.border = border + + +def _group_order(g: str): + return 0 if g == "物理" else 1 if g == "历史" else 2 + + +def _safe_title(text: str) -> str: + return text.replace(":", "-").replace("/", "-").replace("\\", "-").replace("?", "-")[:31] + + +def _raw_sheet(wb: Workbook, group: dict, students: list[dict]): + title = _safe_title(f"{group['group']}方向人员") + ws = wb.create_sheet(title) + exams = group["_exams"] + ws.append(["学号", "姓名", "班级", "方向", "选科", "学籍", "异动"] + exams) + _style_header(ws, 1, ws.max_column) + for s in students: + ws.append( + [ + s["sid"], s["name"], s["class_name"], s["track"], s.get("subjects") or "", + s.get("status") or "", s.get("move") or "", + ] + + [("缺考" if s["scores"].get(e) is None else s["scores"].get(e)) for e in exams] + ) + _borders(ws, 2) + ws.freeze_panes = "A2" + return ws + + +def _probability_sheet(wb: Workbook, group: dict): + title = _safe_title(f"{group['group']}类概率分析") + ws = wb.create_sheet(title) + exams = group["_exams"] + lines = {x["id"]: x for x in group["targets"]} + pair_blocks = {x["line"]: x["pairs"] for x in group["pairs"]} + r = 1 + + def pair_for(line_id: str, a: str, b: str): + line_name = lines[line_id]["name"] + return next( + (p for p in pair_blocks[line_name] if p["from_exam"] == a and p["to_exam"] == b), + None, + ) + + for idx, (a, b) in enumerate(zip(exams, exams[1:]), 1): + top_pair = pair_for("top", a, b) + control_pair = pair_for("control", a, b) + if not top_pair or not control_pair: + continue + ws.cell(r, 1, f"情况{idx}·尖子线前{lines['top']['k']}名") + ws.cell(r, 8, f"情况{idx}·特控线前{lines['control']['k']}名") + r += 1 + for c0, line_id in ((1, "top"), (8, "control")): + ws.cell(r, c0, "分层") + ws.cell(r, c0 + 1, "排名区间") + ws.cell(r, c0 + 2, a) + ws.cell(r, c0 + 3, b) + ws.cell(r, c0 + 4, "概率") + _style_header(ws, r, 12) + r += 1 + max_layers = max(len(top_pair["layers"]), len(control_pair["layers"])) + for i in range(max_layers): + for c0, pair in ((1, top_pair), (8, control_pair)): + if i >= len(pair["layers"]): + continue + x = pair["layers"][i] + ws.cell(r, c0, x["layer"]) + ws.cell(r, c0 + 1, x.get("label") or f"第{x['layer']}层") + ws.cell(r, c0 + 2, x["previous_n"]) + ws.cell(r, c0 + 3, x["hit_n"]) + ws.cell(r, c0 + 4, x["probability"]) + r += 1 + r += 2 + + ws.cell(r, 1, f"汇总·尖子线前{lines['top']['k']}名") + ws.cell(r, 8, f"汇总·特控线前{lines['control']['k']}名") + r += 1 + for c0, line_id in ((1, "top"), (8, "control")): + ws.cell(r, c0, "分层") + ws.cell(r, c0 + 1, "排名区间") + ws.cell(r, c0 + 2, "汇总样本") + ws.cell(r, c0 + 3, "汇总命中") + ws.cell(r, c0 + 4, "合成概率") + ws.cell(r, c0 + 5, "拟合概率") + _style_header(ws, r, 13) + r += 1 + top_layers = group["summary"]["top"]["layers"] + control_layers = group["summary"]["control"]["layers"] + for i in range(max(len(top_layers), len(control_layers))): + for c0, layers in ((1, top_layers), (8, control_layers)): + if i >= len(layers): + continue + x = layers[i] + ws.cell(r, c0, x["layer"]) + ws.cell(r, c0 + 1, x["label"]) + ws.cell(r, c0 + 2, x["previous_n"]) + ws.cell(r, c0 + 3, x["hit_n"]) + ws.cell(r, c0 + 4, x["probability"]) + ws.cell(r, c0 + 5, x["fitted_probability"]) + r += 1 + r += 1 + note = ( + "计算口径：勾选后的每次相邻两次考试都参与比较，最后一次也进入转移样本；同分同名次，后续名次递增；缺考未排除时按0分排名；" + "汇总n/k后做加权二项logistic拟合；第1层保留合成概率，首个命中为0的层及其后取0，并强制概率逐层不上升。" + ) + ws.cell(r, 1, note) + _borders(ws) + ws.column_dimensions["A"].width = 10 + ws.column_dimensions["H"].width = 10 + for c in (2, 9): + ws.column_dimensions[get_column_letter(c)].width = 24 + for c in range(3, 8): + ws.column_dimensions[get_column_letter(c)].width = 16 + for c in range(10, 14): + ws.column_dimensions[get_column_letter(c)].width = 16 + return ws + + +def _metric_sheet(wb: Workbook, group: dict): + title = _safe_title(f"{group['group']}指标数") + ws = wb.create_sheet(title) + r = 1 + for line in group["targets"]: + ws.cell(r, 1, f"{group['group']}方向{line['name']}指标分配") + r += 1 + headers = ["班级"] + [f"{i+1}层人数" for i in range(group["n_layers"])] + ["加权计算", "摊派指标"] + ws.cell(r, 1, headers[0]) + for i, h in enumerate(headers[1:], 2): + ws.cell(r, i, h) + _style_header(ws, r, len(headers)) + r += 1 + ws.cell(r, 1, "排名区间") + for i, label in enumerate(line["labels"], 2): + ws.cell(r, i, label) + r += 1 + ws.cell(r, 1, "拟合概率") + for i, p in enumerate(line["fitted"], 2): + ws.cell(r, i, p) + r += 1 + for row in line["classes"]: + ws.cell(r, 1, row["class_name"]) + for i, count in enumerate(row["layer_counts"], 2): + ws.cell(r, i, count) + ws.cell(r, len(headers) - 1, row["expected"]) + ws.cell(r, len(headers), row["assigned"]) + r += 1 + total_counts = [ + sum(row["layer_counts"][i] for row in line["classes"]) + for i in range(group["n_layers"]) + ] + ws.cell(r, 1, "合计") + for i, count in enumerate(total_counts, 2): + ws.cell(r, i, count) + ws.cell(r, len(headers) - 1, line["sum_expected"]) + ws.cell(r, len(headers), line["sum_assigned"]) + r += 1 + ws.cell(r, 1, "注：加权计算=Σ(各层人数×拟合概率)；摊派指标=加权计算四舍五入。") + r += 3 + ws.column_dimensions["A"].width = 14 + for c in range(2, ws.max_column + 1): + ws.column_dimensions[get_column_letter(c)].width = 13 + _borders(ws) + ws.freeze_panes = "B1" + return ws def build_export(result: dict) -> bytes: wb = Workbook() - thin = Border( - left=Side(style="thin", color="D9D1C3"), - right=Side(style="thin", color="D9D1C3"), - top=Side(style="thin", color="D9D1C3"), - bottom=Side(style="thin", color="D9D1C3"), - ) - head = PatternFill("solid", fgColor="12233D") - head_font = Font(color="F4EFE6", bold=True) - wrap = Alignment(wrap_text=True, vertical="center") - - first = True - for g in result["groups"]: - for t in g["targets"]: - title = f"{g['group']}-{t['name']}"[:31] - ws = wb.active if first else wb.create_sheet(title) - if first: - ws.title = title - first = False - ws.append( - ["班级", "测算人数", "最近一次在线", "滤波期望", "下达指标", "保底下界", "冲刺上界", "相对最近一次"] - ) - for cell in ws[1]: - cell.fill = head - cell.font = head_font - for row in t["classes"]: - ws.append( - [ - row["class_name"], - row["n"], - row["current"], - row["expected"], - row["assigned"], - row["floor"], - row["stretch"], - row["delta"], - ] - ) - ws.append([]) - ws.append(["说明：下达指标为预估中位水平，班级应努力超越该人数。保底/冲刺来自模拟分布的约20%/80%分位。"]) - - ws = wb.create_sheet("学生明细") - exams = result["exams"] - headers = ["学号", "姓名", "班级", "方向", "学籍", "选科"] - headers += [f"{e}-排名" for e in exams] - # collect target names - tcols = [] - for g in result["groups"]: - for t in g["targets"]: - tcols.append((g["group"], t["id"], f"{g['group']}-{t['name']}概率")) - headers += [c[2] for c in tcols] - headers += ["滤波水平", "预测方差"] - ws.append(headers) - for cell in ws[1]: - cell.fill = head - cell.font = head_font - cell.alignment = wrap - for s in result["students"]: - row = [s["sid"], s["name"], s["class_name"], s["group"], s.get("status") or "", s.get("subjects") or ""] - ranks = s.get("ranks") or [] - for i, _e in enumerate(exams): - row.append(ranks[i] if i < len(ranks) else None) - for gname, tid, _ in tcols: - if s["group"] == gname: - row.append((s.get("probs") or {}).get(tid)) - else: - row.append(None) - row.append(s.get("pred_mean")) - row.append(s.get("pred_var")) - ws.append(row) + wb.remove(wb.active) + # Keep the same familiar order as basedata.xlsx: 物理 raw/prob/metric, then 历史. + groups = sorted(result["groups"], key=lambda g: _group_order(g["group"])) + for i, group in enumerate(groups): + students = [s for s in result["students"] if s["group"] == group["group"]] + group["_exams"] = result["exams"] + _raw_sheet(wb, group, students) + for group in groups: + _probability_sheet(wb, group) + for group in groups: + _metric_sheet(wb, group) bio = BytesIO() wb.save(bio) diff --git a/app/layer_method.py b/app/layer_method.py new file mode 100644 index 0000000..16d9df3 --- /dev/null +++ b/app/layer_method.py @@ -0,0 +1,375 @@ +from __future__ import annotations + +import math +from collections import defaultdict +from decimal import Decimal, ROUND_HALF_UP + +from .excel_parse import class_sort_key + +LAYER_SIZE = 20 +MISSING_AS_SCORE = 0.0 + + +def infer_track_group(track: str) -> str: + t = track or "" + if "历史" in t: + return "历史" + if "物理" in t: + return "物理" + return t or "未分方向" + + +def default_line_settings(groups: list[str]) -> dict[str, dict[str, int]]: + out = {} + for g in groups: + if g == "物理": + out[g] = {"top": 80, "control": 262, "layer_size": 20} + elif g == "历史": + # 历史样本较小；basedata.xlsx 的历史方向使用 5 人一层。 + out[g] = {"top": 10, "control": 45, "layer_size": 5} + else: + out[g] = {"top": 20, "control": 70, "layer_size": 20} + return out + + +def select_students(parsed: dict, settings: dict) -> list[dict]: + include_mingyuan = settings.get("include_mingyuan", True) + include_other = settings.get("include_other", True) + exclude_moves = bool(settings.get("exclude_moves", False)) + exams = settings.get("exams") or parsed["exams"] + selected = [] + for s in parsed["students"]: + move_flag = bool((s.get("move") or "").strip()) or s.get("sheet_kind") in ("art", "leave") + if exclude_moves and move_flag: + continue + st = s.get("status") or "在籍" + if st == "明远" and not include_mingyuan: + continue + if st not in ("在籍", "明远", "") and not include_other: + continue + row = dict(s) + row["move_flag"] = move_flag + row["scores"] = {e: s["scores"].get(e) for e in exams} + row["computed_scores"] = { + e: MISSING_AS_SCORE if v is None else float(v) for e, v in row["scores"].items() + } + row["missing_exams"] = [e for e, v in row["scores"].items() if v is None] + row["group"] = infer_track_group(s.get("track") or "") + selected.append(row) + return selected + + +def competition_ranks(values: list[float]) -> list[int]: + """1 = highest; all equal scores receive the same rank and later ranks are skipped.""" + return [1 + sum(1 for v in values if v > x) for x in values] + + +def layer_count(n: int, size: int = LAYER_SIZE) -> int: + return max(1, math.ceil(n / size)) if n else 1 + + +def rank_layer(rank: int, size: int = LAYER_SIZE) -> int: + """A tie at a boundary stays in the upper layer because the next distinct rank is skipped.""" + return max(1, (int(rank) - 1) // size + 1) + + +def layer_label(layer: int, n: int, size: int = LAYER_SIZE) -> str: + start = (layer - 1) * size + 1 + if layer * size >= n: + return f"前{start}名及以后" + return f"前{start}-{layer * size}名" + + +def _sigmoid(x: float) -> float: + if x >= 0: + return 1.0 / (1.0 + math.exp(-min(x, 40.0))) + z = math.exp(max(x, -40.0)) + return z / (1.0 + z) + + +def _log_likelihood(beta: list[float], rows: list[tuple[float, float, float]]) -> float: + total = 0.0 + for t, n, k in rows: + p = _sigmoid(beta[0] - beta[1] * t) + p = min(max(p, 1e-10), 1 - 1e-10) + total += k * math.log(p) + (n - k) * math.log(1 - p) + return total + + +def fit_retention_probabilities( + counts: list[dict], round_digits: int = 6 +) -> list[float]: + """Weighted binomial logistic fit: p=1/(1+exp(-(a-b*layer))).""" + rows = [(float(x["layer"]), float(x["previous_n"]), float(x["hit_n"])) for x in counts] + if not rows: + return [] + + beta = [0.3, 0.15] + best = _log_likelihood(beta, rows) + for _ in range(100): + g1 = g2 = 0.0 + h11 = h12 = h22 = 0.0 + for t, n, k in rows: + p = _sigmoid(beta[0] - beta[1] * t) + w = n * p * (1 - p) + r = k - n * p + xt1, xt2 = 1.0, -t + g1 += xt1 * r + g2 += xt2 * r + h11 += w * xt1 * xt1 + h12 += w * xt1 * xt2 + h22 += w * xt2 * xt2 + # A tiny ridge prevents singular matrices when a layer is perfectly separated. + h11 += 1e-7 + h22 += 1e-7 + det = h11 * h22 - h12 * h12 + if abs(det) < 1e-14: + break + d1 = (h22 * g1 - h12 * g2) / det + d2 = (h11 * g2 - h12 * g1) / det + step = 1.0 + improved = False + for _ls in range(20): + cand = [beta[0] + step * d1, beta[1] + step * d2] + # Keep the curve monotone decreasing and numerically bounded. + cand[0] = min(max(cand[0], -6.0), 14.0) + cand[1] = min(max(cand[1], 0.0), 10.0) + value = _log_likelihood(cand, rows) + if value + 1e-9 >= best: + beta, best = cand, value + improved = True + break + step *= 0.5 + if not improved or math.sqrt(d1 * d1 + d2 * d2) < 1e-8: + break + + raw = [_sigmoid(beta[0] - beta[1] * t) for t, _, _ in rows] + + # basedata.xlsx 的约定：第 1 层不使用 logistic 估计值，而是保留合成概率。 + # 这样小样本首层的边界观测不会被模型收缩。 + if counts and counts[0]["previous_n"]: + raw[0] = counts[0]["hit_n"] / counts[0]["previous_n"] + + # basedata.xlsx 的尾部约定：第一次出现命中为 0 的层及其后，拟合概率取 0。 + first_zero = next((i for i, item in enumerate(counts) if item["hit_n"] == 0), None) + if first_zero is not None: + raw[first_zero:] = [0.0] * (len(raw) - first_zero) + + # Force monotone decreasing, even when sparse layers reverse the fitted trend. + for i in range(1, len(raw)): + raw[i] = min(raw[i], raw[i - 1]) + return [round(max(0.0, min(1.0, p)), round_digits) for p in raw] + + +def round_half_up(value: float) -> int: + return int(Decimal(str(float(value))).quantize(Decimal("1"), rounding=ROUND_HALF_UP)) + + +def _pair_counts( + members: list[dict], e0: str, e1: str, threshold: int, size: int = LAYER_SIZE +) -> list[dict]: + total_layers = layer_count(len(members), size) + buckets = [ + {"layer": i, "previous_n": 0, "hit_n": 0} for i in range(1, total_layers + 1) + ] + for s in members: + layer = rank_layer(s["ranks"][e0], size) + buckets[layer - 1]["previous_n"] += 1 + if s["ranks"][e1] <= threshold: + buckets[layer - 1]["hit_n"] += 1 + return buckets + + +def _merge_counts(parts: list[list[dict]]) -> list[dict]: + n_layers = max(len(p) for p in parts) + out = [{"layer": i, "previous_n": 0, "hit_n": 0} for i in range(1, n_layers + 1)] + for part in parts: + for item in part: + out[item["layer"] - 1]["previous_n"] += item["previous_n"] + out[item["layer"] - 1]["hit_n"] += item["hit_n"] + return out + + +def analyze(parsed: dict, settings: dict) -> dict: + exams = settings.get("exams") or parsed["exams"] + if len(exams) < 2: + raise ValueError("相邻层次法至少需要两次考试") + selected = select_students(parsed, settings) + if not selected: + raise ValueError("没有学生进入测算") + groups = sorted({s["group"] for s in selected}) + line_settings = settings.get("line_settings") or default_line_settings(groups) + + # 用户口径：勾选后的所有相邻两次都要计算，最后一次也进入相邻转移样本。 + probability_exams = exams + + # 排名与层次划分；缺考已按 0 分参加排名。 + for s in selected: + s["ranks"] = {} + s["layers"] = {} + for e in exams: + vals = [s["computed_scores"][e] for s in selected] + # Rankings are global within each direction, so calculate by group. + for g in groups: + gm = [s for s in selected if s["group"] == g] + gvals = [s["computed_scores"][e] for s in gm] + ranks = competition_ranks(gvals) + for s, rank in zip(gm, ranks): + s["ranks"][e] = rank + s["layers"][e] = rank_layer(rank) + + results = [] + all_views = [] + for g in groups: + members = [s for s in selected if s["group"] == g] + n = len(members) + lines = line_settings.get(g) or default_line_settings([g])[g] + top_k = max(1, min(int(lines.get("top", 1)), n)) + control_k = max(top_k, min(int(lines.get("control", top_k)), n)) + line_defs = [ + {"id": "top", "name": "尖子线", "k": top_k}, + {"id": "control", "name": "特控线", "k": control_k}, + ] + layer_size = max(1, min(int(lines.get("layer_size", LAYER_SIZE)), max(1, n))) + n_layers = layer_count(n, layer_size) + labels = [layer_label(i, n, layer_size) for i in range(1, n_layers + 1)] + pair_blocks = [] + summary_lines = {} + target_blocks = [] + for line in line_defs: + pair_parts = [] + pair_details = [] + pair_labels = [layer_label(i, n, layer_size) for i in range(1, n_layers + 1)] + for a, b in zip(probability_exams, probability_exams[1:]): + part = _pair_counts(members, a, b, line["k"], layer_size) + pair_parts.append(part) + pair_details.append( + { + "from_exam": a, + "to_exam": b, + "layers": [ + { + **x, + "label": pair_labels[x["layer"] - 1], + "probability": ( + x["hit_n"] / x["previous_n"] + if x["previous_n"] + else None + ), + } + for x in part + ], + } + ) + merged = _merge_counts(pair_parts) + for item, label in zip(merged, labels): + item["label"] = label + item["probability"] = ( + item["hit_n"] / item["previous_n"] + if item["previous_n"] + else None + ) + fitted = fit_retention_probabilities(merged) + for item, p in zip(merged, fitted): + item["fitted_probability"] = p + + classes = sorted({s["class_name"] for s in members}, key=class_sort_key) + class_rows = [] + for c in classes: + cm = [s for s in members if s["class_name"] == c] + layer_counts = [0] * n_layers + for s in cm: + layer_counts[rank_layer(s["ranks"][exams[-1]], layer_size) - 1] += 1 + expected = sum(a * p for a, p in zip(layer_counts, fitted)) + current = sum(1 for s in cm if s["ranks"][exams[-1]] <= line["k"]) + class_rows.append( + { + "class_name": c, + "n": len(cm), + "current": current, + "layer_counts": layer_counts, + "expected": round(expected, 4), + "assigned": round_half_up(expected), + "delta": round_half_up(expected) - current, + } + ) + summary_lines[line["id"]] = { + "layers": merged, + "labels": labels, + "total_previous": sum(x["previous_n"] for x in merged), + "total_hit": sum(x["hit_n"] for x in merged), + "fitted": fitted, + } + target_blocks.append( + { + **line, + "classes": class_rows, + "layers": merged, + "labels": labels, + "fitted": fitted, + "sum_expected": round(sum(r["expected"] for r in class_rows), 4), + "sum_assigned": sum(r["assigned"] for r in class_rows), + "sum_current": sum(r["current"] for r in class_rows), + } + ) + pair_blocks.append({"line": line["name"], "pairs": pair_details}) + + results.append( + { + "group": g, + "n": n, + "layer_size": layer_size, + "n_layers": n_layers, + "line_settings": {"top": top_k, "control": control_k}, + "pairs": pair_blocks, + "summary": summary_lines, + "targets": target_blocks, + } + ) + + for s in selected: + all_views.append( + { + "sid": s["sid"], + "name": s["name"], + "class_name": s["class_name"], + "group": s["group"], + "track": s.get("track") or "", + "subjects": s.get("subjects") or "", + "status": s.get("status") or "", + "move": s.get("move") or "", + "has_move": bool(s.get("move_flag", False)), + "scores": s["scores"], + "computed_scores": s["computed_scores"], + "missing_exams": s["missing_exams"], + "ranks": s["ranks"], + "layers": s["layers"], + } + ) + + return { + "algorithm": "adjacent-layer-markov-v2", + "exams": exams, + "layer_size": LAYER_SIZE, + "pair_count": len(probability_exams) - 1 if len(probability_exams) > 1 else 0, + "probability_exams": probability_exams, + "missing_as_score": MISSING_AS_SCORE, + "exclude_moves": bool(settings.get("exclude_moves", False)), + "n_selected": len(selected), + "n_total": len(parsed["students"]), + "groups": results, + "students": all_views, + } + + +def class_comment(row: dict, target_name: str) -> str: + assigned = row["assigned"] + current = row["current"] + expected = row["expected"] + if assigned > current: + tail = f"比最后一次到线人数多 {assigned - current} 人。" + elif assigned < current: + tail = f"比最后一次到线人数少 {current - assigned} 人。" + else: + tail = "与最后一次到线人数持平。" + return f"{row['class_name']}班{target_name}拟合摊派 {assigned} 人（加权计算 {expected:.2f}）。{tail}" diff --git a/app/main.py b/app/main.py index 3f7d76f..8fe93b8 100644 --- a/app/main.py +++ b/app/main.py @@ -32,9 +32,9 @@ from .db import ( ) from .excel_parse import parse_workbook, summarize_parse from .export_xlsx import build_export -from .indicators import analyze, class_comment, default_targets, infer_track_group, select_students +from .layer_method import analyze, class_comment, default_line_settings, infer_track_group, select_students -app = FastAPI(title="航标 · 班级达线指标预估") +app = FastAPI(title="指标测算 · 相邻层次法") app.add_middleware(SessionMiddleware, secret_key=SECRET_KEY, max_age=60 * 60 * 12) app.mount("/static", StaticFiles(directory=Path(__file__).parent / "static"), name="static") templates = Jinja2Templates(directory=str(Path(__file__).parent / "templates")) @@ -136,17 +136,20 @@ def demo(request: Request): def _settings_from(parsed: dict, stored: dict | None) -> dict: summary = summarize_parse(parsed) groups = sorted({infer_track_group(s["track"]) for s in parsed["students"]}) + line_settings = default_line_settings(groups) base = { "exams": parsed["exams"], - "include_art": False, - "include_leave": False, + "exclude_moves": False, "include_mingyuan": True, "include_other": True, - "horizon": 1, - "targets": default_targets(groups), + "line_settings": line_settings, } if stored: - for k in ("exams", "include_art", "include_leave", "include_mingyuan", "include_other", "horizon", "targets"): + # 考试列必须始终来自本次上传的工作簿，避免旧分析设置把不存在的列带进来。 + valid_exam_set = set(parsed["exams"]) + stored_exams = [e for e in (stored.get("exams") or []) if e in valid_exam_set] + base["exams"] = stored_exams or parsed["exams"] + for k in ("exclude_moves", "include_mingyuan", "include_other", "line_settings"): if k in stored: base[k] = stored[k] base["summary"] = summary @@ -154,6 +157,7 @@ def _settings_from(parsed: dict, stored: dict | None) -> dict: return base + @app.get("/analysis/{aid}", response_class=HTMLResponse) def analysis_page(request: Request, aid: int, step: int = 1): if (redir := require_login(request)): @@ -189,34 +193,39 @@ async def save_settings(request: Request, aid: int): return HTMLResponse("分析不存在", status_code=404) parsed = load_json(item["parsed_path"]) form = await request.form() - exams = form.getlist("exams") + valid_exam_set = set(parsed["exams"]) + exams = [e for e in form.getlist("exams") if e in valid_exam_set] if not exams: exams = parsed["exams"] + if len(exams) < 2: + return HTMLResponse("至少需要勾选两次考试才能计算相邻转移概率。", status_code=400) groups = sorted({infer_track_group(s["track"]) for s in parsed["students"]}) - targets = {} + defaults = default_line_settings(groups) + line_settings = {} for g in groups: - names = form.getlist(f"t_name_{g}") - ks = form.getlist(f"t_k_{g}") - ids = form.getlist(f"t_id_{g}") - rows = [] - for i, name in enumerate(names): - k_raw = ks[i] if i < len(ks) else "0" - try: - k = int(float(k_raw)) - except ValueError: - k = 0 - tid = ids[i] if i < len(ids) and ids[i] else f"t{i+1}" - if name.strip() and k > 0: - rows.append({"id": tid, "name": name.strip(), "k": k}) - targets[g] = rows or default_targets([g])[g] + try: + top = int(float(form.get(f"top_k_{g}", defaults[g]["top"]))) + except ValueError: + top = defaults[g]["top"] + try: + control = int(float(form.get(f"control_k_{g}", defaults[g]["control"]))) + except ValueError: + control = defaults[g]["control"] + try: + layer_size = int(float(form.get(f"layer_size_{g}", defaults[g]["layer_size"]))) + except ValueError: + layer_size = defaults[g]["layer_size"] + line_settings[g] = { + "top": max(1, top), + "control": max(1, max(top, control)), + "layer_size": max(1, layer_size), + } settings = { "exams": list(exams), - "include_art": form.get("include_art") == "on", - "include_leave": form.get("include_leave") == "on", + "exclude_moves": form.get("exclude_moves") == "on", "include_mingyuan": form.get("include_mingyuan") == "on", "include_other": form.get("include_other") == "on", - "horizon": int(form.get("horizon") or 1), - "targets": targets, + "line_settings": line_settings, } update_analysis(aid, settings_json=json.dumps(settings, ensure_ascii=False), step=2) action = form.get("action") or "next" @@ -230,6 +239,7 @@ async def save_settings(request: Request, aid: int): return RedirectResponse(f"/analysis/{aid}?step=2", status_code=303) + @app.get("/analysis/{aid}/api/student/{sid}") def student_api(request: Request, aid: int, sid: str): if not current_user(request): @@ -257,8 +267,14 @@ def class_api(request: Request, aid: int, group: str, target: str, class_name: s for s in result["students"] if s["group"] == group and s["class_name"] == class_name ] - rows.sort(key=lambda s: -((s.get("probs") or {}).get(target) or 0)) - return {"exams": result["exams"], "students": rows[:80], "n": len(rows)} + rows.sort(key=lambda s: (s.get("ranks") or {}).get(result["exams"][-1], 10**9)) + return { + "exams": result["exams"], + "layer_size": result.get("layer_size"), + "last_exam": result["exams"][-1], + "students": rows, + "n": len(rows), + } @app.get("/analysis/{aid}/export") @@ -269,6 +285,8 @@ def export_analysis(request: Request, aid: int): if not item or not item.get("result_path"): return HTMLResponse("请先完成测算", status_code=400) result = load_json(item["result_path"]) + if result.get("algorithm") != "adjacent-layer-markov-v2": + return HTMLResponse("结果为旧算法，请回到第二步重新测算后再导出。", status_code=400) data = build_export(result) ascii_name = f"hangbiao-{aid}.xlsx" star = quote(f"{item['title']}-下达指标.xlsx") diff --git a/app/static/css/app.css b/app/static/css/app.css index fc852cf..779233e 100644 --- a/app/static/css/app.css +++ b/app/static/css/app.css @@ -147,3 +147,7 @@ input[type=text], input[type=password], input[type=number], select { .hero, .layout, .grid-2, .kpis { grid-template-columns: 1fr; } .rail { position: static; } } +.grid-3 { display: grid; grid-template-columns: repeat(3, 1fr); gap: 12px; } +details { border: 1px solid var(--line); border-radius: 14px; background: #fff; padding: 10px 14px; margin: 10px 0; } +details summary { cursor: pointer; font-size: 14px; font-weight: 600; } +@media (max-width: 900px) { .grid-3 { grid-template-columns: 1fr; } } diff --git a/app/static/js/app.js b/app/static/js/app.js index a8cd9cf..12102f4 100644 --- a/app/static/js/app.js +++ b/app/static/js/app.js @@ -3,77 +3,42 @@ function closeDrawer() { if (el) el.classList.remove("show"); } -function sparkline(svg, exams, zs, xs, lastRank) { - const w = 520, h = 220, pad = 28; - const vals = []; - zs.forEach((z, i) => { if (z !== null && z !== undefined) vals.push(z); }); - xs.forEach((z) => { if (z !== null && z !== undefined) vals.push(z); }); - if (!vals.length) return; - const min = Math.min(...vals, -0.5) - 0.3; - const max = Math.max(...vals, 0.5) + 0.3; - const xAt = (i) => pad + (i * (w - pad * 2)) / Math.max(exams.length - 1, 1); - const yAt = (v) => pad + (max - v) * (h - pad * 2) / (max - min); - const pts = (arr) => arr.map((v, i) => v == null ? null : `${xAt(i)},${yAt(v)}`).filter(Boolean).join(" "); - let raw = ""; - zs.forEach((v, i) => { - if (v == null) return; - raw += ``; - }); - const pathXs = xs.map((v, i) => (v == null ? null : [xAt(i), yAt(v)])).filter(Boolean); - let d = ""; - pathXs.forEach((p, i) => { d += (i === 0 ? `M${p[0]} ${p[1]}` : ` L${p[0]} ${p[1]}`); }); - const labels = exams.map((e, i) => { - const short = String(e).replace("-总分", "").replace("高", ""); - return `${short}`; - }).join(""); - svg.setAttribute("viewBox", `0 0 ${w} ${h}`); - svg.innerHTML = ` - - - ${raw} - ${labels} - 标准分轨迹 · 橙点考试，绿线滤波估计 - `; +function esc(value) { + return String(value ?? "").replace(/[&<>"']/g, c => ({"&":"&","<":"<",">":">",'"':""","'":"'"}[c])); } async function openStudent(aid, sid) { const res = await fetch(`/analysis/${aid}/api/student/${encodeURIComponent(sid)}`); if (!res.ok) return; const s = await res.json(); + const exams = window.HB_EXAMS || []; + const rows = exams.map((exam, i) => { + const score = (s.scores || {})[exam]; + const computed = (s.computed_scores || {})[exam]; + const rank = (s.ranks || {})[exam]; + const layer = (s.layers || {})[exam]; + return ` + ${esc(exam)} + ${score == null ? "缺考（按 0 分）" : esc(score)} + ${rank ?? "—"}${layer ?? "—"} + ${computed ?? "—"} + `; + }).join(""); const box = document.getElementById("drawer"); const body = document.getElementById("drawer-body"); - const probs = Object.entries(s.probs || {}).map(([k, v]) => `${k}: ${(v * 100).toFixed(1)}%`).join("　"); body.innerHTML = `
-

${s.name} ${s.sid}

-
${s.class_name}班 · ${s.group} · ${s.status || "在籍"} · ${s.subjects || ""}
+

${esc(s.name)} ${esc(s.sid)}

+
${esc(s.class_name)}班 · ${esc(s.group)} · ${esc(s.status || "在籍")} · ${esc(s.subjects || "")}
+ ${s.move ? `

异动/备注：${esc(s.move)}

` : ""}
-

- 绿线是卡尔曼估计的稳定水平。如果橙点突然跳开绿线，系统会当成发挥波动，不会立刻改判断。 - 达线概率 ${probs || "—"}。 -

- - - - -
考试分数方向内名次标准分滤波
+ + + ${rows}
考试原始分方向名次层次计算分
`; - const rows = document.getElementById("spark-rows"); - (s.z || []).forEach((z, i) => { - const exam = (window.HB_EXAMS || [])[i] || `第${i + 1}次`; - const score = s.scores ? s.scores[exam] : ""; - const rank = (s.ranks || [])[i]; - const kf = (s.kf_xs || [])[i]; - rows.insertAdjacentHTML("beforeend", ` - ${exam}${score ?? "缺考"}${rank ?? "—"} - ${z == null ? "—" : z.toFixed(2)} - ${kf == null ? "—" : kf.toFixed(2)} - `); - }); - sparkline(document.getElementById("spark"), window.HB_EXAMS || [], s.z || [], s.kf_xs || []); box.classList.add("show"); } @@ -82,27 +47,24 @@ async function openClass(aid, group, target, className) { const res = await fetch(url); if (!res.ok) return; const data = await res.json(); - const box = document.getElementById("drawer"); - const body = document.getElementById("drawer-body"); - const rows = data.students.map((s) => { - const p = ((s.probs || {})[target] || 0) * 100; - const tag = p >= 80 ? "high" : p >= 20 ? "mid" : ""; - return ` - ${s.name}${s.status || ""} - ${(s.ranks || []).slice(-1)[0] ?? "—"} - ${p.toFixed(1)}% + const lastExam = data.last_exam || (data.exams || []).slice(-1)[0]; + const rows = (data.students || []).map((s) => { + const rank = (s.ranks || {})[lastExam]; + const layer = (s.layers || {})[lastExam]; + return ` + ${esc(s.name)}${esc(s.status || "")} + ${rank ?? "—"}${layer ?? "—"} `; }).join(""); + const box = document.getElementById("drawer"); + const body = document.getElementById("drawer-body"); body.innerHTML = `
-

${className}班学生概率

+

${esc(className)}班 · ${esc(lastExam || "")}

-

点击学生可看成绩轨迹。共 ${data.n} 人，下列按达线概率排序。

- - - ${rows} -
姓名学籍最近名次达线概率
+

按最后一次方向名次排序，共 ${data.n} 人。

+ ${rows}
姓名学籍名次层次
`; box.classList.add("show"); } diff --git a/app/templates/analysis.html b/app/templates/analysis.html index 47efd2b..c333261 100644 --- a/app/templates/analysis.html +++ b/app/templates/analysis.html @@ -3,30 +3,31 @@ - {{ item.title }} · 航标 + {{ item.title }} · 相邻层次法
-
航
-
航标{{ item.title }}
+
层
+
指标测算{{ item.title }}
- {% if result %} - 导出 Excel + {% if result and result.get('algorithm') == 'adjacent-layer-markov-v2' %} + 导出总表 {% endif %}
+
-
-
-
- +
+ diff --git a/app/templates/home.html b/app/templates/home.html index 53b16af..6801691 100644 --- a/app/templates/home.html +++ b/app/templates/home.html @@ -3,38 +3,38 @@ - 航标 · 班级达线指标预估 + 指标测算 · 相邻层次法
-
航
-
航标班级达线指标预估系统
+
层
+
指标测算相邻层次指标测算
-

把历史成绩当成导航轨迹，
而不是一张静态排名表。

+

相邻两次考试做层次转移，
用拟合概率下达班级指标。

- 上传各次大考总分后，系统会把分数换成同一次考试内的标准分，用卡尔曼滤波估计每个学生的稳定水平， - 再模拟下一次全年级排名，把达线概率加总到班级。下达的是“按现有轨迹本来会到的人数”，班级工作是超越它。 + 系统按方向内名次分层次，统计每次相邻考试中各层次的到线人数；汇总后计算拟合概率， + 最后按最后一次考试的班级层次分布摊派尖子线和特控线指标。可勾选是否排除异动学生。

上传成绩表

-

xlsx，需含学号/姓名/班级，以及多次考试总分列。若有「方向」或物理/历史分表，会分开测算。

+

xlsx，需含学号/姓名/班级，以及多次考试总分列。物理/历史分表会分开测算。

- +
{% if has_sample %}
- +
{% endif %}
diff --git a/app/templates/login.html b/app/templates/login.html index 126b0f5..5a89720 100644 --- a/app/templates/login.html +++ b/app/templates/login.html @@ -3,15 +3,15 @@ - 登录 · 航标 + 登录 · 指标测算