把采样权重归一后，为什么平均结果反而偏了？

"""Enumerate all ordered IID draws from Q; compare two expectation estimators."""
import csv
import json
from fractions import Fraction as F
from itertools import product
from pathlib import Path

ROOT = Path(__file__).resolve().parent
TARGET = {"A": F(1, 2), "B": F(1, 2)}
VALUES = {"A": F(0), "B": F(1)}
PROPOSALS = {"balanced": {"A": F(1, 2), "B": F(1, 2)},
             "unequal": {"A": F(3, 4), "B": F(1, 4)},
             "rare_B": {"A": F(99, 100), "B": F(1, 100)}}
SIZES = (1, 2, 4)


def check_distributions(target, proposal, values):
    if set(target) != set(proposal) or set(target) != set(values):
        raise ValueError("state keys must match")
    if any(p < 0 for p in target.values()) or any(q < 0 for q in proposal.values()):
        raise ValueError("negative probability")
    if sum(target.values()) != 1 or sum(proposal.values()) != 1:
        raise ValueError("probabilities must sum to one")
    if any(target[x] > 0 and proposal[x] == 0 for x in target):
        raise ValueError("proposal misses target support")


def enumerate_case(name, proposal, n):
    check_distributions(TARGET, proposal, VALUES)
    target_mean = sum(TARGET[x] * VALUES[x] for x in TARGET)
    states = [x for x in proposal if proposal[x] > 0]
    rows = []
    for draws in product(states, repeat=n):
        probability = F(1)
        numerator, denominator = F(0), F(0)
        weights = []
        for x in draws:
            probability *= proposal[x]
            weight = TARGET[x] / proposal[x]
            numerator += weight * VALUES[x]
            denominator += weight
            weights.append(weight)
        if denominator <= 0:
            raise ValueError("self-normalized denominator is zero")
        ordinary, normalized = numerator / n, numerator / denominator
        assert 0 <= normalized <= 1
        # A common unknown positive normalization constant cancels in the ratio.
        assert normalized == (3 * numerator) / (3 * denominator)
        rows.append({"proposal": name, "n": n, "draws": "".join(draws), "probability": str(probability),
                     "weights": ";".join(map(str, weights)), "weighted_sum": str(numerator),
                     "weight_sum": str(denominator), "ordinary_estimate": str(ordinary),
                     "normalized_estimate": str(normalized)})
    assert sum(F(row["probability"]) for row in rows) == 1
    summaries = {}
    for method, field in (("ordinary", "ordinary_estimate"), ("self_normalized", "normalized_estimate")):
        mean = sum(F(row["probability"]) * F(row[field]) for row in rows)
        variance = sum(F(row["probability"]) * (F(row[field]) - mean) ** 2 for row in rows)
        mse = sum(F(row["probability"]) * (F(row[field]) - target_mean) ** 2 for row in rows)
        bias = mean - target_mean
        assert mse == variance + bias ** 2
        summaries[method] = {"mean": str(mean), "bias": str(bias), "variance": str(variance), "mse": str(mse)}
    assert F(summaries["ordinary"]["mean"]) == target_mean
    return rows, {"proposal": name, "n": n, "outcomes": len(rows), "target_mean": str(target_mean),
                  "probability_no_B": str(proposal["A"] ** n), "estimators": summaries}


def main():
    rows, summaries = [], []
    for name, proposal in PROPOSALS.items():
        for n in SIZES:
            emitted, summary = enumerate_case(name, proposal, n)
            rows.extend(emitted)
            summaries.append(summary)
    main_case = next(s for s in summaries if s["proposal"] == "unequal" and s["n"] == 2)
    assert main_case["estimators"]["ordinary"] == {"mean": "1/2", "bias": "0", "variance": "3/8", "mse": "3/8"}
    assert main_case["estimators"]["self_normalized"] == {"mean": "11/32", "bias": "-5/32", "variance": "159/1024", "mse": "23/128"}
    bad = {"A": F(1), "B": F(0)}
    try:
        check_distributions(TARGET, bad, VALUES)
    except ValueError as error:
        rejected = str(error)
    else:
        raise AssertionError("uncovered target support accepted")
    inputs = {"target_P": {k: str(v) for k, v in TARGET.items()}, "function_f": {k: str(v) for k, v in VALUES.items()},
              "proposals_Q": {name: {k: str(v) for k, v in q.items()} for name, q in PROPOSALS.items()},
              "sample_sizes": list(SIZES), "sampling": "ordered independent draws with replacement from Q"}
    report = {"scope": "Exact finite distributions; enumeration of sampling outcomes, no random simulation or learned density ratios",
              "weights": "P(x)/Q(x), P and Q normalized and exactly known",
              "cases": summaries, "sequence_rows": len(rows), "rejected_support": rejected,
              "consistency_note": "Finite-sample bias is not inconsistency; finite positive full support and IID sampling satisfy the ratio law-of-large-numbers argument"}
    (ROOT / "importance-inputs.json").write_text(json.dumps(inputs, indent=2) + "\n", encoding="utf-8")
    (ROOT / "importance-report.json").write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
    with (ROOT / "importance-sequences.csv").open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=list(rows[0]))
        writer.writeheader()
        writer.writerows(rows)
    print(f"PASS importance_ratio: {len(summaries)} cases, {len(rows)} ordered outcomes, missing-support proposal rejected")


if __name__ == "__main__":
    main()