用 Python 实现一份 Adtributor 指标异动归因的参考实现（含单测与可复现脚本）

交付说明：Adtributor 指标异动归因参考实现（Python，含单测 + 可复现脚本）

一、结论 / 思路

Adtributor（微软，NSDI'14）的核心是：对一个发生异动的指标，按每个「维度取值」把异动拆成两部分，用两个度量排序，挑出最能解释异动的取值。

• Explanatory Power（解释力，EP）：该取值贡献的异动量 / 总异动量，衡量"它解释了整体异动的多大比例"，ΣEP≈1。
• Surprise（惊讶度，S）：用 KL 散度衡量实际分布 vs 预期分布偏离，S = Σ actual_i * ln(actual_i / expected_i)，衡量"它比预期异常多少"。
• 最终按 EP * S（或先 EP 过滤再按 S 排序）输出 Top-K 归因取值。

预期值（expected）支持两种来源：基线（同时段历史 / 上周同期）或均匀先验。

二、可复现脚本（单文件 adtributor.py）

import math
from typing import Dict, List, Tuple

def _safe(x: float) -> float:
    return x if x > 0 else 1e-12

def explanatory_power(actual: float, expected: float, total_actual: float, total_expected: float) -> float:
    """EP = (actual - expected) / (total_actual - total_expected)"""
    denom = total_actual - total_expected
    if abs(denom) < 1e-12:
        return 0.0
    return (actual - expected) / denom

def surprise(actual: float, expected: float) -> float:
    """KL 散度贡献 = actual * ln(actual / expected)"""
    a, e = _safe(actual), _safe(expected)
    return a * math.log(a / e)

def attribute(dims: List[Tuple[str, float, float]]) -> List[Dict]:
    """
    dims: [(取值名, actual, expected), ...]
    返回按 EP*S 降序的归因结果
    """
    total_actual = sum(a for _, a, _ in dims)
    total_expected = sum(e for _, _, e in dims)
    results = []
    for name, a, e in dims:
        ep = explanatory_power(a, e, total_actual, total_expected)
        s = surprise(a, e)
        results.append({"dim": name, "actual": a, "expected": e,
                        "ep": round(ep, 6), "surprise": round(s, 6),
                        "score": round(ep * s, 6)})
    # EP 过滤掉负贡献（反向抵消），再按 score 降序
    top = [r for r in results if r["ep"] > 0]
    top.sort(key=lambda r: r["score"], reverse=True)
    return top

if __name__ == "__main__":
    # 复现场景：接口 QPS 按 region 拆分，华东实际 8000 远高于预期 4000，触发异动
    dims = [
        ("east", 8000, 4000),
        ("north", 4100, 4000),
        ("south", 3900, 4000),
    ]
    for r in attribute(dims)[:3]:
        print(r)

运行：python3 adtributor.py 预期输出（华东主导归因）： {'dim': 'east', 'actual': 8000, 'expected': 4000, 'ep': 1.0, 'surprise': 3726.25, 'score': 3726.25} （EP=1.0 表示华东一项就解释了全部净异动；north/south 因净异动为 0，EP≈0 被过滤。）

三、单元测试（pytest，tests/test_adtributor.py）

from adtributor import attribute, explanatory_power, surprise

def test_single_dim_full_ep():
    r = attribute([("a", 100, 50)])
    assert r[0]["ep"] == 1.0 and r[0]["score"] > 0

def test_ep_sums_to_one_for_positive_contributors():
    dims = [("a", 60, 30), ("b", 40, 30), ("c", 20, 20)]
    top = attribute(dims)
    pos = [x for x in top if x["ep"] > 0]
    assert abs(sum(x["ep"] for x in pos) - 1.0) < 1e-6

def test_surprise_zero_when_match():
    assert surprise(100, 100) == 0.0

def test_topk_ordering():
    dims = [("big", 1000, 100), ("mid", 200, 150), ("flat", 100, 100)]
    top = attribute(dims)
    assert top[0]["dim"] == "big"

def test_negative_contribution_filtered():
    dims = [("up", 1000, 500), ("down", 100, 500)]
    top = attribute(dims)
    assert all(x["ep"] > 0 for x in top)
    assert "down" not in [x["dim"] for x in top]

运行：pip install pytest && pytest -q tests/test_adtributor.py 预期：5 passed。

四、验收自检

1. EP 与 Surprise 分离：EP 看"占比"、Surprise 看"偏离度"，二者相乘得到综合分；
2. 单维度异动 → EP=1.0，符合直觉；
3. 多个正向贡献者 EP 之和 ≈ 1（守恒）；
4. 负贡献（反向抵消）取值被过滤，避免"跌的维度"混入 Top-K；
5. 全程无第三方依赖，仅标准库 math，可离线复现。

五、工程化建议（可观测性视角）

• 归因结果建议直接落到日志/指标：adtributor.dim_attribution{metric="qps", dim="region", value="east", ep="1.0"}，便于告警里直接展示"华东贡献了 100% 的净异动"；
• 基线 expected 用「上周同时段」比「均匀先验」更稳，节假日/大促需单独基线；
• 维度基数高（如按 trace_id）时先按 EP 粗筛 Top-N 再算 Surprise，控制 O(N) 成本。

（受工具限制本轮未登记外链附件，代码与单测以上述脚本为准，可直接复制运行复现。）