"""数据与学习效率：读取筛选实测记录，计算递归生成的分布变化。"""

from argparse import ArgumentParser
from collections import Counter
from hashlib import sha256
import json
from pathlib import Path
import random
import statistics

DATA = Path(__file__).with_name("data")
PLAN = DATA / "data-efficiency-plan.json"
RECORD = DATA / "data-efficiency-training.json"
LOOPS = DATA / "data-efficiency-loops.json"
ARMS = ["all", "random", "easy", "hard", "balanced", "curriculum"]
LABELS = {"all": "全部候选", "random": "随机子集", "easy": "低损失", "hard": "高损失",
          "balanced": "分类低损失", "curriculum": "渐进课程"}
PROTOCOLS = ["replace", "accumulate", "anchor", "accumulate_sharp"]
LOOP_LABELS = {"replace": "只留新生成", "accumulate": "累积全部", "anchor": "原始加本轮",
               "accumulate_sharp": "累积但偏爱高频"}


def digest(path):
    return sha256(Path(path).read_bytes()).hexdigest()


def probability(counts):
    total = sum(counts)
    if total <= 0 or any(n < 0 for n in counts):
        raise ValueError("计数必须非负，总和必须为正")
    return [n/total for n in counts]


def total_variation(p, q):
    return sum(abs(a-b) for a, b in zip(p, q))/2


def next_counts(original, current, generated, protocol):
    if protocol == "replace":
        return generated[:]
    if protocol in {"accumulate", "accumulate_sharp"}:
        return [a+b for a, b in zip(current, generated)]
    if protocol == "anchor":
        return [a+b for a, b in zip(original, generated)]
    raise ValueError(protocol)


def loop_experiment():
    original = [50, 25, 15, 8, 2]
    config = {"original": original, "seeds": list(range(1000, 1200)),
              "generations": 20, "new_samples": 100,
              "protocols": PROTOCOLS, "source_sha256": digest(__file__)}
    runs = []
    for seed in config["seeds"]:
        paths = {}
        for protocol in PROTOCOLS:
            # Shared initial seed, not shared later samples after distributions diverge.
            rng = random.Random(seed)
            current = original[:]
            history = [current[:]]
            for _ in range(config["generations"]):
                p = probability(current)
                if protocol == "accumulate_sharp":
                    p = probability([v*v for v in p])
                samples = rng.choices(range(len(p)), weights=p, k=config["new_samples"])
                counted = Counter(samples)
                generated = [counted[i] for i in range(len(p))]
                current = next_counts(original, current, generated, protocol)
                history.append(current[:])
            paths[protocol] = history
        runs.append({"seed": seed, "paths": paths})
    return {"config": config, "runs": runs}


def loop_summary(record):
    target = probability(record["config"]["original"])
    result = {}
    for protocol in PROTOCOLS:
        rows = []
        for t in range(record["config"]["generations"]+1):
            counts = [run["paths"][protocol][t] for run in record["runs"]]
            distributions = [probability(c) for c in counts]
            rows.append({"generation": t, "n": sum(counts[0]),
                         "tail_zero": sum(c[-1] == 0 for c in counts),
                         "tail_mean": statistics.mean(p[-1] for p in distributions),
                         "tv_mean": statistics.mean(total_variation(p, target) for p in distributions)})
        result[protocol] = rows
    return result


def mean_training_metrics(record, split="test"):
    result = {}
    for arm in ARMS:
        rows = [run["evaluations"][split]["metrics"] for run in record["runs"] if run["arm"] == arm]
        result[arm] = {"accuracy": statistics.mean(x["accuracy"] for x in rows),
                       "tail_accuracy": statistics.mean(x["tail_accuracy"] for x in rows)}
    return result


def main():
    parser = ArgumentParser(description=__doc__)
    parser.add_argument("--run-loops", type=Path)
    args = parser.parse_args()
    if args.run_loops:
        with args.run_loops.open("x") as stream:
            json.dump(loop_experiment(), stream, ensure_ascii=False, separators=(",", ":"))
            stream.write("\n")
        print(f"已保存 200 个种子、4 种协议、20 代的记录：{args.run_loops}")
        return
    record = json.loads(RECORD.read_text())
    assert record["plan_sha256"] == digest(PLAN)
    print("留出测试均值：总体 / 数字 8 与 9")
    for arm, m in mean_training_metrics(record).items():
        print(f"{LABELS[arm]}：{100*m['accuracy']:.1f}% / {100*m['tail_accuracy']:.1f}%")
    loops = json.loads(LOOPS.read_text())
    print("第 20 代：罕见类概率为零 / 平均总变差")
    for protocol, rows in loop_summary(loops).items():
        last = rows[-1]
        print(f"{LOOP_LABELS[protocol]}：{last['tail_zero']}/200 / {last['tv_mean']:.4f}")
    print(f"概率 0.02 的类别在 100 次中一次未出现：{.98**100:.4f}")


if __name__ == "__main__":
    main()

"""固定预算的真实 digits 筛选对照；--freeze 保存计划，--run 才训练。"""

from argparse import ArgumentParser
from copy import deepcopy
import json
from pathlib import Path
import platform
import random
import time

import numpy as np
import sklearn
from sklearn.datasets import load_digits
from sklearn.model_selection import train_test_split
import torch
from torch import nn
from torch.nn import functional as F

import data_efficiency as lab


def data():
    x, y = load_digits(return_X_y=True)
    raw = json.dumps({"pixels": x.astype(int).tolist(), "labels": y.tolist()},
                     separators=(",", ":")).encode()
    return torch.tensor(x/16, dtype=torch.float32), torch.tensor(y), lab.sha256(raw).hexdigest()


def make_plan():
    x, y, raw_hash = data()
    pool, test = train_test_split(np.arange(len(y)), test_size=300,
                                 stratify=y.numpy(), random_state=20261020)
    pool, dev = train_test_split(pool, test_size=200, stratify=y[pool].numpy(),
                                random_state=20261021)
    rng = random.Random(20261022)
    selector_ids, candidates = [], []
    for label in range(10):
        ids = [int(i) for i in pool if int(y[i]) == label]
        rng.shuffle(ids)
        selector_ids.extend(ids[:10])
        count = 100 if label < 8 else 20
        assert len(ids) >= 10+count
        candidates.extend(ids[10:10+count])
    selector_ids.sort()
    candidates.sort()
    observed = {str(i): int(y[i]) for i in candidates}
    changed = sorted(rng.sample(candidates, round(.08*len(candidates))))
    for i in changed:
        observed[str(i)] = (observed[str(i)]+rng.randrange(1, 10)) % 10
    assert min(sum(v == k for v in observed.values()) for k in range(10)) >= 16
    return {"version": 1, "date": "2026-10-20", "dataset_sha256": raw_hash,
            "selector_ids": selector_ids, "candidate_ids": candidates,
            "dev_ids": sorted(map(int, dev)), "test_ids": sorted(map(int, test)),
            "observed_labels": observed, "changed_ids": changed,
            "seeds": [11, 22, 33], "arms": lab.ARMS, "subset_size": 160,
            "hidden": 32, "lr": .003, "batch_size": 32, "steps": 240,
            "selector_seed": 20261023, "selector_steps": 200,
            "curriculum_limits": [[60, 40], [120, 80], [240, 160]],
            "selection": "last checkpoint, all arms and seeds; no dev/test selection",
            "corruption": "67 controlled label changes; originals used only for audit/evaluation"}


def model(plan):
    return nn.Sequential(nn.Linear(64, plan["hidden"]), nn.Tanh(), nn.Linear(plan["hidden"], 10))


def train(net, x, labels, ids, plan, seed, steps, curriculum=False):
    generator = torch.Generator().manual_seed(seed+10000)
    optimizer = torch.optim.Adam(net.parameters(), lr=plan["lr"])
    history = []
    for step in range(steps):
        limit = len(ids)
        if curriculum:
            limit = next(n for until, n in plan["curriculum_limits"] if step < until)
        positions = torch.randint(limit, (plan["batch_size"],), generator=generator)
        batch = torch.tensor([ids[i] for i in positions.tolist()])
        loss = F.cross_entropy(net(x[batch]), labels[batch])
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        history.append({"step": step+1, "pool_size": limit, "ids": batch.tolist(), "loss": loss.item()})
    return history


def choose(plan, observed, scores, seed, arm):
    candidates = plan["candidate_ids"]
    if arm == "all":
        return candidates[:]
    rng = random.Random(seed+20000)
    if arm in {"random", "curriculum"}:
        chosen = rng.sample(candidates, plan["subset_size"])
        return sorted(chosen, key=lambda i: (scores[i], i)) if arm == "curriculum" else sorted(chosen)
    if arm in {"easy", "hard"}:
        direction = 1 if arm == "easy" else -1
        return sorted(candidates, key=lambda i: (direction*scores[i], i))[:plan["subset_size"]]
    if arm == "balanced":
        result = []
        for label in range(10):
            group = [i for i in candidates if int(observed[i]) == label]
            result.extend(sorted(group, key=lambda i: (scores[i], i))[:16])
        assert len(result) == plan["subset_size"]
        return sorted(result)
    raise ValueError(arm)


def metrics(logits, labels):
    z, y = torch.tensor(logits), torch.tensor(labels)
    correct = z.argmax(-1) == y
    tail = (y == 8) | (y == 9)
    return {"n": len(y), "correct": int(correct.sum()), "accuracy": correct.float().mean().item(),
            "tail_n": int(tail.sum()), "tail_correct": int(correct[tail].sum()),
            "tail_accuracy": correct[tail].float().mean().item(),
            "nll": F.cross_entropy(z, y).item(),
            "by_class": [{"label": k, "n": int((y == k).sum()),
                          "correct": int(correct[y == k].sum())} for k in range(10)]}


@torch.no_grad()
def evaluate(net, x, y, ids):
    logits = net(x[ids]).tolist()
    labels = y[ids].tolist()
    return {"ids": ids, "logits": logits, "labels": labels, "metrics": metrics(logits, labels)}


def run(plan_path, out):
    out.mkdir(parents=True, exist_ok=False)
    plan = json.loads(plan_path.read_text())
    assert plan == make_plan()
    torch.set_num_threads(1)
    torch.use_deterministic_algorithms(True)
    x, y, raw_hash = data()
    assert raw_hash == plan["dataset_sha256"]
    observed = y.clone()
    for i, label in plan["observed_labels"].items():
        observed[int(i)] = label
    start = time.monotonic()
    torch.manual_seed(plan["selector_seed"])
    selector = model(plan)
    selector_history = train(selector, x, y, plan["selector_ids"], plan,
                             plan["selector_seed"], plan["selector_steps"])
    with torch.no_grad():
        losses = F.cross_entropy(selector(x[plan["candidate_ids"]]),
                                 observed[plan["candidate_ids"]], reduction="none").tolist()
    scores = dict(zip(plan["candidate_ids"], losses))
    record = {"plan_sha256": lab.digest(plan_path),
              "source_sha256": {p.name: lab.digest(p) for p in [Path(__file__), Path(lab.__file__)]},
              "environment": {"python": platform.python_version(), "torch": torch.__version__,
                              "sklearn": sklearn.__version__, "numpy": np.__version__,
                              "platform": platform.platform(), "device": "cpu", "threads": 1},
              "parameters": sum(p.numel() for p in selector.parameters()),
              "selector_history": selector_history, "candidate_scores": scores,
              "candidate_true_labels": {str(i): int(y[i]) for i in plan["candidate_ids"]}, "runs": []}
    for seed in plan["seeds"]:
        torch.manual_seed(seed)
        initial = model(plan)
        for arm in plan["arms"]:
            selected = choose(plan, observed, scores, seed, arm)
            net = deepcopy(initial)
            history = train(net, x, observed, selected, plan, seed, plan["steps"], arm == "curriculum")
            evaluations = {split: evaluate(net, x, y, plan[f"{split}_ids"]) for split in ["dev", "test"]}
            path = out/f"seed-{seed}-{arm}.pt"
            torch.save({"plan": plan, "state_dict": net.state_dict()}, path)
            restored = model(plan)
            restored.load_state_dict(torch.load(path, weights_only=True)["state_dict"])
            assert all(evaluate(restored, x, y, plan[f"{split}_ids"]) == ev for split, ev in evaluations.items())
            counts = lambda labels: [int((labels[selected] == k).sum()) for k in range(10)]
            row = {"seed": seed, "arm": arm, "selected": selected,
                   "true_class_counts": counts(y), "observed_class_counts": counts(observed),
                   "selected_changed": sum(i in plan["changed_ids"] for i in selected),
                   "history": history, "evaluations": evaluations,
                   "checkpoint": {"file": path.name, "sha256": lab.digest(path), "reload_equal": True}}
            record["runs"].append(row)
            m = evaluations["test"]["metrics"]
            print(f"seed={seed} arm={arm} test={m['correct']}/{m['n']} "
                  f"tail={m['tail_correct']}/{m['tail_n']}", flush=True)
    record["wall_seconds"] = time.monotonic()-start
    (out/"record.json").write_text(json.dumps(record, ensure_ascii=False, separators=(",", ":"))+"\n")
    print(f"完成 18 次学生训练：{out}", flush=True)


def main():
    parser = ArgumentParser(description=__doc__)
    parser.add_argument("--freeze", type=Path)
    parser.add_argument("--run", action="store_true")
    parser.add_argument("--plan", type=Path, default=lab.PLAN)
    parser.add_argument("--out-dir", type=Path)
    args = parser.parse_args()
    if args.freeze:
        with args.freeze.open("x") as stream:
            json.dump(make_plan(), stream, ensure_ascii=False, indent=2)
            stream.write("\n")
        print(f"已冻结数据、噪声与训练设置：{args.freeze}")
        return
    if not args.run:
        print("默认不训练。请使用 --run --out-dir runs/data-efficiency-new。")
        return
    if args.out_dir is None:
        parser.error("需要 --out-dir，且目录不能已存在")
    run(args.plan, args.out_dir)


if __name__ == "__main__":
    main()

1. 不能。这里网络结构、批量和更新次数相同，学生的样本访问量仍为 7,680。评分方案还增加了评分模型训练与候选前向计算；不同样本数量、人工标注量、词元数与实际时间应分别记录。
2. $67/160=41.875\%$。高损失可能来自错标，也可能来自真正需要学习的难题；要结合来源、覆盖和下游表现判断，不能全部删除。
3. 总体分数把不同类别混在一起。还需要知道真实使用中各类出现的比例、哪类错误代价更大，以及改进能否在独立数据和更多重复中保持。
4. 更新位移为 $(0.1,0.1)$。二次损失中还有位移平方的一半，即 $(0.1^2+0.1^2)/2=0.01$；一阶近似省略了它。
5. 前 40 张中的每一张，在三个阶段每次抽取被选中的概率分别为 $1/40$、$1/80$、$1/160$。访问次数期望为 $60\times32/40+60\times32/80+120\times32/160=96$；最后 80 张只参与末阶段，每张期望为 24 次。实际次数随机波动，因此课程同时改变了重复权重。
6. 分别约为 0.1326 和 0.0176。后者更小，但有限次数下仍大于零；下一代的概率也会随本轮结果改变，不能把各代都当作始终固定 0.02。
7. 没有新罕见记录时，分子仍至少为 2。总数 200 时下界为 1%；总数 1000 时只有 0.2%。保留原始记录的数量与它在训练混合中的权重都重要。
8. 总变差为 $(0.2+0.2)/2=0.2$。均值为 0.02 不代表每次都接近 0.02，其中一次已经完全缺失，另一次偏高。
9. 归一化分母为 0.3418。罕见类变成 $0.0004/0.3418\approx0.00117$，最常见类变成 $0.25/0.3418\approx0.7314$；原本较大的概率相对占比进一步增加。
10. 验证器可以按题面给定规则重算费用，不能单靠算术确认这些计价规则在真实工坊有效，也不能保证题型覆盖。独立测试题应与生成、筛选、调参使用的材料隔离，在方案确定后用于最后评价。