forked from Gitlink/gitlink-cli
259 lines
10 KiB
Python
259 lines
10 KiB
Python
"""statgate：实验数据的证据门控与规范统计分析。
|
||
|
||
先问「数据够不够下这个结论」，再帮你跑规范的统计检验，最后出一份带效应量、
|
||
带假设检查的分析报告与统计附录——让你别瞎报，也让审稿人挑不出毛病。
|
||
|
||
三个核心能力：
|
||
1. 证据门控（先于检验）——在做任何检验前，检查样本量是否足够、缺失率是否过高、
|
||
分布是否严重偏态、组间方差是否悬殊，给出「能下结论 / 谨慎 / 证据不足」的判定
|
||
与原因，避免在不该下结论的数据上硬跑检验。
|
||
2. 规范统计检验——根据数据形态自动选择并运行合适的检验（双组用 t / Mann-Whitney，
|
||
多组用方差分析，列联表用卡方），同时报告效应量（Cohen's d），不只给 p 值。
|
||
3. 统计附录与报告——输出符合论文写法的统计描述（M±SD、检验统计量、df、p、效应量），
|
||
附录可直接粘进方法/结果部分。
|
||
|
||
纯标准库实现（见 stats_core.py），不依赖 numpy/scipy，可在受限环境运行。
|
||
|
||
用法：
|
||
python statgate.py --groups "12,14,11,13" "18,20,17,19" --paired false
|
||
python statgate.py --data data.json --output report.md
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Any, Sequence
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
import stats_core as sc
|
||
|
||
if hasattr(sys.stdout, "reconfigure"):
|
||
try:
|
||
sys.stdout.reconfigure(encoding="utf-8")
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
MIN_N = 5 # 每组最小样本量阈值
|
||
SKEW_LIMIT = 2.0 # 偏度绝对值告警阈值
|
||
VAR_RATIO_LIMIT = 4.0 # 方差比告警阈值（最大/最小）
|
||
|
||
|
||
def evidence_gate(groups: list[list[float]], raw_counts: list[int] | None = None) -> dict[str, Any]:
|
||
"""证据门控：检验前判断数据够不够下结论。"""
|
||
issues: list[str] = []
|
||
warnings: list[str] = []
|
||
|
||
# 样本量
|
||
small = [i + 1 for i, g in enumerate(groups) if len(g) < MIN_N]
|
||
if small:
|
||
issues.append(f"第 {small} 组样本量 < {MIN_N}，统计功效不足，结论不可靠。")
|
||
|
||
# 缺失率（若提供了原始条数）
|
||
if raw_counts:
|
||
for i, (g, raw) in enumerate(zip(groups, raw_counts), start=1):
|
||
if raw > 0:
|
||
miss = (raw - len(g)) / raw
|
||
if miss > 0.2:
|
||
warnings.append(f"第 {i} 组缺失率 {miss:.0%} 偏高（>20%），需说明缺失处理。")
|
||
|
||
# 偏态
|
||
for i, g in enumerate(groups, start=1):
|
||
if len(g) >= 3:
|
||
sk = sc.skewness(g)
|
||
if abs(sk) > SKEW_LIMIT:
|
||
warnings.append(f"第 {i} 组偏度 {sk:.2f} 较大，分布偏离正态，建议用非参数检验。")
|
||
|
||
# 方差齐性
|
||
variances = [sc.variance(g) for g in groups if len(g) >= 2]
|
||
if len(variances) >= 2:
|
||
vmin, vmax = min(variances), max(variances)
|
||
if vmin == 0 and vmax > 0:
|
||
warnings.append("存在方差为 0 的组（数据无变异），与其它组方差悬殊，"
|
||
"方差严重不齐，优先用非参数检验。")
|
||
elif vmin > 0:
|
||
ratio = vmax / vmin
|
||
if ratio > VAR_RATIO_LIMIT:
|
||
warnings.append(f"组间方差比 {ratio:.1f} 偏大（>{VAR_RATIO_LIMIT}），方差不齐，"
|
||
f"优先用 Welch / 非参数检验。")
|
||
|
||
if issues:
|
||
verdict = "证据不足"
|
||
elif warnings:
|
||
verdict = "谨慎下结论"
|
||
else:
|
||
verdict = "可下结论"
|
||
return {"verdict": verdict, "issues": issues, "warnings": warnings,
|
||
"recommend_nonparametric": any("非参数" in w for w in warnings)}
|
||
|
||
|
||
def describe(g: Sequence[float]) -> dict[str, float]:
|
||
return {"n": len(g), "mean": round(sc.mean(g), 4), "std": round(sc.std(g), 4),
|
||
"median": round(sc.median(g), 4)}
|
||
|
||
|
||
def run_analysis(groups: list[list[float]], paired: bool = False,
|
||
raw_counts: list[int] | None = None) -> dict[str, Any]:
|
||
"""完整分析：门控 → 选检验 → 效应量。"""
|
||
gate = evidence_gate(groups, raw_counts)
|
||
desc = [describe(g) for g in groups]
|
||
result: dict[str, Any] = {"gate": gate, "descriptives": desc, "n_groups": len(groups)}
|
||
|
||
use_np = gate["recommend_nonparametric"]
|
||
if len(groups) == 2:
|
||
a, b = groups
|
||
if paired:
|
||
test = sc.paired_t_test(a, b)
|
||
result["test"] = {"name": "配对 t 检验", **test}
|
||
elif use_np:
|
||
test = sc.mann_whitney_u(a, b)
|
||
result["test"] = {"name": "Mann-Whitney U 检验", **test}
|
||
else:
|
||
test = sc.welch_t_test(a, b)
|
||
result["test"] = {"name": "Welch 独立样本 t 检验", **test}
|
||
d = sc.cohens_d(a, b)
|
||
result["effect_size"] = {"cohens_d": round(d, 4), "magnitude": sc.interpret_d(d)}
|
||
elif len(groups) > 2:
|
||
test = sc.one_way_anova(*groups)
|
||
result["test"] = {"name": "单因素方差分析", **test}
|
||
else:
|
||
result["test"] = {"name": "样本组不足（需 ≥ 2 组）"}
|
||
|
||
return result
|
||
|
||
|
||
def _sig_mark(p: float) -> str:
|
||
if p < 0.001:
|
||
return "***"
|
||
if p < 0.01:
|
||
return "**"
|
||
if p < 0.05:
|
||
return "*"
|
||
return "n.s."
|
||
|
||
|
||
def render_report(result: dict[str, Any]) -> str:
|
||
gate = result["gate"]
|
||
lines = [
|
||
"# 统计分析报告",
|
||
"",
|
||
"## 一、证据门控",
|
||
"",
|
||
f"判定：**{gate['verdict']}**",
|
||
"",
|
||
]
|
||
if gate["issues"]:
|
||
lines.append("阻断性问题：")
|
||
for it in gate["issues"]:
|
||
lines.append(f"- {it}")
|
||
lines.append("")
|
||
if gate["warnings"]:
|
||
lines.append("提示：")
|
||
for w in gate["warnings"]:
|
||
lines.append(f"- {w}")
|
||
lines.append("")
|
||
if not gate["issues"] and not gate["warnings"]:
|
||
lines.append("样本量、分布、方差检查均通过，可进行参数检验。")
|
||
lines.append("")
|
||
|
||
lines += ["## 二、描述统计", "", "| 组 | n | 均值 | 标准差 | 中位数 |", "|:--:|:--:|:--:|:--:|:--:|"]
|
||
for i, d in enumerate(result["descriptives"], start=1):
|
||
lines.append(f"| {i} | {d['n']} | {d['mean']} | {d['std']} | {d['median']} |")
|
||
lines.append("")
|
||
|
||
test = result.get("test", {})
|
||
lines += ["## 三、假设检验", "", f"检验方法：{test.get('name', '—')}", ""]
|
||
if "p_value" in test:
|
||
p = test["p_value"]
|
||
stat_str = ""
|
||
if "t" in test:
|
||
stat_str = f"t({test['df']:.1f}) = {test['t']:.3f}"
|
||
elif "F" in test:
|
||
stat_str = f"F({test['df_between']}, {test['df_within']}) = {test['F']:.3f}"
|
||
elif "U" in test:
|
||
stat_str = f"U = {test['U']:.1f}, z = {test['z']:.3f}"
|
||
lines.append(f"- 统计量：{stat_str}")
|
||
lines.append(f"- p 值：{p:.4f} {_sig_mark(p)}")
|
||
lines.append(f"- 结论：{'差异显著' if p < 0.05 else '差异不显著'}（α = 0.05）")
|
||
lines.append("")
|
||
if "effect_size" in result:
|
||
es = result["effect_size"]
|
||
lines.append(f"- 效应量 Cohen's d = {es['cohens_d']}（{es['magnitude']}）")
|
||
lines.append("")
|
||
|
||
# 统计附录（论文可粘）
|
||
lines += ["## 四、统计附录（可粘入论文）", ""]
|
||
lines.append(_appendix_sentence(result))
|
||
lines += ["", "---", "",
|
||
"由 statgate 生成。先做证据门控再跑检验，统计量用纯标准库计算。"
|
||
"显著性标记：*** p<.001, ** p<.01, * p<.05, n.s. 不显著。"]
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _appendix_sentence(result: dict[str, Any]) -> str:
|
||
"""生成论文风格的一句话统计描述。"""
|
||
test = result.get("test", {})
|
||
desc = result["descriptives"]
|
||
if "p_value" not in test:
|
||
return "（检验未执行，无法生成附录句。）"
|
||
p = test["p_value"]
|
||
p_str = "p < .001" if p < 0.001 else f"p = {p:.3f}"
|
||
if len(desc) == 2 and "t" in test:
|
||
d = result.get("effect_size", {}).get("cohens_d", 0)
|
||
return (f"两组（M₁ = {desc[0]['mean']} ± {desc[0]['std']}，"
|
||
f"M₂ = {desc[1]['mean']} ± {desc[1]['std']}）经{test['name']}比较，"
|
||
f"t({test['df']:.1f}) = {test['t']:.2f}，{p_str}，Cohen's d = {d}。")
|
||
if "F" in test:
|
||
return (f"经单因素方差分析，各组差异 "
|
||
f"F({test['df_between']}, {test['df_within']}) = {test['F']:.2f}，{p_str}。")
|
||
if "U" in test:
|
||
return f"经 Mann-Whitney U 检验，U = {test['U']:.1f}，{p_str}。"
|
||
return "（无法生成附录句。）"
|
||
|
||
|
||
def _parse_group(s: str) -> list[float]:
|
||
return [float(x) for x in s.replace("，", ",").split(",") if x.strip()]
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
p = argparse.ArgumentParser(prog="statgate", description="证据门控与规范统计分析")
|
||
p.add_argument("--groups", nargs="+", help='每组数据，如 "12,14,11" "18,20,17"')
|
||
p.add_argument("--data", type=Path, help="JSON 文件：{\"groups\": [[...],[...]], \"paired\": false}")
|
||
p.add_argument("--paired", choices=["true", "false"], default="false", help="是否配对（仅两组）")
|
||
p.add_argument("--format", choices=["markdown", "json"], default="markdown")
|
||
p.add_argument("--output", type=Path)
|
||
args = p.parse_args(argv)
|
||
|
||
paired = args.paired == "true"
|
||
groups: list[list[float]] = []
|
||
raw_counts = None
|
||
if args.data and args.data.exists():
|
||
cfg = json.loads(args.data.read_text(encoding="utf-8-sig"))
|
||
groups = [[float(v) for v in g] for g in cfg.get("groups", [])]
|
||
paired = bool(cfg.get("paired", paired))
|
||
raw_counts = cfg.get("raw_counts")
|
||
elif args.groups:
|
||
groups = [_parse_group(g) for g in args.groups]
|
||
|
||
if len(groups) < 2:
|
||
print("错误：请用 --groups 或 --data 提供至少两组数据。", file=sys.stderr)
|
||
return 2
|
||
|
||
result = run_analysis(groups, paired=paired, raw_counts=raw_counts)
|
||
out = (json.dumps(result, ensure_ascii=False, indent=2) if args.format == "json"
|
||
else render_report(result))
|
||
if args.output:
|
||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||
args.output.write_text(out, encoding="utf-8")
|
||
print(f"已写入 {args.output}")
|
||
else:
|
||
print(out)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|