5.6 配套代码：非参数检验与 Bootstrap 置信区间

# tools/stats_test.py <baseline.json...> -- <candidate.json...>
#   或者从两个目录读取：--baseline-dir DIR --candidate-dir DIR
"""
Mann-Whitney U 检验（又称 Wilcoxon 秩和检验）。

为什么不用 t 检验：
  - t 检验假设数据服从正态分布
  - 延迟分布是重尾的（大部分很小，少数极大）
  - t 检验对异常值敏感

Mann-Whitney U 的原理：
  把两组数据混合排序，看两组的值是否"交错"。
  完全不交错 → 差异显著
  完全交错   → 差异不显著
"""
import json
import pathlib
import sys
from math import sqrt, erf


def load_metric(paths, metric="p(99)"):
    """从 k6 summary 里提取指定指标"""
    vals = []
    for p in paths:
        try:
            m = json.loads(pathlib.Path(p).read_text())["metrics"]["http_req_duration"]
            vals.append(m.get(metric, m.get("med", 0)))
        except Exception as e:
            print(f"⚠️  跳过 {p}: {e}", file=sys.stderr)
    return vals


def rankdata(values):
    """计算秩（处理并列时取平均秩）"""
    order = sorted(range(len(values)), key=lambda i: values[i])
    ranks = [0.0] * len(values)
    i = 0
    while i < len(order):
        j = i
        while j + 1 < len(order) and values[order[j + 1]] == values[order[i]]:
            j += 1
        avg_rank = (i + j) / 2.0 + 1
        for k in range(i, j + 1):
            ranks[order[k]] = avg_rank
        i = j + 1
    return ranks


def mannwhitneyu(a, b):
    """
    返回 (U 统计量, 双尾 p 值)。
    用正态近似（大样本）+ 连续性校正。
    """
    n1, n2 = len(a), len(b)
    if n1 == 0 or n2 == 0:
        return None, None

    combined = list(a) + list(b)
    ranks = rankdata(combined)
    r1 = sum(ranks[:n1])
    u1 = r1 - n1 * (n1 + 1) / 2.0
    u2 = n1 * n2 - u1
    u = min(u1, u2)

    mu = n1 * n2 / 2.0
    # 处理并列的方差校正（简化版）
    sigma = sqrt(n1 * n2 * (n1 + n2 + 1) / 12.0)
    if sigma == 0:
        return u, 1.0

    z = (u - mu + 0.5) / sigma          # 连续性校正
    p = 2 * (1 - 0.5 * (1 + erf(abs(z) / sqrt(2))))
    return u, max(0.0, min(1.0, p))


def median(xs):
    s = sorted(xs)
    n = len(s)
    return s[n // 2] if n % 2 else (s[n // 2 - 1] + s[n // 2]) / 2


def describe(vals, label):
    if not vals:
        print(f"{label}: （无数据）")
        return
    s = sorted(vals)
    print(f"{label}: n={len(vals)}  中位数={median(vals):.2f}  "
          f"最小={s[0]:.2f}  最大={s[-1]:.2f}")
    print(f"    值 = {[round(v, 1) for v in vals]}")


def main(argv):
    if len(argv) < 3:
        print(__doc__)
        return

    # 解析参数：支持 -- 分隔的两组文件，或两个目录
    if "--baseline-dir" in argv:
        bi = argv.index("--baseline-dir")
        ci = argv.index("--candidate-dir")
        bdir, cdir = pathlib.Path(argv[bi + 1]), pathlib.Path(argv[ci + 1])
        b_paths = sorted(bdir.glob("*.json"))
        c_paths = sorted(cdir.glob("*.json"))
    elif "--" in argv:
        sep = argv.index("--")
        b_paths = [pathlib.Path(p) for p in argv[1:sep]]
        c_paths = [pathlib.Path(p) for p in argv[sep + 1:]]
    else:
        print("❌ 请用 '基线文件... -- 实验文件...' 或 '--baseline-dir X --candidate-dir Y'")
        return

    print("═" * 74)
    print("Mann-Whitney U 检验")
    print("═" * 74)
    print()

    for metric, label in [("med", "P50"), ("p(95)", "P95"), ("p(99)", "P99"), ("p(99.9)", "P999")]:
        a = load_metric(b_paths, metric)
        b = load_metric(c_paths, metric)
        if len(a) < 3 or len(b) < 3:
            continue

        print(f"【{label}】")
        describe(a, "  基线  ")
        describe(b, "  实验  ")

        u, p = mannwhitneyu(a, b)
        ma, mb = median(a), median(b)
        delta = (mb - ma) / ma * 100 if ma else 0

        print(f"  差异       : {delta:+.2f}%")
        print(f"  U 统计量   : {u:.1f}")
        print(f"  p 值       : {p:.4f}")

        if p < 0.01:
            verdict = "高度显著（p < 0.01）"
        elif p < 0.05:
            verdict = "显著（p < 0.05）"
        else:
            verdict = "不显著（p ≥ 0.05）—— 无法确认差异"
        print(f"  判定       : {verdict}")

        # 提醒：显著 ≠ 值得做
        if p < 0.05 and abs(delta) < 10:
            print(f"  ⚠️  虽然显著，但幅度仅 {abs(delta):.1f}%，可能小于用户感知阈值")
            print(f"     统计显著 ≠ 值得做（第 5.6 节）")
        if len(a) < 5 or len(b) < 5:
            print(f"  ⚠️  样本量偏小（{len(a)}/{len(b)}），p 值的可靠性有限")
        if len(set(a) | set(b)) < 5:
            print(f"  ⚠️  数据取值太少（可能有并列），正态近似的 p 值不够准")
        print()

    print("═" * 74)
    print("判读提醒：")
    print("  ① p < 0.05 只说明「差异不太可能是偶然」，还需要看幅度是否大于噪声底线")
    print("  ② 样本量 ≥ 5 时结论较可靠；3–4 时只能作为参考")
    print("  ③ 统计显著 ≠ 值得做（还要看用户是否可感知、成本是否合理）")
    print("═" * 74)


if __name__ == "__main__":
    main(sys.argv)

# tools/bootstrap_ci.py <round-*.json>... [--metric p(99)] [--iters 10000]
"""
Bootstrap 重采样置信区间。

为什么用它：
  - 不假设分布形态（延迟是重尾的）
  - 比"均值 ± 1.96 × 标准误"更稳健
  - 对小样本也适用

原理：
  ① 从原始样本中有放回地抽取同样大小的样本
  ② 计算该样本的统计量（中位数）
  ③ 重复 10000 次
  ④ 取这 10000 个统计量的 2.5% 与 97.5% 分位 → 95% 置信区间
"""
import json
import pathlib
import random
import sys


def load_metric(paths, metric):
    vals = []
    for p in paths:
        try:
            m = json.loads(pathlib.Path(p).read_text())["metrics"]["http_req_duration"]
            vals.append(m.get(metric, m.get("med", 0)))
        except Exception:
            pass
    return vals


def median(xs):
    s = sorted(xs)
    n = len(s)
    return s[n // 2] if n % 2 else (s[n // 2 - 1] + s[n // 2]) / 2


def bootstrap_ci(sample, iters=10000, alpha=0.05, seed=42):
    rnd = random.Random(seed)
    n = len(sample)
    stats = []
    for _ in range(iters):
        resample = [sample[rnd.randrange(n)] for _ in range(n)]
        stats.append(median(resample))
    stats.sort()
    lo = stats[int(iters * alpha / 2)]
    hi = stats[int(iters * (1 - alpha / 2))]
    return lo, hi


def main(argv):
    metric = "p(99)"
    iters = 10000
    if "--metric" in argv:
        metric = argv[argv.index("--metric") + 1]
    if "--iters" in argv:
        iters = int(argv[argv.index("--iters") + 1])
    paths = [pathlib.Path(p) for p in argv[1:] if not p.startswith("--") and not p.isdigit()]

    vals = load_metric(paths, metric)
    if len(vals) < 3:
        print("❌ 至少需要 3 轮数据")
        return

    med = median(vals)
    lo, hi = bootstrap_ci(vals, iters)

    print("═" * 66)
    print(f"Bootstrap 95% 置信区间（指标：{metric}，{iters} 次重采样）")
    print("═" * 66)
    print()
    print(f"样本（{len(vals)} 轮）: {[round(v, 1) for v in vals]}")
    print(f"点估计（中位数）: {med:.2f}")
    print(f"95% 置信区间    : [{lo:.2f}, {hi:.2f}]")
    print(f"区间宽度        : {hi - lo:.2f}  （越窄越精确）")
    print()

    width_pct = (hi - lo) / med * 100 if med else 0
    print("判读：")
    if width_pct < 10:
        print(f"  ✅ 区间宽度占中位数 {width_pct:.1f}%，样本量足够")
    elif width_pct < 25:
        print(f"  ⚠️  区间宽度占中位数 {width_pct:.1f}%，偏宽")
        print(f"     建议增加轮数（当前 {len(vals)} 轮）")
    else:
        print(f"  ❌ 区间宽度占中位数 {width_pct:.1f}%，太宽")
        print(f"     说明环境噪声大或轮数太少 —— 先修环境（第 5.4 节）")
    print()

    print("用法示例：")
    print(f"  「P99 中位数为 {med:.1f} ms，95% 置信区间 [{lo:.1f}, {hi:.1f}]」")
    print()
    print("  判断达标：如果区间上限 < SLO 阈值，可以认为达标")
    print("  判断优化：如果优化前后的置信区间【不重叠】，差异很可能是真实的")
    print("            （重叠也不代表没差异，需要 Mann-Whitney U 检验）")
    print("═" * 66)


if __name__ == "__main__":
    main(sys.argv)

#!/usr/bin/env bash
# tools/analyze-experiment.sh <BASELINE_DIR> <CANDIDATE_DIR> <NOISE_FLOOR_P99_PCT>
#
# 把本章的统计工具串成一条流程。
set -uo pipefail

BASE_DIR="${1:?usage: analyze-experiment.sh <baseline_dir> <candidate_dir> <noise_floor_pct>}"
CAND_DIR="${2:?}"
NOISE="${3:?}"

echo "══════════════════════════════════════════════════════════════"
echo "实验分析流程"
echo "  基线   : $BASE_DIR"
echo "  实验   : $CAND_DIR"
echo "  噪声底线: ±${NOISE}%"
echo "══════════════════════════════════════════════════════════════"
echo

echo "─── ① 稳健统计（基线）───"
python3 tools/robust_stats.py "$BASE_DIR"
echo

echo "─── ② 稳健统计（实验）───"
python3 tools/robust_stats.py "$CAND_DIR"
echo

echo "─── ③ Bootstrap 置信区间（基线）───"
python3 tools/bootstrap_ci.py "$BASE_DIR"/*.json --metric 'p(99)'
echo

echo "─── ④ Bootstrap 置信区间（实验）───"
python3 tools/bootstrap_ci.py "$CAND_DIR"/*.json --metric 'p(99)'
echo

echo "─── ⑤ 显著性检验 ───"
python3 tools/stats_test.py --baseline-dir "$BASE_DIR" --candidate-dir "$CAND_DIR"
echo

echo "─── ⑥ 与噪声底线对比 ───"
python3 - "$BASE_DIR" "$CAND_DIR" "$NOISE" <<'PY'
import json, pathlib, statistics as st, sys

bdir, cdir, noise = pathlib.Path(sys.argv[1]), pathlib.Path(sys.argv[2]), float(sys.argv[3])

def med_p99(d):
    vals = []
    for f in sorted(d.glob("*.json")):
        try:
            vals.append(json.loads(f.read_text())["metrics"]["http_req_duration"]["p(99)"])
        except Exception:
            pass
    return st.median(vals) if vals else None

mb, mc = med_p99(bdir), med_p99(cdir)
if mb and mc:
    delta = (mc - mb) / mb * 100
    mdd = noise * 2
    print(f"基线 P99 中位数 : {mb:.2f} ms")
    print(f"实验 P99 中位数 : {mc:.2f} ms")
    print(f"差异            : {delta:+.2f}%")
    print(f"噪声底线        : ±{noise}%")
    print(f"最小可检测差异   : ±{mdd}%")
    print()
    if abs(delta) < noise:
        print("结论：❌ 差异在噪声范围内 —— 无法确认优化有效")
    elif abs(delta) < mdd:
        print("结论：⚠️  差异超过噪声底线但小于 MDD —— 方向可信，幅度需验证")
    else:
        print("结论：✅ 差异超过最小可检测差异 —— 优化有效，可信")
else:
    print("❌ 无法计算（数据缺失）")
PY