diff --git a/backend/app/analysis/__init__.py b/backend/app/analysis/__init__.py index 314806da..ecdb0c02 100644 --- a/backend/app/analysis/__init__.py +++ b/backend/app/analysis/__init__.py @@ -5,8 +5,11 @@ - run：按请求算一份分析结论。resolve 按调用里写的名字（Q3、来源节点 id）取回输入快照；单元格写成「名字.r行.列」 （Q1.r0.gmv），由这里拆开再经 resolve 取快照。parents 按名字（A3、分析卡里的项 id）取回上一层拆解，下钻用。 调用写错抛 contract.CallError；算不了返回 status refused 的结论。唯一的异步来自 context["probe"]（现查表的日期范围）。 + refuse_data_errors 为真时（分析卡：调用是模板定死的，这期的数据里没有要用的那一项不是写错了），随数据变化的调用错误 + （DATA_ERRORS）也记成算不了的结论：上一层没做成 parent_refused，其余 input_unavailable，原因照实写是哪一项、哪个值； + 结构性的（列不存在、遮罩列、比率当可加、重复行、参数写错）照旧抛。agent 工具不传，调用错误交回模型改调用。 - recompute：按结论里记下的 params、settings、probes 和输入快照（按 sha256 取回）重算一遍，不现查；和封存的用 - contract.same 比。 + contract.same 比。一律按 refuse_data_errors 为真复现：分析卡记下的这类结论复算也成立。 契约（名字、形状、缺省阈值）在 contract.py。各算子在 decompose.py（变化拆解）、series.py（波动基线、趋势）、differ.py （差异检验）、comparable.py（可比性检查）；统计在 stats.py，判定语和数的写法在 phrases.py，现查在 probe.py。 @@ -32,6 +35,14 @@ Resolver = Callable[[str], InputSnapshot | None] #: 按名字取回上一层的分析结论（下钻）；没有返回 None ParentResolver = Callable[[str], AnalysisResult | None] +#: 随数据变化的调用错误（CallError.code）：调用本身没写错，是这期的数据里没有要用的那一项。分析卡里（refuse_data_errors） +#: 记成算不了的结论，不让卡失败：上一层没做成（parent_refused）；下钻的段这期没有（parent_key_missing）；期次这期没有 +#: （period_not_found：变化拆解、波动基线的本期、差异检验按期比、可比性检查）；按位置取的总数那一格越界或为空 +#: （row_out_of_range、headline_empty）；要比的组这期没有（unknown_group、too_few_groups）；要用的那一格这期为空 +#: （empty_cell：乘积拆解的乘积、因子，差异检验的计数） +DATA_ERRORS = frozenset({"parent_refused", "parent_key_missing", "period_not_found", "row_out_of_range", + "headline_empty", "unknown_group", "too_few_groups", "empty_cell"}) + def _op_mode(request: Mapping[str, Any]) -> tuple[str, str | None]: op = request.get("op") @@ -69,9 +80,44 @@ def _env(op: str, mode: str | None, settings: Mapping[str, Any], inputs: Any, fa parents=parents, hypothesis=hypothesis if isinstance(hypothesis, str) else None) +def _data_refusal(op: str, mode: str | None, raw: Mapping[str, Any], env: Any, error: CallError) -> AnalysisResult: + """随数据变化的调用错误记成的算不了的结论。params 记调用原样（label 照样消毒；调用还没走完、规范化不完整，复算拿它重走一遍 + 走到同一处），取过的快照都记进输入（Inputs.settle），阈值和判定标准同这个算子自己算不了时那一份。""" + from app.analysis import decompose, differ, series + from app.analysis import phrases as ph + from app.analysis.common import build + + env.inputs.settle() + params = dict(raw) + if params.get("label") is not None: + params["label"] = ph.clean_label(params["label"]) or None + if op == "decompose": + used, criterion = decompose.refusal_basis(mode, raw, env.settings) # type: ignore[arg-type] + elif op in ("baseline", "trend"): + used, criterion = series.refusal_basis(op, env.settings) + elif op == "differ": + used, criterion = differ.refusal_basis(mode, env.settings) + else: + used, criterion = [], ph.CRITERION_COMPARABLE + code = "parent_refused" if error.code == "parent_refused" else "input_unavailable" + reason = error.reason or error.message + return build(env, params, used, status="refused", reason_code=code, reason=reason, criterion=criterion, + phrase=ph.refused(op, reason)) + + +def _compute_or_refuse(op: str, mode: str | None, raw: Mapping[str, Any], env: Any, refuse_data_errors: bool + ) -> AnalysisResult: + try: + return _compute(op, mode, raw, env) + except CallError as e: + if refuse_data_errors and e.code in DATA_ERRORS: + return _data_refusal(op, mode, raw, env, e) + raise + + async def run(request: AnalysisRequest, resolve: Resolver, *, settings: Mapping[str, Any], context: Context, - parents: ParentResolver | None = None) -> AnalysisResult: - """算一份分析结论。settings 是 contract.effective_settings 算好的阈值。""" + parents: ParentResolver | None = None, refuse_data_errors: bool = False) -> AnalysisResult: + """算一份分析结论。settings 是 contract.effective_settings 算好的阈值。refuse_data_errors 见文件头（分析卡传真）。""" from app.analysis.common import DERIVED_PARAMS from app.analysis.facts import Facts from app.analysis.inputs import Inputs @@ -97,7 +143,7 @@ async def run(request: AnalysisRequest, resolve: Resolver, *, settings: Mapping[ book = Book(known) env = _env(op, mode, full, Inputs(resolve, dialect), Facts(catalogs=context.get("catalogs")), book, today, parents, request.get("hypothesis")) - result = _compute(op, mode, raw, env) + result = _compute_or_refuse(op, mode, raw, env, refuse_data_errors) missing = book.missing if not missing or rounds == MAX_ROUNDS: return result @@ -109,7 +155,8 @@ def recompute(result: AnalysisResult, load: Callable[[str], InputSnapshot | None load_result: Callable[[str], AnalysisResult | None] | None = None) -> AnalysisResult: """复算：load 按 sha256 取回输入快照，load_result 按 sha256 取回 parent 那份结论。用结论里记下的 params（含运行当天、 目录事实）、settings、probes，不现查、不读目录。取不回输入时抛 CallError（code input_missing）；方法升版后调用不再成立 - 时同样抛 CallError（照 run 的说法）。和封存的那份用 contract.same 比。""" + 时同样抛 CallError（照 run 的说法）。随数据变化的调用错误一律记成算不了（同 run 的 refuse_data_errors 为真），分析卡记下 + 的这类结论复算也成立。和封存的那份用 contract.same 比。""" from app.analysis.common import DERIVED_PARAMS from app.analysis.facts import Facts from app.analysis.inputs import _CELL, Inputs @@ -150,7 +197,7 @@ def recompute(result: AnalysisResult, load: Callable[[str], InputSnapshot | None env = _env(op, mode, settings, Inputs(resolve, dialect), Facts(recorded=params.get("catalog") or {}), Book(result.get("probes") or []), today, parents, result.get("hypothesis")) raw = {k: v for k, v in params.items() if k not in DERIVED_PARAMS} - return _compute(op, mode, raw, env) + return _compute_or_refuse(op, mode, raw, env, True) def exhausted(settings: Mapping[str, Any]) -> CallError: @@ -174,4 +221,5 @@ def comparable_reasons(result: Mapping[str, Any]) -> list[str]: return reasons(result) -__all__ = ["ParentResolver", "Resolver", "comparable_reasons", "exhausted", "make_prober", "recompute", "run"] +__all__ = ["DATA_ERRORS", "ParentResolver", "Resolver", "comparable_reasons", "exhausted", "make_prober", "recompute", + "run"] diff --git a/backend/app/analysis/comparable.py b/backend/app/analysis/comparable.py index 3990222e..76041bb6 100644 --- a/backend/app/analysis/comparable.py +++ b/backend/app/analysis/comparable.py @@ -497,8 +497,9 @@ def comparable(raw: Mapping[str, Any], env: Any) -> Any: params: dict[str, Any] = {} two = raw.get("base_query") is not None or raw.get("target_query") is not None if two: - base = env.inputs.snap(need(raw, "base_query", "基期的查询编号"), "基期的查询编号") - target = env.inputs.snap(need(raw, "target_query", "本期的查询编号"), "本期的查询编号") + base = env.inputs.snap(need(raw, "base_query", "基期的查询编号"), "基期的查询编号", role="base") + target = env.inputs.snap(need(raw, "target_query", "本期的查询编号"), "本期的查询编号", + role="target") if base is target: from app.analysis.decompose import SAME_QUERY @@ -529,7 +530,8 @@ def comparable(raw: Mapping[str, Any], env: Any) -> Any: for side, what in (("base", "基期"), ("target", "本期")): if not rows[side]: raise CallError("period_not_found", f"期次列「{period}」里没有「{params[side]}」：{what}要写 {target.ref} " - "里有的期次值") + "里有的期次值", + reason=f"期次列「{period}」里没有{what}「{ph.key_text([params[side]])}」") env.inputs.use(target, "data", [period, *([measure] if measure else [])], rows["base"] + rows["target"]) parsed = pd.parse(values, grain, weeks=pd.week_scheme(target.sql, period, target.dialect)) if parsed is None: diff --git a/backend/app/analysis/contract.py b/backend/app/analysis/contract.py index eb038afd..965e1bbb 100644 --- a/backend/app/analysis/contract.py +++ b/backend/app/analysis/contract.py @@ -83,10 +83,12 @@ VERDICTS: dict[str, tuple[str, ...]] = { "differ": ("differs", "no_evidence"), "comparable": ("comparable", "conditional", "incomparable"), } -#: 算不了的原因（细案 01 §8.2）。算不了也记成结论：写作者可以照实写「历史只有 5 周，判断不了是否异常」 +#: 算不了的原因（细案 01 §8.2）。算不了也记成结论：写作者可以照实写「历史只有 5 周，判断不了是否异常」。最后两种是分析卡 +#: 里随数据变化的调用错误记成的（analysis.run 的 refuse_data_errors，见 app/analysis 的 DATA_ERRORS）：parent_refused 上一层 +#: 没做成；input_unavailable 这次的数据里没有要用的那一项（下钻的段、期次、总数那一格、要比的组……，reason 写是哪一项、哪个值） REASONS: tuple[str, ...] = ( "truncated", "too_few_segments", "history_short", "partial_target", "target_gap", "series_gap", - "period_unreadable", "zero_base", "non_positive", "incomparable_sides", + "period_unreadable", "zero_base", "non_positive", "incomparable_sides", "parent_refused", "input_unavailable", ) #: 附带说明的种类（caveats[].code） CAVEATS: tuple[str, ...] = ( @@ -305,12 +307,16 @@ class AnalysisResult(TypedDict, total=False): class CallError(Exception): """调用错误（细案 01 §8.1）：列不存在、列被遮罩、角色和目录度量类型冲突、同一键同一期多行、期次不在列里、乘积 不闭合、parent 不对、下钻超深、这个 Agent 这次执行的分析次数用完（max_calls 按每个 Agent 每次执行计）。不记结论，message - 交回 agent（一种说法，告诉它怎么改）。""" + 交回 agent（一种说法，告诉它怎么改）。 - def __init__(self, code: str, message: str) -> None: + 随数据变化的那几种（app/analysis 的 DATA_ERRORS：这期没有要用的期次、段、组，上一层没做成……）另带 reason：照实说是哪一项、 + 哪个值，不带改法。分析卡（模板定死了调用）按它记成算不了的结论，见 analysis.run 的 refuse_data_errors。""" + + def __init__(self, code: str, message: str, *, reason: str | None = None) -> None: super().__init__(message) self.code = code self.message = message + self.reason = reason #: 现查一列的实际范围：(数据源名, 表, 列) → Probe；查不了返回 None（算子写「核对不了」，不判不可比） diff --git a/backend/app/analysis/decompose.py b/backend/app/analysis/decompose.py index 0079d2a1..1d42bb88 100644 --- a/backend/app/analysis/decompose.py +++ b/backend/app/analysis/decompose.py @@ -41,6 +41,22 @@ SAME_QUERY = "基期和本期是同一份查询：两期各给一份查询，或 #: 乘积闭合的最小相对误差 _FACTOR_REL = Fraction(1, 10 ** 6) _NOT_ADDITIVE = {"ratio": "比率", "identifier": "标识", "status": "状态", "attribute": "属性"} +#: 各模式的结论里记的阈值（下钻时另加 max_drill_depth） +USED: dict[str, tuple[str, ...]] = { + "additive": ("main_share", "concentration", "offset_ratio", "flat_pct", "table_top_k"), + "ratio": ("main_share", "flat_pct", "min_denominator", "table_top_k"), + "factor": ("main_share", "flat_pct"), +} + + +def refusal_basis(mode: str, raw: Mapping[str, Any], settings: Mapping[str, Any]) -> tuple[list[str], str]: + """还没算就算不了时（分析卡把随数据变化的调用错误记成算不了的结论）结论里记的阈值和判定标准：同这个模式自己算不了时 + 的那一份。""" + used = list(USED[mode]) + (["max_drill_depth"] if mode != "factor" and raw.get("parent") is not None else []) + if mode == "additive": + summed = raw.get("headline") is None and raw.get("parent") is None + return used, ph.criterion_additive(settings, top_k_max=None, summed=summed) + return used, ph.criterion_ratio(settings) if mode == "ratio" else ph.criterion_factor(settings) @dataclass @@ -71,8 +87,9 @@ def _sides(raw: Mapping[str, Any], env: Env, roles: Sequence[tuple[str, str]], * params: dict[str, Any] = {} two = raw.get("base_query") is not None or raw.get("target_query") is not None if two: - base = env.inputs.snap(need(raw, "base_query", "基期的查询编号"), "基期的查询编号") - target = env.inputs.snap(need(raw, "target_query", "本期的查询编号"), "本期的查询编号") + base = env.inputs.snap(need(raw, "base_query", "基期的查询编号"), "基期的查询编号", role="base") + target = env.inputs.snap(need(raw, "target_query", "本期的查询编号"), "本期的查询编号", + role="target") if base is target: raise CallError("same_query", SAME_QUERY) params.update(base_query=base.ref, target_query=target.ref) @@ -116,7 +133,8 @@ def _period_rows(snap: Snap, period: str, wanted: Any, what: str) -> list[int]: if not rows: values = list(dict.fromkeys(ph.key_text([row[i]]) for row in snap.rows if i < len(row))) raise CallError("period_not_found", f"期次列「{period}」里没有「{wanted}」：{what}要写 {snap.ref} 里有的期次值，" - f"比如 {'、'.join(values[:6])}") + f"比如 {'、'.join(values[:6])}", + reason=f"期次列「{period}」里没有{what}「{ph.key_text([wanted])}」") return rows @@ -181,8 +199,12 @@ def _parent(raw: Mapping[str, Any], env: Env, mode: str, dims: int) -> tuple[dic if found.get("op") != "decompose" or found.get("mode") != mode: raise CallError("parent_not_decompose", f"{name} 不是{'按维度' if mode == 'additive' else '比率'}的变化拆解，" "下钻要接在同一种拆解后面") + # 上一层先记进输入：它没做成、这期没有那一段时（分析卡记成算不了的结论），复算也要按内容哈希取回它、走到同一处 + env.inputs.parent(name, content_hash(canonical_json(found))) if found.get("status") != "ok" or not isinstance(found.get("table"), Mapping): - raise CallError("parent_refused", f"{name} 没有做成，不能在它下面下钻") + why = f"（{found['reason']}）" if isinstance(found.get("reason"), str) and found["reason"] else "" + raise CallError("parent_refused", f"{name} 没有做成，不能在它下面下钻", + reason=f"上一层「{name}」没有做成{why}，不能在它下面下钻") depth = int((found.get("params") or {}).get("depth") or 0) + 1 limit = int(env.settings["max_drill_depth"]) if depth > limit: @@ -196,8 +218,8 @@ def _parent(raw: Mapping[str, Any], env: Env, mode: str, dims: int) -> tuple[dic hit = next((row for row in table.get("rows") or [] if all(same_value(k, v) for k, v in zip(key, row[:dim_count]))), None) if hit is None: - raise CallError("parent_key_missing", f"{name} 里没有「{ph.key_text(key)}」这一段：写 {name} 结果表里的段") - env.inputs.parent(name, content_hash(canonical_json(found))) + raise CallError("parent_key_missing", f"{name} 里没有「{ph.key_text(key)}」这一段：写 {name} 结果表里的段", + reason=f"上一层「{name}」里没有「{ph.key_text(key)}」这一段") return {"of": name, "key": key}, dict(zip(columns, hit)), depth @@ -220,7 +242,8 @@ def _headline(raw: Mapping[str, Any], env: Env) -> tuple[dict[str, str] | None, for side, what in (("base", "基期总数"), ("target", "本期总数")): snap, row, col, value = env.inputs.cell(spec[side], f"headline_{side}", what) if value is None: - raise CallError("headline_empty", f"{what}那一格（{snap.ref}.r{row}.{col}）是空的：换一格有值的") + raise CallError("headline_empty", f"{what}那一格（{snap.ref}.r{row}.{col}）是空的：换一格有值的", + reason=f"{what}那一格（{snap.ref}.r{row}.{col}）是空的") out[side] = (value, snap.cell(row, col)) refs[side] = f"{snap.ref}.r{row}.{col}" return refs, out @@ -260,7 +283,7 @@ def _share_rank(segments: Sequence[Any], share: Any) -> list[Any]: def additive(raw: Mapping[str, Any], env: Env) -> Any: - used = ["main_share", "concentration", "offset_ratio", "flat_pct", "table_top_k"] + used = list(USED["additive"]) params, bs, ts, cols, by = _sides(raw, env, [("measure", "度量列")]) measure = cols["measure"] label = label_for(env, raw.get("label"), ts, measure) @@ -299,7 +322,8 @@ def additive(raw: Mapping[str, Any], env: Env) -> Any: criterion=ph.criterion_additive(env.settings, top_k_max=None, **criterion_args), caveats=caveats) segs = _segments(side_b, side_t, Fraction(0)) if len(segs) < 2: - return refuse(env, params, used, "too_few_segments", f"按{ph.dims_text(dims)}只有 {len(segs)} 段，至少要 2 段才拆得开", + return refuse(env, params, used, "too_few_segments", + ph.say("按", ph.dims_text(dims), f"只有 {len(segs)} 段，至少要 2 段才拆得开"), criterion=ph.criterion_additive(env.settings, top_k_max=None, **criterion_args), caveats=caveats) return _additive_result(env, params, used, segs, by, dims, label, heads, parent_row, outcome, caveats, summed, side_b, side_t) @@ -419,7 +443,7 @@ def _additive_result(env: Env, params: dict[str, Any], used: list[str], segs: li def ratio(raw: Mapping[str, Any], env: Env) -> Any: - used = ["main_share", "flat_pct", "min_denominator", "table_top_k"] + used = list(USED["ratio"]) params, bs, ts, cols, by = _sides(raw, env, [("numerator", "分子列"), ("denominator", "分母列")]) num, den = cols["numerator"], cols["denominator"] num_label, den_label = column_label(env, ts, num), column_label(env, ts, den) @@ -462,7 +486,8 @@ def ratio(raw: Mapping[str, Any], env: Env) -> Any: segs = _segments(side_b, side_t, (Fraction(0), Fraction(0))) live = [x for x in segs if x.base[1] != 0 or x.target[1] != 0] if len(live) < 2: - return refuse(env, params, used, "too_few_segments", f"按{ph.dims_text(dims)}只有 {len(live)} 段，至少要 2 段才拆得开", + return refuse(env, params, used, "too_few_segments", + ph.say("按", ph.dims_text(dims), f"只有 {len(live)} 段，至少要 2 段才拆得开"), criterion=criterion) for when, i in (("基期", 0), ("本期", 1)): if sum(((x.base, x.target)[i][1] for x in live), Fraction(0)) == 0: @@ -635,7 +660,7 @@ def lmdi(base: Sequence[float], target: Sequence[float], m_base: float, m_target def factor(raw: Mapping[str, Any], env: Env) -> Any: - used = ["main_share", "flat_pct"] + used = list(USED["factor"]) params, bs, ts, cols, _ = _sides(raw, env, [("product", "乘积列")], by_needed=False) product = cols["product"] wanted = as_list(raw.get("factors")) @@ -672,7 +697,8 @@ def factor(raw: Mapping[str, Any], env: Env) -> Any: for col, what in ((product, "乘积列"), *((c, "因子列") for c in factors)): value = snap.number(r, col, what) if value is None: - raise CallError("empty_cell", f"{snap.ref} 第 {r} 行的「{col}」是空的：乘积、因子都要有值") + raise CallError("empty_cell", f"{snap.ref} 第 {r} 行的「{col}」是空的：乘积、因子都要有值", + reason=f"{'基期' if when == 'base' else '本期'}的「{col}」那一格是空的（{snap.ref} 第 {r} 行）") row.append((value, snap.cell(r, col))) cells[when] = row if bs is ts: @@ -736,5 +762,5 @@ def factor(raw: Mapping[str, Any], env: Env) -> Any: phrase=ph.with_caveats(main, caveats), caveats=caveats) -__all__ = ["HEADLINE_SHAPE", "PARENT_SHAPE", "SAME_QUERY", "SIMPSON_SHARE", "additive", "factor", "lmdi", "ratio", - "ratio_effects"] +__all__ = ["HEADLINE_SHAPE", "PARENT_SHAPE", "SAME_QUERY", "SIMPSON_SHARE", "USED", "additive", "factor", "lmdi", + "ratio", "ratio_effects", "refusal_basis"] diff --git a/backend/app/analysis/differ.py b/backend/app/analysis/differ.py index e8f839af..12e6816b 100644 --- a/backend/app/analysis/differ.py +++ b/backend/app/analysis/differ.py @@ -36,6 +36,16 @@ from app.analysis.numbers import canon, to_fraction TESTS = {"z": "z 检验", "fisher": "Fisher 精确检验", "welch": "Welch t 检验"} +def used_settings(mode: str | None) -> list[str]: + """结论里记的阈值：比率看最小期望频数，均值、逐行看最小样本数。""" + return ["differ_alpha", "min_expected"] if mode == "rate" else ["differ_alpha", "min_n"] + + +def refusal_basis(mode: str | None, settings: Mapping[str, Any]) -> tuple[list[str], str]: + """还没算就算不了时结论里记的阈值和判定标准（同 decompose.refusal_basis）。""" + return used_settings(mode), ph.criterion_differ(settings, mode) # type: ignore[arg-type] + + @dataclass(eq=False) class Group: key: Any @@ -62,7 +72,10 @@ class Pair: def _int(value: Fraction | None, snap: Snap, r: int, column: str) -> int: - if value is None or value.denominator != 1 or value < 0: + if value is None: + raise CallError("empty_cell", f"{snap.ref} 第 {r} 行的「{column}」是空的：计数要有值（在 SQL 里把空值写成 0）", + reason=f"「{column}」那一格是空的（{snap.ref} 第 {r} 行）") + if value.denominator != 1 or value < 0: raise CallError("not_count", f"{snap.ref} 第 {r} 行的「{column}」不是不小于 0 的整数：比率比较的分子、分母要是计数") return int(value) @@ -135,7 +148,7 @@ def _verdict(pairs: Sequence[Pair], alpha: Fraction) -> list[Pair]: def differ(raw: Mapping[str, Any], env: Env) -> Any: mode = env.mode - used = ["differ_alpha", "min_expected"] if mode == "rate" else ["differ_alpha", "min_n"] + used = used_settings(mode) snap = env.inputs.snap(need(raw, "query", "查询编号"), "查询编号") params: dict[str, Any] = {"query": snap.ref} by_period = raw.get("period") is not None @@ -230,7 +243,8 @@ def _groups(snap: Snap, group_col: str, cols: Mapping[str, str], mode: str, by_p if by_period: for w, what in ((params["target"], "本期"), (params["base"], "基期")): if not any(same_value(w, g.key) for g in groups.values()): - raise CallError("period_not_found", f"期次列「{group_col}」里没有「{w}」：{what}要写 {snap.ref} 里有的期次值") + raise CallError("period_not_found", f"期次列「{group_col}」里没有「{w}」：{what}要写 {snap.ref} 里有的期次值", + reason=f"期次列「{group_col}」里没有{what}「{ph.key_text([w])}」") return groups @@ -238,7 +252,8 @@ def _find(groups: Mapping[Any, Group], value: Any, column: str) -> Group: found = next((g for g in groups.values() if same_value(value, g.key)), None) if found is None: shown = "、".join(g.name for g in list(groups.values())[:8]) - raise CallError("unknown_group", f"「{value}」不是「{column}」里的值：要比的组写 {shown} 里的") + raise CallError("unknown_group", f"「{value}」不是「{column}」里的值：要比的组写 {shown} 里的", + reason=f"分组列「{column}」里没有「{ph.key_text([value])}」这一组") return found @@ -261,7 +276,8 @@ def _pairs(raw: Mapping[str, Any], params: dict[str, Any], groups: dict[Any, Gro params["pairs"] = [[p.a.key, p.b.key] for p in out] return out if len(groups) < 2: - raise CallError("too_few_groups", f"「{group_col}」里只有 {len(groups)} 组，至少要两组才比得了") + raise CallError("too_few_groups", f"「{group_col}」里只有 {len(groups)} 组，至少要两组才比得了", + reason=f"分组列「{group_col}」里只有 {len(groups)} 组，至少要两组才比得了") if raw.get("reference") is not None: ref = _find(groups, raw.get("reference"), group_col) params["reference"] = ref.key @@ -372,4 +388,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"] +__all__ = ["TESTS", "differ", "refusal_basis", "used_settings"] diff --git a/backend/app/analysis/inputs.py b/backend/app/analysis/inputs.py index 88229117..819806f7 100644 --- a/backend/app/analysis/inputs.py +++ b/backend/app/analysis/inputs.py @@ -60,6 +60,7 @@ class Snap: dialect: str | None = None columns: list[str] = field(default_factory=list) rows: list[list[Any]] = field(default_factory=list) + named: list[str] = field(default_factory=list) # 调用里认过的列（按先后，Inputs.settle 记用法用） def __post_init__(self) -> None: self.columns = [str(c) for c in self.data.get("columns") or []] @@ -117,6 +118,8 @@ class Snap: column = hits[0] if column.lower() in self.masked: raise CallError("masked_column", f"「{column}」是遮罩列，不能当{role}：换一列，或者在 SQL 里不选它") + if column not in self.named: + self.named.append(column) return column def index(self, column: str) -> int: @@ -157,18 +160,20 @@ class Inputs: self._resolve = resolve self._dialect = dialect self._snaps: dict[str, Snap] = {} + self._roles: dict[str, str] = {} self.records: list[AnalysisInput] = [] - def snap(self, name: Any, role: str) -> Snap: - """按名字取一份快照。""" + def snap(self, name: Any, what: str, *, role: str = "data") -> Snap: + """按名字取一份快照。what 是调用错误里的叫法；role 是这份快照在结论里的角色（第一次取时的为准，settle 用）。""" if not isinstance(name, str) or not name.strip(): - raise CallError("missing_query", f"没写{role}：写本节点查过的查询编号（如 Q3）") + raise CallError("missing_query", f"没写{what}：写本节点查过的查询编号（如 Q3）") key = name.strip() if key not in self._snaps: found = self._resolve(key) if not isinstance(found, Mapping): - raise CallError("unknown_query", f"没有叫「{key}」的查询结果：{role}写本节点查过的查询编号（如 Q3）") + raise CallError("unknown_query", f"没有叫「{key}」的查询结果：{what}写本节点查过的查询编号（如 Q3）") self._snaps[key] = Snap(key, found, self._dialect(key, found) if self._dialect else found.get("dialect")) + self._roles[key] = role return self._snaps[key] def cell(self, ref: Any, role: str, label: str) -> tuple[Snap, int, str, Fraction | None]: @@ -176,10 +181,11 @@ class Inputs: found = _CELL.match(ref.strip()) if isinstance(ref, str) else None if found is None: raise CallError("bad_cell", f"{label}要写成单元格「编号.r行.列」（如 Q1.r0.gmv），现在是「{ref}」") - snap = self.snap(found.group("name"), label) + snap = self.snap(found.group("name"), label, role=role) row = int(found.group("row")) if row >= len(snap.rows): - raise CallError("row_out_of_range", f"{snap.ref} 没有第 {row} 行（共 {len(snap.rows)} 行，从 r0 数起）：改{label}") + raise CallError("row_out_of_range", f"{snap.ref} 没有第 {row} 行（共 {len(snap.rows)} 行，从 r0 数起）：改{label}", + reason=f"{label}那一格（{ref.strip()}）不在：{snap.ref} 这次只有 {len(snap.rows)} 行") column = snap.column(found.group("column"), label) value = snap.number(row, column, label) self.record({"ref": f"{snap.ref}.r{row}.{column}", "role": role, "row": row, "column": column, @@ -209,6 +215,18 @@ class Inputs: if entry not in self.records: self.records.append(entry) # type: ignore[arg-type] + def settle(self) -> None: + """调用在算出结果之前就算不了时（随数据变化的调用错误记成算不了的结论）：取过、还没记下用法的快照照样记一条（角色照 + 第一次取时的，列是调用里认过的那几列，行记 all），复算才按名字取得回它们、走到同一处。""" + recorded = set() + for item in self.records: + cell = _CELL.match(str(item.get("ref") or "")) if "row" in item else None + recorded.add(cell.group("name") if cell else item.get("ref")) + for key, snap in self._snaps.items(): + if key not in recorded: + self.record({"ref": key, "role": self._roles.get(key, "data"), "columns": list(snap.named), + "rows": "all"}, snap) + def as_list(value: Any) -> list[Any]: """参数里一个值或一组值：都规范成列表（None 是空列表）。""" diff --git a/backend/app/analysis/periods.py b/backend/app/analysis/periods.py index 5b8a4c9c..8082d248 100644 --- a/backend/app/analysis/periods.py +++ b/backend/app/analysis/periods.py @@ -257,7 +257,8 @@ def week_scheme(sql: str, column: str, dialect: str | None) -> str | None: return None refs = {(name_key(c.table), name_key(c.name)) for c in value.find_all(exp.Column) if not isinstance(c.this, exp.Star)} - texts = [_native_format(node, dialect) for node in value.find_all(exp.TimeToStr, exp.ToChar)] # type: ignore[arg-type] + texts = [_native_format(node, dialect) # type: ignore[arg-type] + for node in value.find_all(exp.TimeToStr, exp.ToChar)] if len(refs) != 1 or not texts or any(t is None for t in texts): return None marks = [m for t in texts for m in _format_marks(t, dialect)] # type: ignore[arg-type] diff --git a/backend/app/analysis/series.py b/backend/app/analysis/series.py index 0a05ac7d..840f4d56 100644 --- a/backend/app/analysis/series.py +++ b/backend/app/analysis/series.py @@ -40,6 +40,17 @@ from app.analysis.numbers import canon, to_fraction MAD_SCALE = Fraction(14826, 10000) #: lag 的上限（周数据同比是 52） _LAG_MAX = 260 +#: 结论里记的阈值 +USED: dict[str, tuple[str, ...]] = { + "baseline": ("k_warn", "k_alert", "min_history", "max_history"), + "trend": ("trend_min_points", "trend_alpha", "max_history"), +} + + +def refusal_basis(op: str, settings: Mapping[str, Any]) -> tuple[list[str], str]: + """还没算就算不了时结论里记的阈值和判定标准（同 decompose.refusal_basis）。""" + criterion = ph.criterion_baseline if op == "baseline" else ph.criterion_trend + return list(USED[op]), criterion(settings, None, "期") @dataclass @@ -147,7 +158,7 @@ def _completeness(env: Env, series: Series, item: pd.Period) -> tuple[str, str | def baseline(raw: Mapping[str, Any], env: Env) -> Any: - used = ["k_warn", "k_alert", "min_history", "max_history"] + used = list(USED["baseline"]) loaded = _load(raw, env, used) if not isinstance(loaded, Series): return loaded @@ -168,7 +179,8 @@ def baseline(raw: Mapping[str, Any], env: Env) -> Any: found = next((p for p in series.points if same_value(raw["target"], p[0].raw)), None) if found is None: raise CallError("period_not_found", f"期次列「{series.period}」里没有「{raw['target']}」（或那一期为空）：" - "本期要写序列里有值的一期，不写就是最后一期") + "本期要写序列里有值的一期，不写就是最后一期", + reason=f"期次列「{series.period}」里没有本期「{ph.key_text([raw['target']])}」（或那一期为空）") else: found = series.points[-1] target, x_level = found @@ -277,7 +289,7 @@ def run_length(values: list[Fraction]) -> tuple[int, str | None]: def trend(raw: Mapping[str, Any], env: Env) -> Any: - used = ["trend_min_points", "trend_alpha", "max_history"] + used = list(USED["trend"]) loaded = _load(raw, env, used) if not isinstance(loaded, Series): return loaded @@ -357,4 +369,4 @@ def _tail_gap(series: Series, ordinal: int) -> str: return "再往前缺期" -__all__ = ["MAD_SCALE", "baseline", "run_length", "trend", "unreadable_text"] +__all__ = ["MAD_SCALE", "USED", "baseline", "refusal_basis", "run_length", "trend", "unreadable_text"] diff --git a/backend/app/engine/nodes/analysis.py b/backend/app/engine/nodes/analysis.py index 382104a7..80501458 100644 --- a/backend/app/engine/nodes/analysis.py +++ b/backend/app/engine/nodes/analysis.py @@ -13,7 +13,10 @@ driver……），下游条件分支能按它走（周报「超出常规波动 （contract.ledger_entry，带 item），报告的写作目录编成 A{n}；每项另发一条 analysis.end（运行详情、运行对比按项对齐）。 **一项算不了不让卡失败。** 算不了（历史不够、本期不完整、截断……）就是这一项的结论（status refused），照实记、发一条 -analysis_refused 提醒，接着做下一项。调用写错（列不存在、比率当可加……）是模板的毛病：整张卡失败，说清是哪一项、怎么改。 +analysis_refused 提醒，接着做下一项。模板定死了调用，这期的数据里没有要用的那一项（上一层没做成、下钻的段这期没有、期次 +这期没有、总数那一格越界或为空、要比的组这期没有，analysis.DATA_ERRORS）不是写错了，同样记成算不了的一项 +（run 的 refuse_data_errors）。结构性的调用错误（列不存在、遮罩列、比率当可加、重复行、参数写错……）是模板的毛病：整张卡 +失败，说清是哪一项、怎么改。 分析的结果不作口径卡的输入（D16）：口径卡读 nodes.<卡>.results 的，按追不到原始数据处理（受管级发布时另有提示）。 """ @@ -74,8 +77,9 @@ async def run_analysis(state: GraphState, ctx: NodeContext) -> dict[str, Any]: who = f"分析项「{item_id}」（{ANALYSIS_OP_LABEL[op]}）" request = _request(item, state, ctx, who) try: - # 下钻（parent）只认这张卡里排在前面、已经做完的项：done 里此刻只有它们 - result = await analysis_ops.run(request, inputs.get, settings=settings, context=context, parents=done.get) + # 下钻（parent）只认这张卡里排在前面、已经做完的项：done 里此刻只有它们（算不了的也在，下钻那一项照实记成算不了） + result = await analysis_ops.run(request, inputs.get, settings=settings, context=context, parents=done.get, + refuse_data_errors=True) except CallError as e: raise NodeError(ctx.node.id, f"{who}：{e.message}。请在分析卡中改正这一项") from None except Exception as e: # noqa: BLE001 - 算子本身出错：说清是哪一项，原因照实 diff --git a/backend/tests/analysis_samples.py b/backend/tests/analysis_samples.py index 01fae6b0..e6a1b11d 100644 --- a/backend/tests/analysis_samples.py +++ b/backend/tests/analysis_samples.py @@ -52,8 +52,10 @@ CATALOGS = {"shop": {"orders": ORDERS}} async def analyze(op: str, mode: str | None = None, *, snaps: Sequence[Mapping[str, Any]], today: date = TODAY, catalogs: Mapping[str, Any] | None = None, probe: Any = None, facts: Any = None, parents: Mapping[str, Mapping[str, Any]] | None = None, settings: Mapping[str, Any] | None = None, - hypothesis: str | None = None, check: bool = True, **params: Any) -> dict[str, Any]: - """跑一次 analysis.run；check 时再复算一遍，核对和封存的逐字段相同（hypothesis 除外）。""" + hypothesis: str | None = None, check: bool = True, refuse_data_errors: bool = False, + **params: Any) -> dict[str, Any]: + """跑一次 analysis.run；check 时再复算一遍，核对和封存的逐字段相同（hypothesis 除外）。refuse_data_errors 同 run（分析卡 + 传真）；复算一律按真。""" by_ref = {s["ref"]: s for s in snaps} request: dict[str, Any] = {"op": op, "mode": mode, "params": params} if hypothesis is not None: @@ -64,7 +66,7 @@ async def analyze(op: str, mode: str | None = None, *, snaps: Sequence[Mapping[s if facts is not None: context["facts"] = facts result = await analysis.run(request, by_ref.get, settings=effective_settings(settings), context=context, - parents=(parents or {}).get) + parents=(parents or {}).get, refuse_data_errors=refuse_data_errors) if check: again = recomputed(result, snaps, parents) assert same(result, again), _diff(result, again) diff --git a/backend/tests/fixtures/analysis_ops.py b/backend/tests/fixtures/analysis_ops.py index c9371b10..a785ed4c 100644 --- a/backend/tests/fixtures/analysis_ops.py +++ b/backend/tests/fixtures/analysis_ops.py @@ -30,7 +30,7 @@ class FakeOps: monkeypatch.setattr(analysis_ops, "run", self.run) async def run(self, request: dict[str, Any], resolve: Any, *, settings: Any, context: Any, - parents: Any = None) -> dict[str, Any]: + parents: Any = None, refuse_data_errors: bool = False) -> dict[str, Any]: params = request.get("params") or {} names = [params[k] for k in ("query", "base_query", "target_query") if params.get(k)] snapshots = {name: resolve(name) for name in names} @@ -38,7 +38,7 @@ class FakeOps: if isinstance(params.get("parent"), dict) and parents is not None: parent = parents(params["parent"]["of"]) self.calls.append({"request": request, "snapshots": snapshots, "settings": dict(settings), "context": context, - "parent": parent, "parents": parents, + "parent": parent, "parents": parents, "refuse_data_errors": refuse_data_errors, "visible": {n: parents is not None and parents(n) is not None for n in self.peek}}) error = self.errors.pop(0) if self.errors else None if error is not None: diff --git a/backend/tests/test_analysis_card_data.py b/backend/tests/test_analysis_card_data.py new file mode 100644 index 00000000..755df707 --- /dev/null +++ b/backend/tests/test_analysis_card_data.py @@ -0,0 +1,70 @@ +"""分析卡里随数据变化的调用错误（复核 R6）：模板定死了调用，这期的数据里没有要用的那一项（上一层没做成、下钻的段这期没有……） +记成算不了的一项，不让卡失败；结构性的调用错误照旧让卡失败并说清哪一项。接真算子跑（tests/test_analysis_e2e 的虚构零售库 +shop），每份结论按封存的工件复算、逐字段相同。""" +from __future__ import annotations + +from app.core import artifact_store +from app.engine.runner import run_manager +from tests.test_analysis_e2e import ( # noqa: F401 - engine_up、shop 是自动用的夹具 + BY_REGION, EAST, TOOL, TWO_WEEKS, WEEK, chain, engine_up, events, node, recomputes, shop, wait, +) + +ONLY_EAST = (f"SELECT region, {WEEK} AS week, SUM(amount) AS gmv FROM orders " + "WHERE region = '华东' AND day >= '2026-05-11' GROUP BY region, week ORDER BY region, week") +WHERE = {"id": "where", "op": "decompose", "mode": "additive", "from": "by_region", + "params": {"by": ["region"], **TWO_WEEKS}} +DRILL = {"id": "east", "op": "decompose", "mode": "additive", "from": "east_q", + "params": {"by": ["channel"], **TWO_WEEKS, "parent": {"of": "where", "key": ["华东"]}}} + + +def card_graph(region_sql, items): + return chain(node("start", "input"), node("by_region", "tool", tool=TOOL, args={"sql": region_sql}), + node("east_q", "tool", tool=TOOL, args={"sql": EAST}), + node("card", "analysis", items=items), + node("out", "output", fields=[{"name": "r", "value": "{{ nodes.card.results.east.phrase }}"}])) + + +async def _card(region_sql, items): + run = await run_manager.start(graph=card_graph(region_sql, items), input_payload={}) + row = await wait(run.id) + return row + + +async def test_a_drill_under_a_refused_parent_is_refused_and_the_card_goes_on(): + row = await _card(ONLY_EAST, [WHERE, DRILL]) + assert row.status == "succeeded", row.error + [finished] = await events(row.id, "node.finished", "card") + results = artifact_store.load(finished["artifact"])["results"] + assert results["where"]["status"] == "refused" and results["east"]["status"] == "refused" + east = artifact_store.load(results["east"]["artifact"]) + assert east["reason_code"] == "parent_refused" + assert east["reason"] == "上一层「where」没有做成（按 region 只有 1 段，至少要 2 段才拆得开），不能在它下面下钻" + assert row.output["r"] == "变化拆解没有做成：" + east["reason"] + ends = {e["item"]: e for e in await events(row.id, "analysis.end", "card")} + assert ends["east"]["reason_code"] == "parent_refused" and ends["east"]["status"] == "refused" + warned = [e for e in await events(row.id, "log", "card") if e.get("code") == "analysis_refused"] + assert [e["item"] for e in warned] == ["where", "east"] + assert [(e["item"], e["status"]) for e in finished["evidence"]] == [("where", "refused"), ("east", "refused")] + recomputes(artifact_store.load(results["where"]["artifact"])) + recomputes(east) + + +async def test_a_segment_missing_this_time_is_refused_and_the_card_goes_on(): + drill = {**DRILL, "params": {**DRILL["params"], "parent": {"of": "where", "key": ["东北"]}}} + row = await _card(BY_REGION, [WHERE, drill]) + assert row.status == "succeeded", row.error + [finished] = await events(row.id, "node.finished", "card") + results = artifact_store.load(finished["artifact"])["results"] + assert results["where"]["status"] == "ok" and results["east"]["status"] == "refused" + east = artifact_store.load(results["east"]["artifact"]) + assert east["reason_code"] == "input_unavailable" and east["reason"] == "上一层「where」里没有「东北」这一段" + assert [i["role"] for i in east["inputs"]] == ["parent", "data"] + recomputes(east) + + +async def test_a_structural_call_error_still_fails_the_card_and_names_the_item(): + wrong = {**WHERE, "params": {**WHERE["params"], "by": ["channel"]}} + row = await _card(BY_REGION, [wrong]) + assert row.status == "failed" + assert "分析项「where」（变化拆解）" in row.error and "没有「channel」这一列" in row.error + assert "请在分析卡中改正这一项" in row.error diff --git a/backend/tests/test_analysis_run.py b/backend/tests/test_analysis_run.py index 836dba63..36c09a44 100644 --- a/backend/tests/test_analysis_run.py +++ b/backend/tests/test_analysis_run.py @@ -153,3 +153,97 @@ async def test_ledger_entries_name_the_real_input_snapshots(): def test_the_call_limit_error_is_spoken_with_the_limit(): error = analysis.exhausted({"max_calls": 5}) assert error.code == "max_calls" and error.message == "这个 Agent 这次执行最多做 5 次分析，已经用完：用已有的分析结论写，不再新做分析" + + +# -------------------------------------------------------------------------- +# 随数据变化的调用错误（分析卡传 refuse_data_errors）：记成算不了的结论，复算也成立 +# -------------------------------------------------------------------------- + +ONE_REGION = snap("Q6", WEEKLY_SQL, ["region", "week", "gmv"], [["华东", "2026-W41", 820], ["华东", "2026-W42", 700]]) +TOTALS = snap("Q7", "SELECT week, SUM(amount) AS gmv FROM orders GROUP BY week", ["week", "gmv"], + [["2026-W42", 1926], ["2026-W41", None]]) +FACTORS = snap("Q8", "SELECT week, SUM(amount) AS gmv, COUNT(*) AS orders, AVG(amount) AS aov FROM orders " + "GROUP BY week", + ["week", "gmv", "orders", "aov"], [["2026-W41", 2120, 20, 106], ["2026-W42", None, 0, None]]) +ONE_GROUP = snap("Q9", "SELECT region, SUM(refund_flag) AS refunds, COUNT(*) AS orders FROM orders GROUP BY region", + ["region", "refunds", "orders"], [["西南", 80, 1000]]) +WEEK_RATES = snap("Q10", "SELECT week, SUM(refund_flag) AS refunds, COUNT(*) AS orders FROM orders GROUP BY week", + ["week", "refunds", "orders"], [["2026-W41", 30, 1000], ["2026-W42", None, 1000]]) +REFUSALS = [WEEKLY, SERIES, RATES, ONE_REGION, TOTALS, FACTORS, ONE_GROUP, WEEK_RATES] +BY_REGION = {"query": "Q2", "by": ["region"], "period": "week", "base": "2026-W41", "target": "2026-W42", + "measure": "gmv"} + + +async def _refused(op, mode=None, *, parents=None, **params): + """refuse_data_errors 为真跑两次：逐字节相同、形状照契约、复算逐字段相同（analyze 里按真复算）；不传时照旧是调用错误。""" + one = await analyze(op, mode, snaps=REFUSALS, parents=parents, refuse_data_errors=True, **params) + two = await analyze(op, mode, snaps=REFUSALS, parents=parents, refuse_data_errors=True, **params) + assert json.dumps(one, sort_keys=True, ensure_ascii=False) == json.dumps(two, sort_keys=True, ensure_ascii=False) + _assert_shape(one) + assert one["status"] == "refused" and one["phrase"] == f"{c.OP_LABELS[op]}没有做成：{one['reason']}" + error = await call_error(op, mode, snaps=REFUSALS, parents=parents, **params) + assert error.code in analysis.DATA_ERRORS and error.reason == one["reason"] + return one + + +def test_the_data_errors_are_exported(): + assert analysis.DATA_ERRORS == {"parent_refused", "parent_key_missing", "period_not_found", "row_out_of_range", + "headline_empty", "unknown_group", "too_few_groups", "empty_cell"} + assert {"parent_refused", "input_unavailable"} <= set(c.REASONS) + + +async def test_a_drill_under_a_refused_parent_is_itself_refused(): + parent = await analyze("decompose", "additive", snaps=REFUSALS, **{**BY_REGION, "query": "Q6"}) + assert parent["reason_code"] == "too_few_segments" + result = await _refused("decompose", "additive", parents={"where": parent}, + **{**BY_REGION, "by": ["region"], "parent": {"of": "where", "key": ["华东"]}}) + assert result["reason_code"] == "parent_refused" + assert result["reason"] == "上一层「where」没有做成（按地区只有 1 段，至少要 2 段才拆得开），不能在它下面下钻" + # 上一层、取过的快照都记进输入（复算按它们走到同一处）；参数照调用原样，阈值同这个模式自己算不了时 + assert [i["role"] for i in result["inputs"]] == ["parent", "data"] + assert result["inputs"][1] == {"ref": "Q2", "artifact": WEEKLY["artifact"], "role": "data", + "columns": ["region", "gmv", "week"], "rows": "all"} + assert result["params"]["parent"] == {"of": "where", "key": ["华东"]} + assert result["settings"] == {**{k: c.DEFAULTS[k] for k in ("main_share", "concentration", "offset_ratio", + "flat_pct", "table_top_k")}, "max_drill_depth": 3} + assert result["table"] is None and result["verdict"] is None + + +async def test_segments_periods_cells_and_groups_missing_this_time(): + parent = await analyze("decompose", "additive", snaps=REFUSALS, **BY_REGION) + key = await _refused("decompose", "additive", parents={"A1": parent}, + **{**BY_REGION, "parent": {"of": "A1", "key": ["东北"]}}) + assert key["reason_code"] == "input_unavailable" and key["reason"] == "上一层「A1」里没有「东北」这一段" + period = await _refused("decompose", "additive", **{**BY_REGION, "target": "2026-W43", "label": "销售额，主要来自华东"}) + assert period["reason"] == "期次列「week」里没有本期「2026-W43」" and period["params"]["label"] == "销售额" + row = await _refused("decompose", "additive", **BY_REGION, headline={"base": "Q7.r5.gmv", "target": "Q7.r0.gmv"}) + assert row["reason"] == "基期总数那一格（Q7.r5.gmv）不在：Q7 这次只有 2 行" + assert [i["role"] for i in row["inputs"]] == ["data", "headline_base"] + empty = await _refused("decompose", "additive", **BY_REGION, headline={"base": "Q7.r1.gmv", "target": "Q7.r0.gmv"}) + assert empty["reason"] == "基期总数那一格（Q7.r1.gmv）是空的" + factor = await _refused("decompose", "factor", query="Q8", period="week", base="2026-W41", target="2026-W42", + product="gmv", factors=["orders", "aov"]) + assert factor["reason"] == "本期的「gmv」那一格是空的（Q8 第 1 行）" + target = await _refused("baseline", query="Q1", period="week", measure="gmv", target="2020-01-06") + assert target["reason"] == "期次列「week」里没有本期「2020-01-06」（或那一期为空）" + assert target["settings"] == {k: c.DEFAULTS[k] for k in ("k_warn", "k_alert", "min_history", "max_history")} + side = await _refused("comparable", query="Q2", period="week", base="2026-W40", target="2026-W42") + assert side["reason"] == "期次列「week」里没有基期「2026-W40」" and side["settings"] == {} + rates = {"by": ["region"], "numerator": "refunds", "denominator": "orders"} + group = await _refused("differ", "rate", query="Q3", **rates, pairs=[["华东", "东北"]]) + assert group["reason"] == "分组列「region」里没有「东北」这一组" + assert group["settings"] == {"differ_alpha": 0.05, "min_expected": 5} + few = await _refused("differ", "rate", query="Q9", **rates) + assert few["reason"] == "分组列「region」里只有 1 组，至少要两组才比得了" + weeks = {"query": "Q10", "period": "week", "numerator": "refunds", "denominator": "orders"} + gone = await _refused("differ", "rate", **weeks, base="2026-W40", target="2026-W41") + assert gone["reason"] == "期次列「week」里没有基期「2026-W40」" + null = await _refused("differ", "rate", **weeks, base="2026-W41", target="2026-W42") + assert null["reason"] == "「refunds」那一格是空的（Q10 第 1 行）" + + +async def test_structural_call_errors_still_fail_even_when_data_errors_are_refused(): + for params in ({**BY_REGION, "by": ["channel"]}, {**BY_REGION, "by": ["region", "week"]}): + with pytest.raises(CallError) as e: + await analyze("decompose", "additive", snaps=REFUSALS, refuse_data_errors=True, check=False, **params) + assert e.value.code not in analysis.DATA_ERRORS and e.value.reason is None