平均概率都一样，为什么一组模型比另一组更有分歧？

"""Finite categorical mixtures; no model training or posterior inference."""
from copy import deepcopy
from decimal import Decimal, localcontext
import json
import math
from pathlib import Path

TOL = 1e-12
checks = []


def check(name, condition):
    if not condition:
        raise AssertionError(name)
    checks.append(name)


def near(left, right):
    return math.isclose(left, right, rel_tol=0.0, abs_tol=TOL)


def distribution(values):
    if not isinstance(values, (list, tuple)) or not values:
        raise ValueError("nonempty probability sequence required")
    if any(isinstance(x, bool) or not isinstance(x, (int, float))
           or not math.isfinite(x) or not 0 <= x <= 1 for x in values):
        raise ValueError("finite probabilities in [0,1] required")
    if not near(math.fsum(values), 1.0):
        raise ValueError("probabilities must sum to one")


def entropy(probabilities):
    distribution(probabilities)
    return math.fsum(-p * math.log2(p) for p in probabilities if p > 0)


def summarize(weights, members):
    distribution(weights)
    if not isinstance(members, (list, tuple)) or len(weights) != len(members):
        raise ValueError("one distribution per weight required")
    for row in members:
        distribution(row)
    if len({len(row) for row in members}) != 1:
        raise ValueError("all members need the same category order and count")
    for weight, row in zip(weights, members):
        if any(weight > 0 and p > 0 and weight * p == 0 for p in row):
            raise ValueError("positive joint mass underflows this float implementation")
    mean = [math.fsum(w * row[k] for w, row in zip(weights, members))
            for k in range(len(members[0]))]
    member_entropies = [entropy(row) for row in members]
    expected = math.fsum(w * h for w, h in zip(weights, member_entropies))
    predictive = entropy(mean)
    # A separate expression: average divergence from each member to the mixture.
    weighted_kl = math.fsum(
        w * math.fsum(p * (math.log2(p) - math.log2(q))
                      for p, q in zip(row, mean) if p > 0)
        for w, row in zip(weights, members) if w > 0
    )
    return {"mean": mean, "predictive_entropy_bits": predictive,
            "member_entropy_bits": member_entropies, "expected_entropy_bits": expected,
            "gap_bits": predictive - expected, "weighted_kl_bits": weighted_kl}


def decimal_joint_information(weights, members):
    """Independent reference: I(M;Y) from the high-precision joint table."""
    with localcontext() as context:
        context.prec = 70
        mass = [[Decimal(str(w)) * Decimal(str(p)) for p in row]
                for w, row in zip(weights, members)]
        model_marginal = [sum(row, Decimal(0)) for row in mass]
        label_marginal = [sum((row[k] for row in mass), Decimal(0))
                          for k in range(len(members[0]))]
        information = Decimal(0)
        for m, row in enumerate(mass):
            for k, joint in enumerate(row):
                if joint:
                    independent = model_marginal[m] * label_marginal[k]
                    information += joint * (joint / independent).ln()
        return float(information / Decimal(2).ln())


CASES = [
    ("confident_disagreement", [0.5, 0.5], [[0.99, 0.01], [0.01, 0.99]]),
    ("shared_uncertainty", [0.5, 0.5], [[0.5, 0.5], [0.5, 0.5]]),
    ("unanimous_confident", [0.5, 0.5], [[0.99, 0.01], [0.99, 0.01]]),
    ("weighted_disagreement", [0.8, 0.2], [[1.0, 0.0], [0.0, 1.0]]),
    ("three_categories", [0.2, 0.3, 0.5], [[0.8, 0.1, 0.1], [0.1, 0.7, 0.2], [0.2, 0.2, 0.6]]),
    ("single_member", [1.0], [[0.2, 0.3, 0.5]]),
    ("zero_weight", [1.0, 0.0], [[0.2, 0.8], [1.0, 0.0]]),
    ("structural_zero", [0.5, 0.5], [[1.0, 0.0, 0.0], [0.0, 1.0, 0.0]]),
]


def main():
    report = []
    by_name = {}
    for name, weights, members in CASES:
        before = deepcopy((weights, members))
        result = summarize(weights, members)
        reference = decimal_joint_information(weights, members)
        check(name + " mixture mass", near(math.fsum(result["mean"]), 1.0))
        check(name + " entropy gap equals weighted KL", near(result["gap_bits"], result["weighted_kl_bits"]))
        check(name + " independent Decimal joint table", near(result["gap_bits"], reference))
        check(name + " nonnegative up to tolerance", result["gap_bits"] >= -TOL)
        check(name + " bounded by model-index entropy", result["gap_bits"] <= entropy(weights) + TOL)
        check(name + " bounded by category count", result["gap_bits"] <= math.log2(len(members[0])) + TOL)
        check(name + " input unchanged", (weights, members) == before)
        row = {"case": name, "weights": weights, "members": members, **result,
               "decimal_joint_information_bits": reference}
        report.append(row)
        by_name[name] = row
        print(f"{name}: predictive={result['predictive_entropy_bits']:.6f}; "
              f"expected={result['expected_entropy_bits']:.6f}; gap={result['gap_bits']:.6f}")

    left = by_name["confident_disagreement"]
    right = by_name["shared_uncertainty"]
    check("same predictive distribution", left["mean"] == right["mean"] == [0.5, 0.5])
    check("same predictive entropy", near(left["predictive_entropy_bits"], 1.0)
          and near(right["predictive_entropy_bits"], 1.0))
    check("different member uncertainty", left["expected_entropy_bits"] < right["expected_entropy_bits"])
    check("different disagreement", left["gap_bits"] > right["gap_bits"])
    check("identical uncertain members have zero gap", near(right["gap_bits"], 0.0))
    check("unequal one-hot weights give H(weights)", near(by_name["weighted_disagreement"]["gap_bits"], entropy([0.8, 0.2])))
    check("single member has zero gap", near(by_name["single_member"]["gap_bits"], 0.0))
    check("zero-weight member does not contribute", near(by_name["zero_weight"]["gap_bits"], 0.0))
    check("unused category needs no epsilon", near(by_name["structural_zero"]["gap_bits"], 1.0))

    unanimous = by_name["unanimous_confident"]
    hypothetical_label = 1
    predictions = [max(range(len(row)), key=row.__getitem__) for row in unanimous["members"]]
    check("low disagreement can accompany a specified error", predictions == [0, 0]
          and hypothetical_label == 1 and near(unanimous["gap_bits"], 0.0))
    print("specified_counterexample: predictions=[0,0]; label=1; zero_gap_does_not_prove_correctness")

    invalid = [
        ([], []), ([0.4, 0.4], [[1, 0], [0, 1]]), ([-0.1, 1.1], [[1, 0], [0, 1]]),
        ([1], []), ([1], [[0.2, 0.2]]), ([1], [[-0.1, 1.1]]),
        ([0.5, 0.5], [[1, 0], [0.2, 0.3, 0.5]]), ([1], [[float('nan'), 1]]),
        ([1], [[float('inf'), 0]]), ([True], [[1, 0]]), ([1], [[True, False]]),
        ([1], [["0.5", "0.5"]]), ([1e-300, 1.0], [[1e-300, 1.0], [0.5, 0.5]]),
    ]
    for index, args in enumerate(invalid):
        try:
            summarize(*args)
        except ValueError:
            check("reject invalid or unrepresentable input " + str(index), True)
        else:
            raise AssertionError("invalid input accepted: " + str(index))
    print(f"PASS checks={len(checks)}; reference=weighted_KL+Decimal_joint_table; unit=bits")
    Path('report.json').write_text(json.dumps({
        "scope": "specified finite categorical mixtures; no training, posterior inference, or real-data evaluation",
        "unit": "bits", "tolerance": TOL, "cases": report,
        "specified_error": {"predictions": predictions, "hypothetical_label": hypothetical_label},
        "checks": len(checks), "passed": checks,
    }, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')


if __name__ == '__main__':
    main()