分析能力入口（六）：口径卡的可比性挂钩（engine/nodes/metrics.py 的 _comparability）：一个指标的输入里取自查询快照的格（cell() 取的格、按位置取的格、Agent 按出处交来的字段）恰好落在两份不同的快照上，两份 SQL 按规范化（data/sql_semantics.normalize_sql，按数据源的方言）同形、只差字面量、查的是同一个数据源时（本周对上周），口径卡执行时调分析算子 comparable（早的那期作基期：第一处不同的字面量是日期或同样写法的期次的按先后，认不出的表达式里先出现的那份算本期），结论落成工件，指标上记 comparability（contract.MetricComparability：工件、判定、判定语、fail / warn 的原因，原因先按结论表的 status 一列取，没有再按判定语拆）。SQL 结构不同（退款额 / 销售额）、两次是同一条 SQL、数据源不同的不是同一个量的两期，不查也不记日志；同一对快照一张卡里只查一次。快照读不出来、探查失败、算子出错、算不了、还没有算子（第 1 波的 run 抛 NotImplementedError）都不挡口径卡，照实记一条 metric_comparability_skipped，这个指标不写 comparability

backend/app/engine/nodes/metrics.py

@@ -55,6 +55,9 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
 
     取数的那条 SQL 对照数据目录查出 error 级问题的（查询快照的 checks，见 data/sqlcheck.py：一对多关联后重复计算、
     存量跨期加总），同样值照算，标 sql_check_failed 并写明原因，出具契约据此降档（issuance.sql_check_gaps）。
+
+    一个指标的输入落在两份不同的查询快照上、两份 SQL 同形只差字面量（本周对上周）时，自动做一次可比性检查（分析算子
+    comparable），结论落成工件、记在指标的 comparability 上（_comparability）。做不成不挡口径卡，照实记一条日志。
     """
     definitions = ctx.cfg("metrics", []) or []
     caliber_raw, caliber_version = ctx.cfg("caliber", "") or ctx.node.title, str(ctx.cfg("caliber_version", "") or "v1")
@@ -84,6 +87,8 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
     failed_by: dict[str, list[QueryProblem]] = {}
     #: 查询快照工件 id → 快照本身（读输入的格落在哪一列「查询区间里这张表没有数据」，_no_data_inputs）
     snapshots: dict[str, Any] = {}
+    #: 可比性检查：同一对查询快照只查一次（几个指标都用本周、上周两份快照时），和做检查要用的几样（_Comparing）
+    comparing = _Comparing()
     for definition in definitions:
         metric_id = str(definition.get("id") or "").strip()
         expr = str(definition.get("expression") or "").strip()
@@ -167,6 +172,8 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
         no_data: dict[str, Any] = {}
         if lacking := _no_data_inputs(inputs, carried, snapshots):
             no_data = {"no_data": True, "no_data_reason": _no_data_reason(lacking), "no_data_sources": lacking}
+        comparability = await _comparability(metric_id, str(definition.get("name") or metric_id), inputs, ctx,
+                                             comparing, snapshots)
         metrics.append(
             {
                 "id": metric_id,
@@ -192,6 +199,8 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
                 **sql_check,
                 # 只在输入取自「查询区间里这张表没有数据」的列时出现，同上
                 **no_data,
+                # 只在输入落在两份同形、只差字面量的查询快照上、可比性检查做成了时出现（contract.MetricComparability）
+                **({"comparability": comparability} if comparability else {}),
             }
         )
 
@@ -462,6 +471,178 @@ def _rendered(value: Any, definition: dict[str, Any], display: dict[str, Any]) -
         return MISSING
 
 
+# --------------------------------------------------------------------------
+# 可比性检查（docs/design/analysis-round/00-总方案.md §3.4 第 3 条）
+# --------------------------------------------------------------------------
+
+#: 可比性检查没做成（探查失败、算子出错、还没有算子……）时那条日志的 code
+COMPARABILITY_SKIPPED = "metric_comparability_skipped"
+#: 那条日志的说法
+COMPARABILITY_SKIPPED_TEXT = "指标「{name}」用到的两次查询没做成可比性检查：{why}。指标照常计算，两期能不能直接比请自行核对"
+_DAY = re.compile(r"^(\d{4})-?(\d{2})-?(\d{2})(?:[ T].*)?$")
+
+
+class _Comparing:
+    """一张口径卡执行时做可比性检查记下的：同一对快照的结论（没做成的记 None，不再试）、数据源的类型（规范化 SQL 按
+    它的方言解析）。阈值、运行期依赖到真要检查时才读。"""
+
+    def __init__(self) -> None:
+        self.done: dict[tuple[str, str], dict[str, Any] | None] = {}
+        self.kinds: dict[str, str | None] = {}
+
+
+async def _comparability(metric_id: str, name: str, inputs: list[dict[str, Any]], ctx: NodeContext,
+                         comparing: _Comparing, snapshots: dict[str, Any]) -> dict[str, Any] | None:
+    """指标的可比性（contract.MetricComparability：结论工件、判定、判定语、原因）。只在这种时候做：输入里取自查询快照的格
+    （cell() 取的格、按位置取的格、Agent 按出处交来的字段）恰好落在两份不同的快照上，两份 SQL 同形、只差字面量，而且查的是
+    同一个数据源——本周对上周。SQL 结构不同的（退款额 / 销售额）不是同一个量的两期，不查。
+
+    哪份是基期：两边第一处不同的字面量是日期（或同样写法的期次）的，早的是基期；认不出的，表达式里先出现的那份算本期
+    （「(本周 - 上周) / 上周」的写法）。
+
+    做不成（快照读不出来、探查失败、算子出错、算不了、还没有算子）不挡口径卡：照实记一条日志（COMPARABILITY_SKIPPED），
+    这个指标不写 comparability。同一对快照只查一次。"""
+    from app.engine.lineage import SNAPSHOT_VIAS
+
+    artifacts = list(dict.fromkeys(str(i["artifact"]) for i in inputs if i.get("via") in SNAPSHOT_VIAS
+                                   and isinstance(i.get("artifact"), str) and i["artifact"]))
+    if len(artifacts) != 2:
+        return None
+    pair = (artifacts[0], artifacts[1])
+    key = (min(pair), max(pair))        # 本周减上周、上周减本周用的是同一对快照
+    if key in comparing.done:
+        return comparing.done[key]
+    found: dict[str, Any] | None = None
+    try:
+        found = await _check_comparable(pair, inputs, ctx, comparing, snapshots)
+    except _NotPeriods:
+        found = None
+    except Exception as e:  # noqa: BLE001 - 检查是附加的，不能挡住口径卡
+        why = e.message if isinstance(e, _Skipped) else describe_exception(e)
+        ctx.emit(EventType.LOG, level="warn", code=COMPARABILITY_SKIPPED, metric=metric_id, metric_name=name,
+                 reason=why, message=COMPARABILITY_SKIPPED_TEXT.format(name=name, why=why))
+    comparing.done[key] = found
+    return found
+
+
+class _NotPeriods(Exception):
+    """两份快照不是同一个量的两期（SQL 结构不同、数据源不同、字面量一样、没有 SQL）：不查，也不记日志。"""
+
+
+class _Skipped(Exception):
+    """该查、却做不成（结论算不了、判定认不出）：记一条日志，message 是给人看的原因。"""
+
+    def __init__(self, message: str) -> None:
+        super().__init__(message)
+        self.message = message
+
+
+async def _check_comparable(pair: tuple[str, str], inputs: list[dict[str, Any]], ctx: NodeContext,
+                            comparing: _Comparing, snapshots: dict[str, Any]) -> dict[str, Any]:
+    import app.analysis as analysis_ops   # 按模块取 run：测试、合并前的替身都只换模块上的这一个函数
+
+    from app.analysis.contract import VERDICTS
+    from app.data.sql_semantics import normalize_sql
+    from app.tools.analysis import (
+        context_of, input_snapshot, masked_of, run_snapshot, settings_of, store, tool_context,
+    )
+
+    contents = []
+    for artifact in pair:
+        if artifact not in snapshots:
+            try:
+                snapshots[artifact] = artifact_store.load(artifact)
+            except (ValueError, OSError):
+                snapshots[artifact] = None
+        if not isinstance(snapshots[artifact], dict):
+            raise _Skipped("查询快照读不出来")
+        contents.append(snapshots[artifact])
+    sqls = [c.get("sql") for c in contents]
+    sources = [str(c.get("source") or "") for c in contents]
+    if not all(isinstance(q, str) and q.strip() for q in sqls) or sources[0] != sources[1]:
+        raise _NotPeriods()
+    dialect = await _dialect_of(sources[0], comparing)
+    norms = [normalize_sql(q, dialect) for q in sqls]
+    if not norms[0].same_shape(norms[1]) or norms[0].literals == norms[1].literals:
+        raise _NotPeriods()
+    first, second = _period_order(norms[0].literals, norms[1].literals)
+    order = (pair[first], pair[second])           # (基期, 本期)
+    refs: dict[str, str] = {}
+    for role, artifact in zip(("base", "target"), order):
+        producer = next((str(i.get("node_id") or "") for i in inputs if i.get("artifact") == artifact), "")
+        refs[artifact] = producer if producer and producer not in refs.values() else f"{producer or 'query'}·{role}"
+    snaps = {}
+    for artifact in order:
+        content = snapshots[artifact]
+        snap = input_snapshot(content, ref=refs[artifact], artifact=artifact, node_id=refs[artifact].split("·")[0])
+        snap["masked"] = await masked_of(content, artifact)
+        snaps[refs[artifact]] = snap
+    settings = await settings_of(run_snapshot(ctx.run))
+    context = await context_of(list(snaps.values()), tool_ctx=tool_context(ctx.run, ctx.node.id, ctx.emit))
+    request = {"op": "comparable", "mode": None,
+               "params": {"base_query": refs[order[0]], "target_query": refs[order[1]]}}
+    result = await analysis_ops.run(request, snaps.get, settings=settings, context=context)
+    verdict = (result.get("verdict") or {}).get("code")
+    if result.get("status") != "ok" or verdict not in VERDICTS["comparable"]:
+        raise _Skipped(str(result.get("reason") or "检查没有给出判定"))
+    artifact = await store(result, run_id=ctx.run.run_id, node_id=ctx.node.id)
+    return {"artifact": artifact, "verdict": verdict, "phrase": str(result.get("phrase") or ""),
+            "reasons": comparability_reasons(result)}
+
+
+async def _dialect_of(source: str, comparing: _Comparing) -> str | None:
+    """数据源的类型（规范化 SQL 按它的方言解析）；合并查询在 SQLite 里执行。认不出为 None（通用读法）。"""
+    from app.engine.merge_query import MERGE_SOURCE
+
+    if source == MERGE_SOURCE:
+        return "sqlite"
+    if source not in comparing.kinds:
+        from sqlalchemy import select
+
+        from app.db.base import SessionLocal
+        from app.db.models import DataSource
+
+        async with SessionLocal() as session:
+            comparing.kinds[source] = (await session.execute(
+                select(DataSource.kind).where(DataSource.name == source))).scalar_one_or_none()
+    return comparing.kinds[source]
+
+
+def _period_order(a: tuple[str, ...], b: tuple[str, ...]) -> tuple[int, int]:
+    """两份同形 SQL 的字面量 → (基期的下标, 本期的下标)。第一处不同、两边都认得出的日期（YYYY-MM-DD、YYYYMMDD，带不带
+    时间）或同样长短的期次写法（2026-W41、202610）定先后，早的是基期；都认不出的，表达式里先出现的那份（下标 0）是本期。"""
+    for x, y in zip(a, b):
+        if x == y:
+            continue
+        left, right = x.strip("'\""), y.strip("'\"")
+        dx, dy = _DAY.match(left), _DAY.match(right)
+        if dx and dy:
+            if dx.groups() != dy.groups():
+                return (0, 1) if dx.groups() < dy.groups() else (1, 0)
+            continue
+        if len(left) == len(right) and re.fullmatch(r"[\dW\-/Qq]+", left) and re.fullmatch(r"[\dW\-/Qq]+", right):
+            return (0, 1) if left < right else (1, 0)
+    return 1, 0
+
+
+def comparability_reasons(result: dict[str, Any]) -> list[str]:
+    """可比性结论里 fail、warn 的原因，一句一个（contract.MetricComparability.reasons）。结果表里有 status 一列和一列说明
+    （reason / text / detail / message）的照表取；没有的按判定语「两期不可比：原因；原因」拆。可比的没有原因。"""
+    table = result.get("table") or {}
+    columns = [str(c) for c in table.get("columns") or []]
+    text_col = next((c for c in ("reason", "text", "detail", "message") if c in columns), None)
+    if "status" in columns and text_col is not None:
+        si, ti = columns.index("status"), columns.index(text_col)
+        found = [str(row[ti]) for row in table.get("rows") or []
+                 if isinstance(row, list) and len(row) > max(si, ti) and row[si] in ("fail", "warn") and row[ti]]
+        if found:
+            return found
+    phrase = str(result.get("phrase") or "")
+    if (result.get("verdict") or {}).get("code") == "comparable" or "：" not in phrase:
+        return []
+    return [part.strip() for part in phrase.split("：", 1)[1].split("；") if part.strip()]
+
+
 #: 清单里输入取自「查询区间里这张表没有数据」的列的指标后面那句：直接读清单的模型节点不能把 0 当成真的是 0
 LISTED_NO_DATA = "（这段时间表里没有数据：不要当成 0 解读）"
 #: 指标的 no_data_reason 打头那半句（后面接查询记下的说明）

backend/tests/test_metric_comparability.py

@@ -0,0 +1,168 @@
+"""口径卡的可比性挂钩（docs/design/analysis-round/00-总方案.md §3.4 第 3 条）：一个指标的输入落在两份不同的查询快照上、
+两份 SQL 同形只差字面量（本周对上周）时，口径卡执行时自动做一次可比性检查，结论落成工件、记在指标的 comparability 上。
+
+- 同形只差日期 → 有 comparability，早的那期是基期（和表达式里谁先写无关）；同一对快照只查一次
+- SQL 结构不同（退款额 / 销售额）、两次是同一条 SQL → 不查、不记日志
+- 算子出错、算不了、还没有算子（第 1 波的 run 抛 NotImplementedError）→ 不挡口径卡，记一条 metric_comparability_skipped
+"""
+from __future__ import annotations
+
+import asyncio
+import sqlite3
+
+import pytest
+from sqlalchemy import select
+
+from app.analysis.contract import CallError
+from app.core import artifact_store
+from app.data.engine import engines
+from app.db.base import SessionLocal
+from app.db.models import DataSource, Run, RunEvent
+from app.engine.nodes.metrics import COMPARABILITY_SKIPPED, comparability_reasons
+from app.engine.runner import run_manager
+from tests.fixtures.analysis_ops import FakeOps
+
+TOOL = "db_query__shop"
+THIS = "SELECT SUM(amount) AS gmv FROM orders WHERE day BETWEEN '2026-10-12' AND '2026-10-18'"
+LAST = "SELECT SUM(amount) AS gmv FROM orders WHERE day BETWEEN '2026-10-05' AND '2026-10-11'"
+REFUNDS = "SELECT SUM(refund) AS gmv FROM orders WHERE refund > 0 AND day BETWEEN '2026-10-12' AND '2026-10-18'"
+WOW = "(cell(nodes.this_week, 0, 'gmv') - cell(nodes.last_week, 0, 'gmv')) / cell(nodes.last_week, 0, 'gmv')"
+
+
+@pytest.fixture(autouse=True)
+async def engine_up():
+    await run_manager.setup()
+    yield
+    await run_manager.shutdown()
+
+
+@pytest.fixture(autouse=True)
+async def shop(tmp_path):
+    path = tmp_path / "shop.db"
+    db = sqlite3.connect(path)
+    db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, day TEXT, region TEXT, amount REAL, refund REAL);")
+    db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", [
+        (1, "2026-10-06", "华东", 500.0, 0.0), (2, "2026-10-08", "华南", 400.0, 20.0),
+        (3, "2026-10-13", "华东", 450.0, 0.0), (4, "2026-10-15", "华北", 300.0, 15.0)])
+    db.commit()
+    db.close()
+    async with SessionLocal() as session:
+        row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none()
+        if row is None:
+            row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True)
+            session.add(row)
+        row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, {}
+        await session.commit()
+        source_id = row.id
+    await engines.invalidate(source_id)
+    yield
+    await engines.invalidate(source_id)
+
+
+def node(nid, ntype, **config):
+    return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}}
+
+
+def graph(metrics, *, this=THIS, last=LAST):
+    nodes = [node("start", "input"),
+             node("this_week", "tool", tool=TOOL, args={"sql": this}),
+             node("last_week", "tool", tool=TOOL, args={"sql": last}),
+             node("card", "metrics", caliber="周报口径", metrics=metrics),
+             node("out", "output", fields=[{"name": "r", "value": "{{ nodes.card.text }}"}])]
+    return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
+
+
+async def run_card(g) -> tuple[Run, dict, list[dict]]:
+    run = await run_manager.start(graph=g, input_payload={})
+    for _ in range(400):
+        await asyncio.sleep(0.05)
+        async with SessionLocal() as session:
+            row = await session.get(Run, run.id)
+        if row.status in ("succeeded", "failed") and row.manifest_seq is not None:
+            break
+    assert row.status == "succeeded", row.error
+    async with SessionLocal() as session:
+        evts = [(e.type, e.node_id, e.data) for e in (await session.execute(
+            select(RunEvent).where(RunEvent.run_id == run.id).order_by(RunEvent.seq))).scalars()]
+    [finished] = [d for t, n, d in evts if t == "node.finished" and n == "card"]
+    card = artifact_store.load(finished["artifact"])
+    logs = [d for t, n, d in evts if t == "log" and n == "card"]
+    return row, card, logs
+
+
+def by_id(card):
+    return {m["id"]: m for m in card["metrics"]}
+
+
+async def test_two_periods_of_the_same_query_get_a_comparability_check(monkeypatch):
+    ops = FakeOps(monkeypatch)
+    _, card, logs = await run_card(graph([
+        {"id": "wow", "name": "销售额环比", "expression": WOW, "decimals": 4},
+        {"id": "diff", "name": "销售额变化", "expression": "cell(nodes.last_week, 0, 'gmv') - cell(nodes.this_week, 0, 'gmv')"},
+        {"id": "this", "name": "本周销售额", "expression": "cell(nodes.this_week, 0, 'gmv')"}]))
+    metrics = by_id(card)
+    [call] = ops.calls          # 同一对快照只查一次
+    # 早的那期是基期：表达式里先写的是本周还是上周都一样
+    assert call["request"] == {"op": "comparable", "mode": None,
+                               "params": {"base_query": "last_week", "target_query": "this_week"}}
+    base, target = call["snapshots"]["last_week"], call["snapshots"]["this_week"]
+    assert base["sql"] == LAST and target["sql"] == THIS and base["node_id"] == "last_week"
+    assert base["artifact"] == metrics["wow"]["inputs"][1]["artifact"] and base["masked"] == []
+    found = metrics["wow"]["comparability"]
+    assert found == metrics["diff"]["comparability"]
+    stored = artifact_store.load(found["artifact"])
+    assert stored["op"] == "comparable" and found == {"artifact": found["artifact"], "verdict": "comparable",
+                                                      "phrase": "两期可比", "reasons": []}
+    # 只用到一份快照的指标不查
+    assert "comparability" not in metrics["this"]
+    assert not [log for log in logs if log.get("code") == COMPARABILITY_SKIPPED]
+
+
+async def test_an_incomparable_pair_lists_its_reasons(monkeypatch):
+    FakeOps(monkeypatch, verdicts=["incomparable"])
+    _, card, _ = await run_card(graph([{"id": "wow", "name": "销售额环比", "expression": WOW}]))
+    found = by_id(card)["wow"]["comparability"]
+    assert found["verdict"] == "incomparable" and found["phrase"] == "两期不可比：本期只有 3 天数据"
+    assert found["reasons"] == ["本期只有 3 天数据"]
+
+
+async def test_different_structures_and_the_same_query_are_not_checked(monkeypatch):
+    ops = FakeOps(monkeypatch)
+    ratio = "cell(nodes.this_week, 0, 'gmv') / cell(nodes.last_week, 0, 'gmv')"
+    _, card, logs = await run_card(graph([{"id": "refund_rate", "name": "退款率", "expression": ratio}], last=REFUNDS))
+    assert "comparability" not in by_id(card)["refund_rate"] and ops.calls == []
+    _, card, logs2 = await run_card(graph([{"id": "same", "name": "同一条", "expression": ratio}], last=THIS))
+    assert "comparability" not in by_id(card)["same"] and ops.calls == []
+    assert not [log for log in [*logs, *logs2] if log.get("code") == COMPARABILITY_SKIPPED]
+
+
+async def test_without_the_operator_the_card_still_runs():
+    """第 1 波的 app.analysis.run 还抛 NotImplementedError：口径卡照常算，照实记一条没做成。"""
+    _, card, logs = await run_card(graph([{"id": "wow", "name": "销售额环比", "expression": WOW, "decimals": 4}]))
+    wow = by_id(card)["wow"]
+    assert "comparability" not in wow and wow["value"] == pytest.approx((450 + 300 - 900) / 900, abs=1e-4)
+    [log] = [log for log in logs if log.get("code") == COMPARABILITY_SKIPPED]
+    assert log["level"] == "warn" and log["metric"] == "wow" and "销售额环比" in log["message"]
+    assert "没做成可比性检查" in log["message"]
+
+
+@pytest.mark.parametrize("ops_kw", [
+    {"errors": [CallError("no_window", "认不出时间条件")]},
+    {"errors": [RuntimeError("探查超时")]},
+    {"statuses": ["refused"]},
+])
+async def test_failures_do_not_block_the_card(monkeypatch, ops_kw):
+    FakeOps(monkeypatch, **ops_kw)
+    _, card, logs = await run_card(graph([{"id": "wow", "name": "销售额环比", "expression": WOW}]))
+    assert "comparability" not in by_id(card)["wow"]
+    [log] = [log for log in logs if log.get("code") == COMPARABILITY_SKIPPED]
+    assert log["reason"]
+
+
+def test_reasons_come_from_the_check_table_when_there_is_one():
+    result = {"verdict": {"code": "conditional"}, "phrase": "两期有条件可比：两段的周末天数不同",
+              "table": {"columns": ["check", "status", "reason"], "labels": {},
+                        "rows": [["同口径", "pass", None], ["窗口长度", "warn", "窗口长度不同：7 天对 6 天"],
+                                 ["星期构成", "warn", "两段的周末天数不同：2 天对 1 天"], ["整期", "unknown", "核对不了"]]}}
+    assert comparability_reasons(result) == ["窗口长度不同：7 天对 6 天", "两段的周末天数不同：2 天对 1 天"]
+    assert comparability_reasons({"verdict": {"code": "comparable"}, "phrase": "两期可比"}) == []