7.8 配套代码：三问纪律与收益报告

#!/usr/bin/env bash
# tools/optimization-three-questions.sh
#
# 每项优化都必须回答的三个问题。
# 缺任何一问，这次改动的证据链就是断的。
set -uo pipefail

echo "═══════════════════════════════════════════════════════════════"
echo "优化的三问检查"
echo "═══════════════════════════════════════════════════════════════"
echo

FAIL=0

ask_required() {
  local question="$1"
  local hint="$2"
  echo "── $question"
  echo "   （$hint）"
  read -r -p "   已完成？[y/N] " ans
  if [[ "${ans,,}" != "y" ]]; then
    echo "   ❌ 未完成 —— 这一问不能跳过"
    FAIL=$((FAIL+1))
  else
    echo "   ✅"
  fi
  echo
}

echo "【第一问：提升多少？】"
echo
ask_required "有 before/after 数据吗？" \
  "必须是同一套脚本、同一环境、≥3 轮"
ask_required "报了中位数与百分位（不是平均值）吗？" \
  "至少 P50/P95/P99 + 错误率"
ask_required "做了统计检验吗？" \
  "Mann-Whitney U 或 Bootstrap 置信区间（第 5 章）"
ask_required "与噪声底线/MDD 对比了吗？" \
  "差异必须超过 MDD 才可信（第 5 章 5.7）"

echo "【第二问：代价是什么？】"
echo
ask_required "显式列出了代价吗？" \
  "复杂度/一致性/资源/风险 —— 代价不会自己消失"
ask_required "有降级方案吗？" \
  "新组件挂了会怎样？服务还能用吗？"

echo "【第三问：回归验证了吗？】"
echo
ask_required "功能正确性验证了吗？" \
  "单元测试 + 关键路径手工验证"
ask_required "性能回归验证了吗？" \
  "⚠️ 优化 A 可能让 B 变慢 —— 用同一套脚本复跑基线"
ask_required "长稳（浸泡）验证了吗？" \
  "引入新组件或改并发模型时必须做（第 3 章 3.5）"

echo "═══════════════════════════════════════════════════════════════"
if [ "$FAIL" -eq 0 ]; then
  echo "✅ 三问全部通过 —— 这次优化可以写进报告"
else
  echo "❌ 有 $FAIL 项未完成"
  echo
  echo "提醒：答不出第三问的优化，本质上是【技术债】而不是优化。"
  echo "      它会在此后的每一次故障排查中增加不确定性。"
fi
echo "═══════════════════════════════════════════════════════════════"

# tools/generate-optimization-report.py <optimizations.json> [--out report.md]
"""
从结构化数据生成优化收益报告。

输入格式：
{
  "title": "订单查询接口优化",
  "root_cause": "缺索引导致全表扫描（贡献占比 72%）",
  "experiments": {"before": "E03", "after": "E08"},
  "noise_floor_pct": 13.2,
  "mdd_pct": 26.4,
  "optimizations": [
    {
      "name": "给 status 加索引",
      "level": 2,
      "level_name": "② 算法与数据访问",
      "changes": "CREATE INDEX idx_orders_status_created ON orders(status, created_at DESC)",
      "rationale": "消除全表扫描，Rows Removed by Filter 从 900000 降到 200",
      "metrics": {
        "before": {"p50": 20.1, "p99": 210.4, "qps": 498, "error_rate": 0.0005},
        "after":  {"p50": 8.3,  "p99": 95.2,  "qps": 499, "error_rate": 0.0003}
      },
      "p_value": 0.008,
      "costs": [
        "索引占用额外磁盘（约 120 MB）",
        "写入时有轻微的索引维护开销（实测 < 2%）"
      ],
      "rollback": "DROP INDEX idx_orders_status_created",
      "regressions": {
        "functional": "单元测试全通过；手工验证了 5 个关键路径",
        "performance": "其他接口 P99 变化 < 3%（在噪声范围内）",
        "soak": "40 分钟浸泡，内存与连接数稳定"
      }
    }
  ],
  "rejected": [
    {
      "name": "加 Redis 缓存",
      "expected_benefit": "P50 -70%",
      "reason": "批量查询已解决主要问题；key 分散导致命中率预估仅 40%，P99 不会改善；引入一致性风险",
      "reconsider_when": "命中率能到 90% 以上，或数据量增长导致即使批量查询也变慢"
    }
  ],
  "not_covered": ["冷启动路径", "数据量增长 10 倍后的表现", "多副本部署"],
  "next_steps": ["监控 P99 一周，确认稳定", "把索引加入部署脚本"]
}
"""
import json
import sys
from datetime import datetime


def fmt_delta(before, after):
    if before == 0:
        return "n/a"
    return f"{(after - before) / before * 100:+.1f}%"


def generate(data):
    noise = data.get("noise_floor_pct", 0)
    mdd = data.get("mdd_pct", 0)

    lines = [
        f"# 优化收益报告：{data['title']}",
        "",
        f"> 生成时间：{datetime.now().isoformat(timespec='seconds')}　|　"
        f"关联实验：{' – '.join(data.get('experiments', {}).values())}",
        "",
        "---",
        "",
        "## 1. 问题与根因",
        "",
        data["root_cause"],
        "",
        "---",
        "",
    ]

    for i, opt in enumerate(data["optimizations"], 1):
        lines += [
            f"## 2.{i} 优化措施 {i}：{opt['name']}",
            "",
            f"- **金字塔层级**：{opt.get('level_name', opt.get('level', '?'))}",
            f"- **改动**：`{opt['changes']}`",
            f"- **原理**：{opt['rationale']}",
            "",
            "### 第一问：提升多少",
            "",
            "| 指标 | before | after | 变化 |",
            "| --- | --- | --- | --- |",
        ]

        b, a = opt["metrics"]["before"], opt["metrics"]["after"]
        for key, label in [("p50", "P50 (ms)"), ("p99", "P99 (ms)"),
                           ("qps", "QPS"), ("error_rate", "错误率")]:
            if key in b and key in a:
                bv, av = b[key], a[key]
                if key == "error_rate":
                    bv_s, av_s = f"{bv:.2%}", f"{av:.2%}"
                    delta = f"{(av - bv) * 100:+.2f}pp"
                elif key == "qps":
                    bv_s, av_s = f"{bv:.1f}", f"{av:.1f}"
                    delta = fmt_delta(bv, av)
                else:
                    bv_s, av_s = f"{bv:.1f}", f"{av:.1f}"
                    delta = fmt_delta(bv, av)
                lines.append(f"| {label} | {bv_s} | {av_s} | {delta} |")

        lines += ["", "**显著性判定**：", ""]
        lines.append(f"- 环境噪声底线：±{noise}%")
        lines.append(f"- 最小可检测差异（MDD）：±{mdd}%")
        if "p_value" in opt:
            p = opt["p_value"]
            sig = "✅ 显著" if p < 0.05 else "❌ 不显著"
            lines.append(f"- 统计检验：Mann-Whitney U，p = {p}（{sig}）")

        # 判断可信度
        if "p99" in b and "p99" in a:
            delta_pct = (a["p99"] - b["p99"]) / b["p99"] * 100
            if abs(delta_pct) < noise:
                lines.append(f"- **结论**：⚠️ 变化 {delta_pct:+.1f}% 在噪声范围内，**无法确认优化有效**")
            elif abs(delta_pct) < mdd:
                lines.append(f"- **结论**：⚠️ 变化 {delta_pct:+.1f}% 超过噪声底线但小于 MDD，方向可信、幅度需验证")
            else:
                lines.append(f"- **结论**：✅ 变化 {delta_pct:+.1f}%，超过 MDD，**优化有效且可信**")

        lines += [
            "",
            "### 第二问：代价是什么",
            "",
        ]
        for c in opt.get("costs", []):
            lines.append(f"- {c}")
        if opt.get("rollback"):
            lines += ["", f"**回滚方式**：`{opt['rollback']}`"]

        lines += [
            "",
            "### 第三问：回归验证",
            "",
        ]
        reg = opt.get("regressions", {})
        lines.append(f"- [{'x' if reg.get('functional') else ' '}] **功能正确性**：{reg.get('functional', '未验证')}")
        lines.append(f"- [{'x' if reg.get('performance') else ' '}] **性能回归**：{reg.get('performance', '未验证')}")
        lines.append(f"- [{'x' if reg.get('soak') else ' '}] **长稳（浸泡）**：{reg.get('soak', '未验证')}")
        lines += ["", "---", ""]

    # 被否决的方案
    if data.get("rejected"):
        lines += [
            "## 3. 被否决的方案",
            "",
            "| 方案 | 预期收益 | 否决理由 | 重新评估的条件 |",
            "| --- | --- | --- | --- |",
        ]
        for r in data["rejected"]:
            lines.append(
                f"| {r['name']} | {r['expected_benefit']} | {r['reason']} | {r.get('reconsider_when', '—')} |")
        lines += ["", "---", ""]

    # 未覆盖场景
    lines += ["## 4. 未覆盖的场景", ""]
    for s in data.get("not_covered", []):
        lines.append(f"- {s}")
    lines += ["", "---", ""]

    # 后续
    lines += ["## 5. 后续建议", ""]
    for s in data.get("next_steps", []):
        lines.append(f"- {s}")
    lines.append("")

    return "\n".join(lines)


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(__doc__)
        sys.exit(1)

    data = json.loads(open(sys.argv[1]).read())
    report = generate(data)

    if "--out" in sys.argv:
        out = sys.argv[sys.argv.index("--out") + 1]
        open(out, "w", encoding="utf-8").write(report)
        print(f"✅ 报告已生成 → {out}")
    else:
        print(report)

# tools/rejection-log.py
"""
管理「被否决的方案」——它们不是垃圾，而是有价值的决策记录。

用法：
    python3 tools/rejection-log.py add <EXP_ID> "<方案>" "<预期收益>" "<否决理由>" "<重新评估条件>"
    python3 tools/rejection-log.py list <EXP_ID>
    python3 tools/rejection-log.py review <EXP_ID>    # 检查是否有需要重新评估的
"""
import json
import pathlib
import sys
from datetime import datetime


def log_path(exp_id):
    d = pathlib.Path("docs/experiments") / exp_id
    d.mkdir(parents=True, exist_ok=True)
    return d / "rejections.json"


def load(exp_id):
    p = log_path(exp_id)
    return json.loads(p.read_text(encoding="utf-8")) if p.exists() else {"rejections": []}


def save(exp_id, data):
    log_path(exp_id).write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")


def cmd_add(exp_id, name, benefit, reason, reconsider):
    data = load(exp_id)
    data["rejections"].append({
        "id": f"R{len(data['rejections']) + 1}",
        "name": name,
        "expected_benefit": benefit,
        "reason": reason,
        "reconsider_when": reconsider,
        "date": datetime.now().isoformat(timespec="seconds"),
    })
    save(exp_id, data)
    print(f"✅ 已记录否决：{name}")


def cmd_list(exp_id):
    data = load(exp_id)
    if not data["rejections"]:
        print("（无否决记录）")
        return
    print("═" * 88)
    print(f"被否决的方案：{exp_id}")
    print("═" * 88)
    print()
    for r in data["rejections"]:
        print(f"【{r['id']}】{r['name']}　（{r['date']}）")
        print(f"  预期收益　　: {r['expected_benefit']}")
        print(f"  否决理由　　: {r['reason']}")
        print(f"  重新评估条件: {r['reconsider_when']}")
        print()


def cmd_review(exp_id):
    """检查是否有需要重新评估的否决"""
    data = load(exp_id)
    print("═" * 78)
    print("否决记录复查")
    print("═" * 78)
    print()
    print("提醒：以下否决是【有条件的】——如果条件已经变化，应该重新评估。")
    print()
    for r in data["rejections"]:
        print(f"  {r['id']} {r['name']}")
        print(f"     → 当「{r['reconsider_when']}」时，重新评估")
        print()
    print("═" * 78)
    print("为什么要复查：")
    print("  ① 数据量可能增长（原来命中率低的缓存现在可能有效）")
    print("  ② 业务可能变化（原来不需要实时，现在需要）")
    print("  ③ 技术可能演进（新的序列化器、新的 GC）")
    print("  ④ 瓶颈可能转移（原来占 5% 的部分现在占 40%）")
    print("═" * 78)


if __name__ == "__main__":
    if len(sys.argv) < 3:
        print(__doc__)
        sys.exit(1)
    cmd, exp = sys.argv[1], sys.argv[2]
    if cmd == "add" and len(sys.argv) >= 7:
        cmd_add(exp, sys.argv[3], sys.argv[4], sys.argv[5], sys.argv[6])
    elif cmd == "list":
        cmd_list(exp)
    elif cmd == "review":
        cmd_review(exp)
    else:
        print(__doc__)

# 记录一个否决
python3 tools/rejection-log.py add E08-orders-opt \
  "加 Redis 缓存" \
  "P50 -70%" \
  "批量查询已解决主要问题；key 分散导致命中率预估仅 40%，P99 不会改善；引入一致性风险" \
  "命中率能到 90% 以上，或数据量增长导致批量查询也变慢"

# 查看
python3 tools/rejection-log.py list E08-orders-opt

# 定期复查（每季度）
python3 tools/rejection-log.py review E08-orders-opt

#!/usr/bin/env bash
# tools/optimization-workflow.sh <EXP_ID>
#
# 优化工作的完整流程：三问 → 报告 → 否决记录
set -uo pipefail

EXP_ID="${1:?usage: optimization-workflow.sh <exp_id>}"
DIR="docs/experiments/${EXP_ID}"
mkdir -p "$DIR"

echo "═══════════════════════════════════════════════════════════════"
echo "优化工作流程：$EXP_ID"
echo "═══════════════════════════════════════════════════════════════"
echo

echo "【步骤 1】三问检查"
tools/optimization-three-questions.sh
echo

echo "【步骤 2】记录被否决的方案"
echo "  即使在实施优化时，也要记录【没有采纳】的方案："
echo "    python3 tools/rejection-log.py add $EXP_ID \"方案\" \"预期收益\" \"否决理由\" \"重新评估条件\""
echo

echo "【步骤 3】生成收益报告"
echo "  1. 创建 $DIR/optimizations.json（参考 tools/generate-optimization-report.py 的格式）"
echo "  2. 运行：python3 tools/generate-optimization-report.py $DIR/optimizations.json --out $DIR/REPORT.md"
echo

echo "【步骤 4】收尾检查"
echo "  □ 每项优化都有三问"
echo "  □ 至少记录了一项被否决的方案"
echo "  □ 报告里有「未覆盖的场景」"
echo "  □ 有效的优化已固化（加了监控或门禁）"
echo

echo "═══════════════════════════════════════════════════════════════"