差异检验拿不到 p 值照实说、敏感性只在真实的各组里找离群值(复核 R8、R9):一对都拿不到 p 值(比率有一组样本量为 0;均值、逐行有一组不到 2 个样本,或均值、标准差为空)时算不了,契约 REASONS 加 untestable,原因写哪一组怎么了(「华东只有 1 个样本、华南只有 1 个样本,算不了差异」),不再判 ok + 可能只是样本波动;只有一部分拿不到的,那几对不算进比了几对、不参与校正,表里写清哪一组怎么了,判定语后接「另有 n 对算不了(…)」(契约 CAVEATS 加 untested,排在判定语附带说明的前面)。逐行模式比三组以上时,离均值最远的那一笔只在真实的各组里找(原来池子里混着各对的「其余各组合计」,第一组以外的离群值删不到它自己那一组,漏报 outlier_sensitive),去掉以后由各组重建每一对的「其余」合计再检验

This commit is contained in:
ThearchyHelios
2026-10-21 15:56:16 +08:00
parent 460af221f9
commit 427a6df23c
4 changed files with 104 additions and 30 deletions
+4 -2
View File
@@ -85,16 +85,18 @@ VERDICTS: dict[str, tuple[str, ...]] = {
}
#: 算不了的原因(细案 01 §8.2)。算不了也记成结论:写作者可以照实写「历史只有 5 周,判断不了是否异常」。最后两种是分析卡
#: 里随数据变化的调用错误记成的(analysis.run 的 refuse_data_errors,见 app/analysis 的 DATA_ERRORS):parent_refused 上一层
#: 没做成;input_unavailable 这次的数据里没有要用的那一项(下钻的段、期次、总数那一格、要比的组……,reason 写是哪一项、哪个值)
#: 没做成;input_unavailable 这次的数据里没有要用的那一项(下钻的段、期次、总数那一格、要比的组……,reason 写是哪一项、哪个值)。
#: untestable:差异检验一对都拿不到 p 值(两组各 1 个样本、样本量为 0……)
REASONS: tuple[str, ...] = (
"truncated", "too_few_segments", "history_short", "partial_target", "target_gap", "series_gap",
"period_unreadable", "zero_base", "non_positive", "incomparable_sides", "parent_refused", "input_unavailable",
"untestable",
)
#: 附带说明的种类(caveats[].code)
CAVEATS: tuple[str, ...] = (
"offset", "simpson", "small_denominator", "record_high", "record_low", "weekday_effect",
"multiple_comparisons", "outlier_sensitive", "small_sample", "measure_unlabeled", "null_values",
"headline_basis", "comparability", "partial_periods",
"headline_basis", "comparability", "partial_periods", "untested",
)
# --------------------------------------------------------------------------
+51 -24
View File
@@ -16,6 +16,10 @@
按 A − B 的正负);否则 no_evidence。用了 Fisher(比率)或有一组样本数小于 min_n(均值)→ small_sample。判定语不出现「p 值」
「显著性」:differs 说「差异站得住」,no_evidence 说「差异可能只是样本波动」,p 值、区间在表和证据面板里。
拿不到 p 值的一对(比率有一组样本量为 0;均值、逐行有一组不到 2 个样本,或均值、标准差为空)不算进比了几对、不参与校正,表里
照实写哪一组怎么了;一对都拿不到时算不了(untestable),不硬判「可能只是样本波动」;只有一部分拿不到的,判定语后接一句
「另有 n 对算不了」(untested)。
subjects 是判定说的那一对 [[A], [B]](只比一对时就是那一对,比了几对时是校正后最站得住的那一对;几对都不站得住时为空),
direction 是 A 相对 B。
"""
@@ -80,9 +84,13 @@ def _int(value: Fraction | None, snap: Snap, r: int, column: str) -> int:
return int(value)
#: 「其余各组合计」的键(不是分组列里的值)
REST_KEY = ("__rest__",)
def _pool(name: str, groups: Sequence[Group], mode: str) -> Group:
"""「其余各组合计」:比率把成功数、试验数加起来;均值按各组的均值、方差、样本数合并;逐行的值并在一起。"""
out = Group(("__rest__",), name, rows=[r for g in groups for r in g.rows])
out = Group(REST_KEY, name, rows=[r for g in groups for r in g.rows])
if mode == "rate":
out.x, out.n = sum(g.x for g in groups), sum(g.n for g in groups)
return out
@@ -137,6 +145,22 @@ def _test(pair: Pair, mode: str, min_expected: Fraction) -> None:
pair.ci = stats.welch_interval(float(a.mean), float(a.var), a.n, float(b.mean), float(b.var), b.n)
def _gap(group: Group, mode: str) -> str | None:
"""一组为什么算不了检验;算得了为 None。"""
if mode == "rate":
return ph.say(group.name, "的样本量为 0") if group.n == 0 else None
if group.n < 2:
return ph.say(group.name, "没有样本" if group.n == 0 else "只有 1 个样本")
if group.mean is None or group.var is None:
return ph.say(group.name, "的均值或标准差为空")
return None
def untested_text(pair: Pair, mode: str) -> str:
"""这一对拿不到 p 值的原因:哪一组、怎么了(两组都有就都写)。"""
return "、".join(dict.fromkeys(why for g in (pair.a, pair.b) if (why := _gap(g, mode)))) or "样本不够"
def _verdict(pairs: Sequence[Pair], alpha: Fraction) -> list[Pair]:
"""Holm 校正后站得住的那几对(按校正后的 p 从小到大,一样的按先后)。"""
adjusted = stats.holm([p.p for p in pairs])
@@ -295,6 +319,12 @@ def _decide(env: Env, params: dict[str, Any], used: list[str], pairs: list[Pair]
min_expected = env.threshold("min_expected") if mode == "rate" else Fraction(0)
for pair in pairs:
_test(pair, mode, min_expected)
tested = [p for p in pairs if p.p is not None]
untested = [p for p in pairs if p.p is None]
if not tested:
shown = list(dict.fromkeys(untested_text(p, mode) for p in pairs))
return refuse(env, params, used, "untestable",
",".join(shown[:3]) + ("等" if len(shown) > 3 else "") + ",算不了差异", criterion=criterion)
hits = _verdict(pairs, alpha)
caveats: list[dict[str, str]] = []
if mode == "rows":
@@ -309,24 +339,26 @@ def _decide(env: Env, params: dict[str, Any], used: list[str], pairs: list[Pair]
small = sorted((g for p in pairs for g in (p.a, p.b) if g.n < min_n), key=lambda g: g.n)
if small:
caveats.append(ph.caveat("small_sample", ph.small_sample_mean_text(small[0].name, small[0].n, min_n)))
tested = [p for p in pairs if p.p is not None]
if len(tested) > 1:
caveats.append(ph.caveat("multiple_comparisons", ph.multiple_text(len(tested))))
if untested:
caveats.append(ph.caveat("untested", ph.untested_pairs_text(len(untested),
[untested_text(p, mode) for p in untested])))
fmt = (lambda v: ph.pct(v, signed=False)) if mode == "rate" else (lambda v: ph.amount(v))
if hits:
code = "differs"
head = hits[0]
main = ph.differ_phrase([ph.differ_pair(p.a.name, p.b.name, label, fmt(p.a_value), fmt(p.b_value),
p.diff > 0) for p in hits], # type: ignore[operator]
len(pairs))
len(tested))
subjects = [[head.a.name], [head.b.name]]
direction = sign_word(head.diff) # type: ignore[arg-type]
else:
code = "no_evidence"
head = pairs[0] if len(pairs) == 1 else None
head = tested[0] if len(tested) == 1 else None
sizes = (ph.amount(head.a.n), ph.amount(head.b.n)) if head is not None else None
main = ph.no_evidence_phrase(head.a.name if head else None, head.b.name if head else None, label, sizes,
len(pairs))
len(tested))
subjects = [[head.a.name], [head.b.name]] if head is not None else []
direction = None
value_name = "比率" if mode == "rate" else "均值"
@@ -334,7 +366,7 @@ def _decide(env: Env, params: dict[str, Any], used: list[str], pairs: list[Pair]
for p in pairs:
flags = []
if p.p is None:
flags.append("样本不够,算不了")
flags.append(f"{untested_text(p, mode)},算不了")
if p.test == "fisher":
flags.append("样本少,用了精确检验")
rows.append([p.a.name, p.b.name, canon(p.a_value), canon(p.b_value), canon(p.diff),
@@ -342,7 +374,7 @@ def _decide(env: Env, params: dict[str, Any], used: list[str], pairs: list[Pair]
TESTS.get(p.test or "") or None, canon(p.p), canon(p.p_adj), "、".join(flags) or None])
columns = ["a", "b", "a_value", "b_value", "diff", "ci_lo", "ci_hi", "a_n", "b_n", "test", "p", "p_adj", "flag"]
# 组、对照组照分组列(期次列)在快照里的类型;和「其余各组合计」比时对照组那一列有一格是叫法,写 text
rest = any(p.b.key == ("__rest__",) for p in pairs)
rest = any(p.b.key == REST_KEY for p in pairs)
table = {"columns": columns,
"labels": {"a": "组", "b": "对照组", "a_value": f"组的{value_name}", "b_value": f"对照组的{value_name}",
"diff": "差", "ci_lo": "差的区间下限", "ci_hi": "差的区间上限", "a_n": "组的样本量",
@@ -350,37 +382,32 @@ def _decide(env: Env, params: dict[str, Any], used: list[str], pairs: list[Pair]
"rows": rows,
"column_types": column_types(columns, dims={"a": kind, "b": "text" if rest else kind},
texts=("test", "flag"))}
shown = head if head is not None else pairs[0]
shown = head if head is not None else tested[0]
values = {"a": canon(shown.a_value), "b": canon(shown.b_value), "diff": canon(shown.diff),
"ci_lo": canon(shown.ci[0]) if shown.ci else None, "ci_hi": canon(shown.ci[1]) if shown.ci else None,
"p": canon(shown.p), "p_adj": canon(shown.p_adj), "n_a": str(shown.a.n), "n_b": str(shown.b.n),
"test": shown.test, "pairs": str(len(pairs)), "differs": str(len(hits))}
verdict = {"code": code, "subjects": subjects, "direction": direction}
phrase = ph.with_caveats(main, caveats, skip=("multiple_comparisons",) if len(pairs) > 1 else ())
phrase = ph.with_caveats(main, caveats, skip=("multiple_comparisons",) if len(tested) > 1 else ())
return build(env, params, used, table=table, values=values, verdict=verdict, criterion=criterion, phrase=phrase,
caveats=caveats)
def _sensitivity(pairs: list[Pair], mode: str, alpha: Fraction, min_expected: Fraction,
differs: bool) -> tuple[bool, bool] | None:
"""rows 模式的敏感性:去掉比较的各组里离均值最远的那一个值重算,判定变了交回 (去掉的是不是偏大的那一笔, 现在站不站得住)。"""
everything = [(g, i, v) for g in dict.fromkeys(g for p in pairs for g in (p.a, p.b)) for i, v in enumerate(g.values)]
"""rows 模式的敏感性:去掉比较的各组里离均值最远的那一个值重算,判定变了交回 (去掉的是不是偏大的那一笔, 现在站不站得住)。
只在真实的各组里找(「其余各组合计」是它们的值并在一起,不另算一份),去掉以后由各组重建每一对的「其余」合计。"""
groups = list(dict.fromkeys(g for p in pairs for g in (p.a, p.b) if g.key != REST_KEY))
everything = [(g, i, v) for g in groups for i, v in enumerate(g.values)]
if len(everything) < 3:
return None
mean = sum((v for _, _, v in everything), Fraction(0)) / len(everything)
group, index, value = max(everything, key=lambda t: abs(t[2] - mean))
trimmed = []
for p in pairs:
def cut(g: Group) -> Group:
if g is not group and not (g.key == ("__rest__",) and any(v is value for v in g.values)):
return g
values = list(g.values)
for k, v in enumerate(values):
if v is value:
del values[k]
break
return _moments(Group(g.key, g.name, rows=g.rows, values=values))
trimmed.append(Pair(cut(p.a), cut(p.b)))
cut = {g: g for g in groups}
cut[group] = _moments(Group(group.key, group.name, rows=group.rows,
values=[v for k, v in enumerate(group.values) if k != index]))
trimmed = [Pair(cut[p.a], _pool(p.b.name, [cut[g] for g in groups if g is not p.a], mode)
if p.b.key == REST_KEY else cut[p.b]) for p in pairs]
for p in trimmed:
_test(p, mode, min_expected)
now = bool(_verdict(trimmed, alpha))
@@ -388,4 +415,4 @@ def _sensitivity(pairs: list[Pair], mode: str, alpha: Fraction, min_expected: Fr
return (value > mean, now) if now != differs else None
__all__ = ["TESTS", "differ", "refusal_basis", "used_settings"]
__all__ = ["REST_KEY", "TESTS", "differ", "refusal_basis", "untested_text", "used_settings"]
+9 -2
View File
@@ -32,8 +32,8 @@ from app.analysis.contract import OP_LABELS
from app.analysis.numbers import fixed, to_fraction
#: 附带说明进判定语的先后;不在这里的(measure_unlabeled、headline_basis、partial_periods)只在证据面板
PHRASE_ORDER = ("comparability", "simpson", "offset", "outlier_sensitive", "small_sample", "record_high", "record_low",
"multiple_comparisons", "weekday_effect", "small_denominator", "null_values")
PHRASE_ORDER = ("comparability", "untested", "simpson", "offset", "outlier_sensitive", "small_sample", "record_high",
"record_low", "multiple_comparisons", "weekday_effect", "small_denominator", "null_values")
#: 判定语里最多接几条附带说明
PHRASE_CAVEATS = 2
#: 消毒后的叫法最多几个字
@@ -343,6 +343,12 @@ def multiple_text(pairs: int) -> str:
return f"比了 {pairs} 对,已按次数校正"
def untested_pairs_text(count: int, reasons: Sequence[str]) -> str:
"""差异检验里拿不到 p 值的那几对(不算进比了几对):另有 1 对算不了(华北的样本量为 0)。"""
shown = list(dict.fromkeys(reasons))
return f"另有 {count} 对算不了({','.join(shown[:3])}{'等' if len(shown) > 3 else ''})"
def outlier_text(largest: bool, now_differs: bool) -> str:
which = "最大" if largest else "最小"
return f"去掉{which}的一笔后,差异{'变得站得住' if now_differs else '不再站得住'}"
@@ -567,6 +573,7 @@ __all__ = [
"decompose_flat", "decompose_unclosed", "differ_pair", "differ_phrase", "dims_text", "factor_both",
"factor_driver", "key_text", "multiple_text", "names", "no_evidence_phrase", "null_periods_text",
"null_segments_text", "number", "offset_text", "outlier_text", "partial_periods_text", "pct", "pick_label",
"untested_pairs_text",
"points", "ratio_both",
"ratio_change", "ratio_mix", "ratio_rate", "ratio_value", "record_text", "refused", "say", "share_words",
"simpson_text", "small_denominator_text", "small_sample_mean_text", "small_sample_rate_text", "threshold_pct",
+40 -2
View File
@@ -136,6 +136,44 @@ async def test_two_periods_of_the_same_group():
async def test_a_group_with_one_value_cannot_be_tested():
s = snap("Q1", "SELECT channel, amount FROM orders", ["channel", "amount"], [["线上", 5], ["线下", 4], ["线下", 6]])
result = await analyze("differ", "rows", snaps=[s], query="Q1", by="channel", value="amount")
assert result["verdict"]["code"] == "no_evidence" and result["values"]["p"] is None
assert result["table"]["rows"][0][-1] == "样本不够,算不了"
assert result["status"] == "refused" and result["reason_code"] == "untestable"
assert result["reason"] == "线上只有 1 个样本,算不了差异"
assert result["phrase"] == "差异检验没有做成:线上只有 1 个样本,算不了差异"
async def test_no_pair_with_a_p_value_is_refused_not_said_to_be_noise():
sql = "SELECT region, AVG(amount) AS m, 0 AS sd, COUNT(*) AS n FROM orders GROUP BY 1"
means = snap("Q1", sql, ["region", "m", "sd", "n"], [["华东", 120, 0, 1], ["华南", 80, 0, 1]])
mean = await analyze("differ", "mean", snaps=[means], query="Q1", by="region", mean="m", sd="sd", n="n",
label="客单价")
assert mean["reason_code"] == "untestable" and mean["reason"] == "华东只有 1 个样本、华南只有 1 个样本,算不了差异"
assert mean["verdict"] is None and mean["settings"] == {"differ_alpha": 0.05, "min_n": 30}
rows = snap("Q2", "SELECT region, amount FROM orders", ["region", "amount"], [["华东", 120], ["华南", 80]])
one = await analyze("differ", "rows", snaps=[rows], query="Q2", by="region", value="amount")
assert one["reason"] == "华东只有 1 个样本、华南只有 1 个样本,算不了差异"
zero = await rate([["华东", 0, 0], ["华南", 3, 10]])
assert zero["reason_code"] == "untestable" and zero["reason"] == "华东的样本量为 0,算不了差异"
async def test_pairs_without_a_p_value_are_named_and_not_counted():
result = await rate([["西南", 80, 1000], ["华东", 40, 1000], ["华北", 0, 0]])
rows = {r[0]: r for r in result["table"]["rows"]}
assert rows["华北"][-1] == "华北的样本量为 0,算不了" and rows["华北"][10] is None
note = next(c for c in result["caveats"] if c["code"] == "untested")
assert note["text"] == "另有 1 对算不了(华北的样本量为 0)"
# 华北一单都没有:西南的其余合计就是华东,两对都站得住;华北那一对不算进「比了几对」
assert result["phrase"] == ("比了 2 对,2 对差异站得住:西南的退款率(8.0%)高于其余各地区(4.0%);华东的退款率(4.0%)"
"低于其余各地区(8.0%);另有 1 对算不了(华北的样本量为 0)")
assert {"code": "multiple_comparisons", "text": "比了 2 对,已按次数校正"} in result["caveats"]
async def test_the_outlier_is_looked_for_in_the_real_groups_not_in_the_pooled_rest():
# 三组各和其余合计比:最远的那一笔在第二组(华南)。只在真实的各组里找、由各组重建「其余」合计,去掉它以后华南高出来就站得住
base = [10, 11, 9, 10, 12, 8, 10, 11, 10, 9, 11, 10]
rows = ([["华东", v] for v in base] + [["华南", v + 2] for v in base[:-1]] + [["华南", 100]]
+ [["华北", v] for v in base])
s = snap("Q1", "SELECT region, amount FROM orders", ["region", "amount"], rows)
result = await analyze("differ", "rows", snaps=[s], query="Q1", by="region", value="amount", label="客单价")
assert result["verdict"]["code"] == "no_evidence"
assert {"code": "outlier_sensitive", "text": "去掉最大的一笔后,差异变得站得住"} in result["caveats"]
assert math.isfinite(float(result["values"]["diff"]))