mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 05:11:59 +00:00
ThearchyHelios
156 lines
10 KiB
Python
156 lines
10 KiB
Python
"""测试用的分析结论（只给测试用）：第 1 波各包并行开发时，证据接入、入口、说法核对的测试不等算子写完，按契约的形状
|
||
拿一份假的结论。算子落地后，端到端的测试改用 analysis.run 真算。
|
||
|
||
每个算子的假结论带它自己形状的结果表、具名标量、参数和输入（列 id、叫法、类型照各算子的实现：变化拆解按维度、波动基线和
|
||
趋势按期次、差异检验按对、可比性检查按检查项），数是编的、彼此对得上判定语。"""
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
from app.analysis.contract import METHODS, AnalysisResult, effective_settings
|
||
|
||
_PHRASES = {
|
||
"driver": "销售额下降 194 万（-8.7%），主要来自华东（-120 万，占 62%）",
|
||
"concentrated": "销售额下降 194 万（-8.7%），集中在华东、华北（合计占 85%）",
|
||
"dispersed": "销售额下降 194 万（-8.7%），分散在各地区，没有哪一段占到一半",
|
||
"within_band": "销售额本期变化 -3.1%，在过去 12 周的常规波动范围内（-6.1% 至 +5.3%）",
|
||
"outside_band": "销售额本期变化 -8.7%，超出过去 12 周的常规波动范围（-6.1% 至 +5.3%）",
|
||
"rising": "过去 10 周线上销售额呈上升趋势（每周约 +3.0%）",
|
||
"no_evidence": "西南和华东的退款率差异可能只是样本波动（样本量 40、230）",
|
||
"comparable": "两期可比",
|
||
"incomparable": "两期不可比：本期只有 3 天数据",
|
||
}
|
||
|
||
_WEEKS = ["2026-07-20", "2026-07-27", "2026-08-03", "2026-08-10", "2026-08-17", "2026-08-24", "2026-08-31",
|
||
"2026-09-07", "2026-09-14", "2026-09-21", "2026-09-28", "2026-10-05", "2026-10-12"]
|
||
|
||
|
||
def _decompose() -> tuple[dict[str, Any], dict[str, Any], dict[str, Any], list[str]]:
|
||
table = {
|
||
"columns": ["region", "base", "target", "delta", "pct", "share", "flag"],
|
||
"labels": {"region": "地区", "base": "基期", "target": "本期", "delta": "变化", "pct": "变化率",
|
||
"share": "占变化的比重", "flag": "说明"},
|
||
"rows": [["华东", "820", "700", "-120", "-0.1463", "0.6186", None],
|
||
["华北", "600", "560", "-40", "-0.0667", "0.2062", None],
|
||
["华南", "700", "666", "-34", "-0.0486", "0.1753", None],
|
||
["西南", "100", "100", "0", "0", "0", None]],
|
||
}
|
||
values = {"base_total": "2220", "target_total": "2026", "delta_total": "-194", "pct_total": "-0.0874",
|
||
"top_key": "华东", "top_share": "0.6186", "segments": "4"}
|
||
params = {"query": "Q2", "by": ["region"], "period": "week", "base": "W41", "target": "W42", "measure": "gmv",
|
||
"label": "销售额"}
|
||
return table, values, params, ["region", "week", "gmv"]
|
||
|
||
|
||
def _baseline() -> tuple[dict[str, Any], dict[str, Any], dict[str, Any], list[str]]:
|
||
changes = ["0.012", "-0.021", "0.034", "-0.008", "0.019", "-0.045", "0.027", "-0.015", "0.006", "0.041", "-0.032",
|
||
"-0.004"]
|
||
levels = ["2210", "2164", "2238", "2220", "2262", "2160", "2218", "2185", "2198", "2288", "2215", "2206"]
|
||
columns = ["week", "value", "change", "role"]
|
||
rows = [[w, v, c, "历史"] for w, v, c in zip(_WEEKS[:-1], levels, changes)] + [[_WEEKS[-1], "2138", "-0.031", "本期"]]
|
||
table = {"columns": columns, "labels": {"week": "周", "value": "销售额", "change": "变化率", "role": "说明"},
|
||
"rows": rows, "column_types": {"week": "text", "value": "number", "change": "number", "role": "text"}}
|
||
values = {"x": "-0.031", "value": "2138", "median": "0.001", "band_lo": "-0.061", "band_hi": "0.053", "z": "-1.12",
|
||
"rank": "11", "n_history": "12", "hist_min": "-0.045", "hist_max": "0.041"}
|
||
params = {"query": "Q1", "period": "week", "measure": "gmv", "label": "销售额", "grain": "week", "basis": "change",
|
||
"lag": 1, "target": _WEEKS[-1], "today": "2026-10-21"}
|
||
return table, values, params, ["week", "gmv"]
|
||
|
||
|
||
def _trend() -> tuple[dict[str, Any], dict[str, Any], dict[str, Any], list[str]]:
|
||
levels = [100, 103, 106, 109, 113, 116, 119, 123, 127, 130]
|
||
rows = [[w, str(v), str(v - levels[i - 1]) if i else None] for i, (w, v) in enumerate(zip(_WEEKS[-10:], levels))]
|
||
table = {"columns": ["week", "value", "change"], "labels": {"week": "周", "value": "线上销售额", "change": "比上一期"},
|
||
"rows": rows, "column_types": {"week": "text", "value": "number", "change": "number"}}
|
||
values = {"slope": "3.333333333333", "rel_slope": "0.03", "p": "0.000002755732", "s": "45", "n": "10",
|
||
"run_length": "9", "run_direction": "up"}
|
||
params = {"query": "Q1", "period": "week", "measure": "gmv", "label": "线上销售额", "grain": "week",
|
||
"today": "2026-10-21"}
|
||
return table, values, params, ["week", "gmv"]
|
||
|
||
|
||
def _differ() -> tuple[dict[str, Any], dict[str, Any], dict[str, Any], list[str]]:
|
||
columns = ["a", "b", "a_value", "b_value", "diff", "ci_lo", "ci_hi", "a_n", "b_n", "test", "p", "p_adj", "flag"]
|
||
table = {"columns": columns,
|
||
"labels": {"a": "组", "b": "对照组", "a_value": "组的比率", "b_value": "对照组的比率", "diff": "差",
|
||
"ci_lo": "差的区间下限", "ci_hi": "差的区间上限", "a_n": "组的样本量", "b_n": "对照组的样本量",
|
||
"test": "检验", "p": "p 值", "p_adj": "校正后的 p 值", "flag": "说明"},
|
||
"rows": [["西南", "华东", "0.05", "0.021739130435", "0.028260869565", "-0.026", "0.137", "40", "230",
|
||
"Fisher 精确检验", "0.27", "0.27", "样本少，用了精确检验"]],
|
||
"column_types": {"a": "text", "b": "text", **{c: "number" for c in columns[2:9]}, "test": "text",
|
||
"p": "number", "p_adj": "number", "flag": "text"}}
|
||
values = {"a": "0.05", "b": "0.021739130435", "diff": "0.028260869565", "ci_lo": "-0.026", "ci_hi": "0.137",
|
||
"p": "0.27", "p_adj": "0.27", "n_a": "40", "n_b": "230", "test": "fisher", "pairs": "1", "differs": "0"}
|
||
params = {"query": "Q3", "by": ["region"], "numerator": "refunds", "denominator": "orders", "label": "退款率",
|
||
"pairs": [["西南", "华东"]]}
|
||
return table, values, params, ["region", "refunds", "orders"]
|
||
|
||
|
||
def _comparable() -> tuple[dict[str, Any], dict[str, Any], dict[str, Any], list[str]]:
|
||
columns = ["check", "status", "base", "target", "note"]
|
||
table = {"columns": columns,
|
||
"labels": {"check": "检查项", "status": "结果", "base": "基期", "target": "本期", "note": "说明"},
|
||
"rows": [["同口径", "通过", None, None, None],
|
||
["窗口长度", "通过", "2026-10-05 到 2026-10-11（7 天）", "2026-10-12 到 2026-10-18（7 天）", None],
|
||
["整期", "通过", "通过", "通过", None], ["星期构成", "通过", None, None, None],
|
||
["数据问题", "通过", None, None, None]],
|
||
"column_types": dict.fromkeys(columns, "text")}
|
||
values = {"same_query": "pass", "window": "pass", "complete": "pass", "weekdays": "pass", "data": "pass",
|
||
"base_start": "2026-10-05", "base_end": "2026-10-11", "base_days": "7", "base_weekend": "2",
|
||
"target_start": "2026-10-12", "target_end": "2026-10-18", "target_days": "7", "target_weekend": "2"}
|
||
params = {"base_query": "Q4", "target_query": "Q5", "measure": "gmv", "today": "2026-10-21"}
|
||
return table, values, params, ["gmv"]
|
||
|
||
|
||
_SHAPES = {"decompose": _decompose, "baseline": _baseline, "trend": _trend, "differ": _differ,
|
||
"comparable": _comparable}
|
||
_CRITERIA = {"decompose": "一段占总变化一半以上算主要来源",
|
||
"baseline": "本期和过去 12 周的中位数比，差不到 2 倍稳健标准差算常规波动",
|
||
"trend": "Mann–Kendall 检验的双侧 p 值小于 0.05 时按 Theil–Sen 斜率的正负判上升或下降",
|
||
"differ": "双侧 p 值小于 0.05 算差异站得住",
|
||
"comparable": "只差日期、窗口一样长、是整期、星期构成相同才算可比"}
|
||
|
||
|
||
def fake_result(op: str = "decompose", mode: str | None = "additive", verdict: str | None = "driver", *,
|
||
subjects: list[list[str]] | None = None, direction: str | None = "down", status: str = "ok",
|
||
reason_code: str | None = None, inputs: list[dict[str, Any]] | None = None,
|
||
**overrides: Any) -> AnalysisResult:
|
||
"""一份形状完整的假结论。缺省是「销售额下降主要来自华东」的可加拆解；别的算子给各自形状的结果表、具名标量和参数。"""
|
||
if op not in ("decompose", "differ"):
|
||
mode = None
|
||
elif op == "differ" and mode not in ("rate", "mean", "rows"):
|
||
mode = "rate"
|
||
table, values, params, columns = _SHAPES.get(op, _decompose)()
|
||
if op == "comparable":
|
||
default_inputs = [{"ref": ref, "artifact": "0" * 64, "role": role, "columns": columns, "rows": "all"}
|
||
for ref, role in (("Q4", "base"), ("Q5", "target"))]
|
||
else:
|
||
default_inputs = [{"ref": str(params.get("query") or "Q2"), "artifact": "0" * 64, "role": "data",
|
||
"columns": columns, "rows": "all"}]
|
||
result: AnalysisResult = {
|
||
"op": op,
|
||
"mode": mode,
|
||
"method": METHODS.get((op, mode), f"{op}@1"),
|
||
"params": params,
|
||
"settings": effective_settings(None),
|
||
"inputs": inputs if inputs is not None else default_inputs,
|
||
"probes": [],
|
||
"status": status, # type: ignore[typeddict-item]
|
||
"reason_code": reason_code,
|
||
"reason": None if status == "ok" else "历史只有 5 期，至少要 8 期",
|
||
"table": table if status == "ok" else None, # type: ignore[typeddict-item]
|
||
"values": values if status == "ok" else {},
|
||
"verdict": ({"code": verdict, "subjects": subjects if subjects is not None else
|
||
([["华东"]] if verdict == "driver" else []), "direction": direction} # type: ignore[typeddict-item]
|
||
if status == "ok" and verdict else None),
|
||
"criterion": _CRITERIA.get(op, _CRITERIA["decompose"]),
|
||
"phrase": _PHRASES.get(verdict or "", "") if status == "ok" else "变化拆解没有做成：历史只有 5 期，至少要 8 期",
|
||
"caveats": [],
|
||
"hypothesis": "地区拖累",
|
||
}
|
||
result.update(overrides) # type: ignore[typeddict-item]
|
||
return result
|
||
|
||
|
||
__all__ = ["fake_result"]
|