name: data-forensics description: > 原始数据真伪鉴定。触发条件: (1) 用户说"鉴定数据"/"数据鉴伪"/"data forensics"/"查假", (2) 接收客户数据时自动调用, (3) 用户说"这个数据可信吗"/"数据看起来真不真"。 核心能力: 7项统计鉴伪检验（GRIM/SPRITE/Benford/末位数字/方差齐性/p值分布/重复模式）， 输出红旗报告，作为接单前的风控关卡。 与 data-profiler（描述数据结构）和 research-integrity-audit（稿件-数据溯源）互补不重叠。 allowed-tools: Read, Write, Edit, Glob, Grep, Bash

Data Forensics - 原始数据真伪鉴定

核心理念

data-profiler 问"数据长什么样"，research-integrity-audit 问"稿件和数据一致吗"，本 Skill 问"数据本身是真的吗"。

三者形成自然管线：

客户数据 → data-profiler（认识）→ data-forensics（鉴真）→ statistical-analysis（分析）→ ... → research-integrity-audit（溯源）

造假数据有统计指纹。人手编的数据在末位数字、均值粒度、方差分布上都会留下痕迹。本 Skill 用 7 项经过学术验证的检验方法系统扫描这些指纹。

触发条件

1. 用户说 "鉴定数据" / "数据鉴伪" / "data forensics" / "查假" / "验真"
2. 用户说 "这个数据可信吗" / "数据看起来真不真" / "数据有没有问题"
3. paper-pipeline 在 Step 1.5（data-profiler 之后、statistical-analysis 之前）自动调用
4. 用户提供客户数据并说 "先验一下"

不触发:

• 用户只想了解数据结构 → data-profiler
• 用户想验证稿件数字 → research-integrity-audit
• 用户想跑统计分析 → statistical-analysis

输入

必需:

• 原始数据文件路径（CSV/Excel/SPSS/Stata）

可选:

• data_dictionary.md（来自 data-profiler，有则利用，无则自行推断）
• 论文/报告中的汇总统计量（用于 GRIM/SPRITE 验证已发表数据）
• 声称的样本量 N（用于校准检验力）

输出

• forensics_report.md — 完整鉴伪报告
• 风险评级：GREEN / YELLOW / RED
• 每项检验的详细结果和可视化描述

与现有 Skill 的边界（严格不重叠）

绝对不做的事:

• 不生成数据字典（data-profiler 的活）
• 不做 NTC 数字溯源（research-integrity-audit 的活）
• 不做统计分析（statistical-analysis 的活）
• 不评估稿件质量（manuscript-reviewer 的活）

工作流

+--------------------------------------------------+
|           数据鉴伪工作流 (Data Forensics)           |
|                                                    |
|  Step 0: 输入准备                                   |
|    +-- 读取原始数据                                  |
|    +-- 复用 data-profiler 输出（如有）                |
|    +-- 识别数值变量 / 分类变量 / 汇总统计量           |
|                                                    |
|  Step 1: 七项鉴伪检验（并行执行）                     |
|    +-- T1: GRIM Test（均值粒度检验）                  |
|    +-- T2: SPRITE（汇总统计量反推验证）               |
|    +-- T3: Benford's Law（首位数字分布）             |
|    +-- T4: Terminal Digit（末位数字分布）             |
|    +-- T5: Variance Uniformity（组间方差齐性异常）    |
|    +-- T6: Distribution Plausibility（分布合理性）   |
|    +-- T7: Duplicate Pattern（重复模式检测）          |
|                                                    |
|  Step 2: 综合评估                                   |
|    +-- 汇总各项红旗                                  |
|    +-- 计算综合风险评级                               |
|    +-- 考虑合理解释（非造假原因）                     |
|                                                    |
|  Step 3: 生成报告                                   |
|    +-- forensics_report.md                          |
|    +-- 风险评级 + 建议行动                            |
+--------------------------------------------------+

Step 0: 输入准备

import pandas as pd
import numpy as np
from scipy import stats
from collections import Counter

def prepare_input(file_path, data_dict_path=None):
    """准备鉴伪输入"""
    # 读取数据
    df = read_data(file_path)  # 复用 data-profiler 的读取逻辑

    # 分类变量
    numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
    categorical_cols = df.select_dtypes(include=['object', 'category']).columns.tolist()

    # 如果有 data-profiler 输出，复用其变量分类
    if data_dict_path:
        var_info = parse_data_dictionary(data_dict_path)
    else:
        var_info = infer_variable_types(df)

    # 识别连续变量 vs 离散变量（影响检验选择）
    continuous_cols = [c for c in numeric_cols
                       if df[c].nunique() > 20 and not var_info.get(c, {}).get('is_id')]
    discrete_cols = [c for c in numeric_cols
                      if df[c].nunique() <= 20]

    return df, {
        'numeric': numeric_cols,
        'continuous': continuous_cols,
        'discrete': discrete_cols,
        'categorical': categorical_cols,
        'n': len(df),
    }

Step 1: 七项鉴伪检验

T1: GRIM Test（均值粒度检验）

原理: 给定整数数据的样本量 N，均值只能取特定的离散值。例如 N=20 的整数数据，均值只能是 X.X0 或 X.X5 的形式。如果报告的均值在数学上不可能，说明均值或 N 有问题。

适用范围: 离散/整数变量的分组均值（如问卷得分、计数数据）。

def grim_test(values, n, decimals=2):
    """GRIM Test: 检验均值在给定 N 下是否数学上可能

    Args:
        values: 报告的均值列表
        n: 样本量
        decimals: 均值的小数位数

    Returns:
        list of dict: 每个均值的检验结果
    """
    results = []
    granularity = 1 / n  # 均值的最小粒度

    for mean_val in values:
        # 均值 * N 应该是整数（允许浮点误差）
        product = mean_val * n
        remainder = product - round(product)
        is_consistent = abs(remainder) < 10 ** (-decimals)

        results.append({
            'reported_mean': mean_val,
            'n': n,
            'product': product,
            'remainder': remainder,
            'consistent': is_consistent,
            'granularity': granularity,
        })

    return results


def grim_test_from_data(df, group_col, value_cols):
    """从原始数据执行 GRIM 检验 — 检查分组均值的内部一致性

    对比: 直接计算的均值 vs 四舍五入后是否匹配报告值
    这里用于检测数据本身的内部一致性
    """
    flags = []
    for col in value_cols:
        if not is_integer_like(df[col]):
            continue  # GRIM 只适用于整数型数据

        for group, subset in df.groupby(group_col):
            n = len(subset)
            if n < 5:
                continue
            mean = subset[col].mean()
            # 检验: mean * n 是否为整数
            product = mean * n
            if abs(product - round(product)) > 0.01:
                flags.append({
                    'test': 'GRIM',
                    'variable': col,
                    'group': group,
                    'n': n,
                    'mean': round(mean, 4),
                    'product': round(product, 4),
                    'flag': 'GRIM_INCONSISTENT',
                    'severity': 'HIGH',
                    'note': f'mean*n={product:.4f}, 应为整数但偏差>{abs(product - round(product)):.4f}',
                })
    return flags


def is_integer_like(series):
    """判断一列数据是否本质上是整数"""
    clean = series.dropna()
    if len(clean) == 0:
        return False
    return np.allclose(clean, clean.round(0), atol=0.001)

T2: SPRITE（汇总统计量反推验证）

原理: 给定 N、mean、SD、min、max，通过蒙特卡洛模拟生成所有可能的原始数据组合，检验这组汇总统计量是否存在至少一个合法的原始数据集。

适用范围: 有明确边界的离散变量（如 Likert 量表 1-5 分）。

def sprite_test(n, mean, sd, min_val, max_val, n_iter=10000):
    """SPRITE: 反推检验汇总统计量是否自洽

    Args:
        n: 样本量
        mean: 报告的均值
        sd: 报告的标准差
        min_val: 可能的最小值
        max_val: 可能的最大值
        n_iter: 蒙特卡洛迭代次数

    Returns:
        dict: 检验结果
    """
    target_sum = round(mean * n)
    target_var = sd ** 2

    # 先检查 GRIM：sum 必须是整数（如果是整数数据）
    if abs(mean * n - target_sum) > 0.5:
        return {'consistent': False, 'reason': 'GRIM_FAIL', 'attempts': 0}

    # 蒙特卡洛搜索
    found = False
    for _ in range(n_iter):
        # 生成随机整数数组，约束总和
        arr = generate_constrained_array(n, target_sum, min_val, max_val)
        if arr is None:
            continue
        # 检查 SD 是否匹配（允许四舍五入容差）
        # [v1.1 校准] 放宽 SD 容差，避免窄范围数据的假阳性
        sd_tolerance = max(0.1, sd * 0.05)
        if abs(np.std(arr, ddof=1) - sd) < sd_tolerance:
            found = True
            break

    return {
        'consistent': found,
        'reason': 'MATCH_FOUND' if found else 'NO_VALID_DATASET',
        'attempts': n_iter,
        'params': {'n': n, 'mean': mean, 'sd': sd, 'range': f'{min_val}-{max_val}'},
    }


def sprite_from_data(df, discrete_cols, group_col=None):
    """从原始数据对离散变量执行 SPRITE 检验"""
    flags = []
    groups = df.groupby(group_col) if group_col else [('all', df)]

    for group_name, subset in groups:
        for col in discrete_cols:
            clean = subset[col].dropna()
            if len(clean) < 10:
                continue
            if not is_integer_like(clean):
                continue

            n = len(clean)
            value_range = int(clean.max()) - int(clean.min())

            # [v1.1 校准] SPRITE 仅适用于小样本 + 窄值域
            # 大 N 或宽值域下蒙特卡洛搜索空间过大，无法在有限迭代内
            # 找到合法数据集，导致假阳性（真实数据被误判）
            if n > 50:
                continue  # N>50 时搜索空间爆炸，跳过
            if value_range > 10:
                continue  # 值域>10 时组合数过多，跳过

            result = sprite_test(
                n=n,
                mean=clean.mean(),
                sd=clean.std(ddof=1),
                min_val=int(clean.min()),
                max_val=int(clean.max()),
                # [v1.1] 增大 SD 容差，允许 max(0.1, sd*0.05) 的偏差
            )
            if not result['consistent']:
                flags.append({
                    'test': 'SPRITE',
                    'variable': col,
                    'group': group_name,
                    'flag': 'SPRITE_INCONSISTENT',
                    'severity': 'HIGH',
                    'detail': result,
                })
    return flags

T3: Benford's Law（首位数字分布检验）

原理: 自然产生的数值数据，首位数字的分布遵循 Benford 定律（1 出现约 30.1%，2 约 17.6%...）。人为编造的数据通常首位数字分布更均匀。

适用范围: 跨越多个数量级的连续变量（如实验室检验值、费用、人口数据）。不适用于受限范围变量（如年龄、BMI）。

BENFORD_EXPECTED = {
    1: 0.301, 2: 0.176, 3: 0.125, 4: 0.097,
    5: 0.079, 6: 0.067, 7: 0.058, 8: 0.051, 9: 0.046
}

def benford_test(series, alpha=0.05):
    """Benford's Law 首位数字检验

    Args:
        series: 数值序列
        alpha: 显著性水平

    Returns:
        dict: 检验结果
    """
    clean = series.dropna()
    clean = clean[clean > 0]  # Benford 仅适用于正数

    if len(clean) < 100:
        return {'applicable': False, 'reason': 'N < 100, 检验力不足'}

    # 检查数据是否跨越足够的数量级
    magnitude_range = np.log10(clean.max()) - np.log10(clean.min())
    if magnitude_range < 1.5:
        return {'applicable': False, 'reason': f'数量级跨度仅 {magnitude_range:.1f}, 不适用 Benford'}

    # 提取首位数字
    first_digits = clean.apply(lambda x: int(str(abs(x)).lstrip('0').lstrip('.')[0]))
    observed = Counter(first_digits)
    n = len(first_digits)

    # 卡方检验
    chi2 = 0
    digit_details = []
    for d in range(1, 10):
        obs = observed.get(d, 0)
        exp = BENFORD_EXPECTED[d] * n
        chi2 += (obs - exp) ** 2 / exp
        digit_details.append({
            'digit': d,
            'observed_pct': obs / n * 100,
            'expected_pct': BENFORD_EXPECTED[d] * 100,
            'deviation': (obs / n - BENFORD_EXPECTED[d]) * 100,
        })

    p_value = 1 - stats.chi2.cdf(chi2, df=8)

    # MAD (Mean Absolute Deviation) — 比卡方更稳健的 Benford 偏离指标
    mad = np.mean([abs(d['observed_pct'] - d['expected_pct']) for d in digit_details])
    # Nigrini (2012) 的 MAD 阈值: <0.6 接近一致, 0.6-1.2 可接受, 1.2-1.5 边缘, >1.5 不一致
    mad_verdict = (
        'CONFORMING' if mad < 0.6 else
        'ACCEPTABLE' if mad < 1.2 else
        'MARGINAL' if mad < 1.5 else
        'NONCONFORMING'
    )

    return {
        'applicable': True,
        'chi2': chi2,
        'p_value': p_value,
        'mad': mad,
        'mad_verdict': mad_verdict,
        'significant': p_value < alpha,
        'digit_details': digit_details,
        'n': n,
        'magnitude_range': magnitude_range,
    }

T4: Terminal Digit Analysis（末位数字分布检验）

原理: 真实测量数据的末位数字应近似均匀分布（0-9 各约 10%）。人为编造的数据末位数字常偏好 0 和 5（心理偏好整数），或避免某些数字。

适用范围: 任何包含小数的连续测量值。

def terminal_digit_test(series, alpha=0.05):
    """末位数字均匀性检验

    Args:
        series: 数值序列
        alpha: 显著性水平

    Returns:
        dict: 检验结果
    """
    clean = series.dropna()
    if len(clean) < 50:
        return {'applicable': False, 'reason': 'N < 50'}

    # 提取末位有效数字
    def get_last_digit(x):
        s = f'{x:.10g}'  # 去除尾随零
        s = s.rstrip('0').rstrip('.')
        if not s or s == '-':
            return None
        return int(s[-1])

    last_digits = clean.apply(get_last_digit).dropna().astype(int)
    n = len(last_digits)
    observed = Counter(last_digits)

    # 卡方检验（期望均匀分布）
    expected = n / 10
    chi2 = sum((observed.get(d, 0) - expected) ** 2 / expected for d in range(10))
    p_value = 1 - stats.chi2.cdf(chi2, df=9)

    # 特征分析
    zero_five_pct = (observed.get(0, 0) + observed.get(5, 0)) / n * 100

    # [v1.1 校准] 精度效应检测（Precision Effect Detection）
    # 如果所有值的小数位数完全一致（如全是1位小数），说明是仪器精度或
    # .round(N) 导致的末位数字非均匀——这是测量特征，不是造假信号。
    # 例: BMI 保留1位 → 末位只有 0-9 中的部分值，自然偏向 0 和 5。
    decimal_counts = clean.apply(count_decimals)
    precision_effect = decimal_counts.std() == 0 and decimal_counts.iloc[0] > 0

    # 根据是否存在精度效应，使用不同的判定阈值
    if precision_effect:
        # 有精度效应时放宽阈值：仅在极端偏好 0+5 时才标记
        zero_five_flag = zero_five_pct > 40
    else:
        # 无精度效应时使用组合标准：统计显著 且 0+5 偏高
        zero_five_flag = p_value < alpha and zero_five_pct > 30

    # 最高和最低频率数字
    digit_pcts = {d: observed.get(d, 0) / n * 100 for d in range(10)}
    max_digit = max(digit_pcts, key=digit_pcts.get)
    min_digit = min(digit_pcts, key=digit_pcts.get)

    return {
        'applicable': True,
        'chi2': chi2,
        'p_value': p_value,
        'significant': p_value < alpha,
        'zero_five_pct': zero_five_pct,
        'zero_five_flag': zero_five_flag,
        'precision_effect': precision_effect,  # [v1.1] 新增字段
        'digit_distribution': digit_pcts,
        'max_digit': (max_digit, digit_pcts[max_digit]),
        'min_digit': (min_digit, digit_pcts[min_digit]),
        'n': n,
    }

T5: Variance Uniformity（组间方差齐性异常检测）

原理: 真实实验中，不同组的方差应有自然波动。如果多个组的标准差几乎完全相同（变异系数 < 5%），这在自然数据中极不寻常，可能是编造的信号。

适用范围: 有分组变量的连续数据（如实验组/对照组的指标比较）。

def variance_uniformity_test(df, group_col, value_cols):
    """组间方差齐性异常检测

    Args:
        df: 数据框
        group_col: 分组变量
        value_cols: 待检验的连续变量列表

    Returns:
        list: 异常标记

    [v1.1 校准] CV 阈值改为动态，根据组数自适应:
    - 2-3 组: CV < 0.02 才标记（少组数时 SD 自然相近）
    - 4-6 组: CV < 0.03
    - 7+ 组:  CV < 0.05（多组数下 SD 一致才真正可疑）
    """
    flags = []

    for col in value_cols:
        group_stats = df.groupby(group_col)[col].agg(['std', 'mean', 'count'])
        group_stats = group_stats[group_stats['count'] >= 5]  # 至少 5 人

        n_groups = len(group_stats)
        if n_groups < 2:
            continue

        sds = group_stats['std'].values
        if np.any(sds == 0):
            flags.append({
                'test': 'VARIANCE_UNIFORMITY',
                'variable': col,
                'flag': 'ZERO_VARIANCE',
                'severity': 'HIGH',
                'note': '某组标准差为零 — 所有值完全相同',
                'group_sds': dict(zip(group_stats.index, sds)),
            })
            continue

        # SD 的变异系数 = SD(各组SD) / Mean(各组SD)
        cv_of_sd = np.std(sds, ddof=1) / np.mean(sds)

        # [v1.1] 动态阈值：组数越少，允许的 CV 下限越低
        if n_groups <= 3:
            cv_threshold = 0.02
        elif n_groups <= 6:
            cv_threshold = 0.03
        else:
            cv_threshold = 0.05

        if cv_of_sd < cv_threshold:
            flags.append({
                'test': 'VARIANCE_UNIFORMITY',
                'variable': col,
                'flag': 'SUSPICIOUSLY_UNIFORM_VARIANCE',
                'severity': 'MEDIUM',
                'cv_of_sd': round(cv_of_sd, 4),
                'threshold': cv_threshold,
                'group_sds': {str(k): round(v, 4) for k, v in zip(group_stats.index, sds)},
                'note': f'各组 SD 的变异系数仅 {cv_of_sd:.4f}，自然数据中极为罕见',
            })

    return flags

T6: Distribution Plausibility（分布合理性检验）

原理: 真实生物医学数据通常不会完美符合正态分布（Shapiro-Wilk p ≈ 1.0 是可疑的）。同时，真实数据也不应出现截断式的完美边界。检测"过于完美"和"明显不合理"两个方向。

def distribution_plausibility(df, continuous_cols):
    """分布合理性检验 — 检测过于完美或明显不合理的分布"""
    flags = []

    for col in continuous_cols:
        clean = df[col].dropna()
        if len(clean) < 20:
            continue

        # 1. 过于完美的正态性（p > 0.999）
        if len(clean) <= 5000:
            _, shapiro_p = stats.shapiro(clean)
            if shapiro_p > 0.999:
                flags.append({
                    'test': 'DISTRIBUTION',
                    'variable': col,
                    'flag': 'TOO_PERFECT_NORMAL',
                    'severity': 'MEDIUM',
                    'shapiro_p': shapiro_p,
                    'note': '数据过于完美地符合正态分布，真实生物数据中极罕见',
                })

        # 2. 偏度/峰度异常
        skew = clean.skew()
        kurt = clean.kurtosis()

        # 完全零偏度零峰度 = 人造正态
        if abs(skew) < 0.001 and abs(kurt) < 0.001 and len(clean) > 50:
            flags.append({
                'test': 'DISTRIBUTION',
                'variable': col,
                'flag': 'ZERO_SKEW_KURT',
                'severity': 'MEDIUM',
                'skewness': skew,
                'kurtosis': kurt,
                'note': '偏度和峰度同时近乎为零，高度可疑',
            })

        # 3. 值域合理性（医学常识检查）
        # 这部分需要领域知识，暂用通用规则
        if clean.min() == clean.max():
            flags.append({
                'test': 'DISTRIBUTION',
                'variable': col,
                'flag': 'CONSTANT_VALUE',
                'severity': 'HIGH',
                'note': f'所有值相同: {clean.iloc[0]}',
            })

        # 4. 异常精确的均值（小数位数过多且全组一致）
        decimal_places = clean.apply(count_decimals)
        if decimal_places.std() == 0 and decimal_places.iloc[0] > 4:
            flags.append({
                'test': 'DISTRIBUTION',
                'variable': col,
                'flag': 'UNIFORM_PRECISION',
                'severity': 'LOW',
                'decimal_places': int(decimal_places.iloc[0]),
                'note': '所有值的小数位数完全相同，可能是仪器特征或人为对齐',
            })

    return flags


def count_decimals(x):
    """计算小数位数"""
    s = f'{x:.15g}'
    if '.' in s:
        return len(s.split('.')[1].rstrip('0'))
    return 0

T7: Duplicate Pattern Detection（重复模式检测）

原理: 造假数据常出现：(a) 整行复制粘贴，(b) 块状重复（一段数据被复制到另一个位置），(c) 变量间异常高的完美相关。

def duplicate_pattern_test(df, numeric_cols):
    """重复模式检测"""
    flags = []

    # 1. 完全重复行
    dup_rows = df.duplicated(keep=False)
    n_dup = dup_rows.sum()
    if n_dup > 0:
        dup_pct = n_dup / len(df) * 100
        # 少量重复可能是合法的（如相同基线特征的患者）
        severity = 'HIGH' if dup_pct > 5 else 'MEDIUM' if dup_pct > 1 else 'LOW'
        flags.append({
            'test': 'DUPLICATE',
            'flag': 'DUPLICATE_ROWS',
            'severity': severity,
            'n_duplicates': int(n_dup),
            'pct': round(dup_pct, 1),
            'note': f'{n_dup} 行完全重复 ({dup_pct:.1f}%)',
        })

    # 2. 块状重复（相邻N行与另一段相同）
    block_flags = detect_block_duplicates(df, numeric_cols, min_block=3)
    flags.extend(block_flags)

    # 3. 异常完美相关（r > 0.999 但非数学派生关系）
    if len(numeric_cols) >= 2:
        for i, col_a in enumerate(numeric_cols[:20]):  # 限制计算量
            for col_b in numeric_cols[i+1:20]:
                clean = df[[col_a, col_b]].dropna()
                if len(clean) < 10:
                    continue
                r, _ = stats.pearsonr(clean[col_a], clean[col_b])
                if abs(r) > 0.999:
                    # 排除数学派生关系（如 BMI = weight/height^2）
                    if not is_derived_pair(col_a, col_b):
                        flags.append({
                            'test': 'DUPLICATE',
                            'flag': 'PERFECT_CORRELATION',
                            'severity': 'HIGH',
                            'variables': [col_a, col_b],
                            'correlation': round(r, 6),
                            'note': f'{col_a} 与 {col_b} 相关系数 {r:.6f}，非派生关系下极为可疑',
                        })

    return flags


def detect_block_duplicates(df, cols, min_block=3):
    """检测块状重复 — 连续 N 行与另一段完全相同"""
    flags = []
    values = df[cols].values
    n = len(values)

    for block_size in [min_block, 5, 10]:
        if n < block_size * 2:
            continue
        seen = {}
        for i in range(n - block_size + 1):
            block = tuple(map(tuple, values[i:i+block_size]))
            block_hash = hash(block)
            if block_hash in seen and abs(i - seen[block_hash]) >= block_size:
                flags.append({
                    'test': 'DUPLICATE',
                    'flag': 'BLOCK_DUPLICATE',
                    'severity': 'HIGH',
                    'block_size': block_size,
                    'location_1': seen[block_hash],
                    'location_2': i,
                    'note': f'行 {seen[block_hash]}-{seen[block_hash]+block_size-1} 与行 {i}-{i+block_size-1} 完全相同',
                })
                break  # 每种 block_size 只报告第一个
            seen[block_hash] = i

    return flags


def is_derived_pair(col_a, col_b):
    """判断两个变量是否可能是数学派生关系"""
    derived_patterns = [
        ('bmi', 'weight'), ('bmi', 'height'),
        ('total', 'sub'), ('sum', 'item'),
        ('rate', 'count'), ('pct', 'n'),
    ]
    a_lower, b_lower = col_a.lower(), col_b.lower()
    for p1, p2 in derived_patterns:
        if (p1 in a_lower and p2 in b_lower) or (p2 in a_lower and p1 in b_lower):
            return True
    return False

Step 2: 综合评估

def synthesize_results(all_flags, n_tests_run):
    """综合评估所有检验结果

    核心原则: 单项红旗可能有合理解释，多项红旗叠加才构成强证据
    """
    # 按严重性分类
    high_flags = [f for f in all_flags if f.get('severity') == 'HIGH']
    medium_flags = [f for f in all_flags if f.get('severity') == 'MEDIUM']
    low_flags = [f for f in all_flags if f.get('severity') == 'LOW']

    # 风险评分（加权）
    risk_score = len(high_flags) * 3 + len(medium_flags) * 1.5 + len(low_flags) * 0.5

    # 归一化到 0-10
    max_possible = n_tests_run * 3  # 假设每项检验最多产生一个 HIGH
    normalized_score = min(risk_score / max(max_possible, 1) * 10, 10)

    # 风险评级
    if len(high_flags) >= 3 or normalized_score >= 7:
        rating = 'RED'
        verdict = '高风险 — 多项统计指纹异常，强烈建议核实数据来源'
    elif len(high_flags) >= 1 or normalized_score >= 3.5:
        rating = 'YELLOW'
        verdict = '中风险 — 存在可疑信号，需进一步调查或要求客户解释'
    else:
        rating = 'GREEN'
        verdict = '低风险 — 未发现明显的统计造假指纹'

    # 考虑合理解释
    alternative_explanations = generate_alternatives(all_flags)

    return {
        'rating': rating,
        'verdict': verdict,
        'risk_score': round(normalized_score, 1),
        'high_flags': high_flags,
        'medium_flags': medium_flags,
        'low_flags': low_flags,
        'total_flags': len(all_flags),
        'tests_run': n_tests_run,
        'alternative_explanations': alternative_explanations,
    }


def generate_alternatives(flags):
    """为每类红旗生成合理的非造假解释

    重要: 避免冤枉诚实的研究者。每个红旗都可能有合理解释。
    """
    alternatives = {
        'GRIM_INCONSISTENT': '可能原因: 数据包含非整数值被四舍五入、分组后有缺失值未排除、均值计算包含了权重',
        'SPRITE_INCONSISTENT': '可能原因: 原始数据非严格整数、存在编码错误（非造假）、汇总统计量经过二次处理',
        'BENFORD_NONCONFORMING': '可能原因: 数据范围受限（如年龄20-90）、阈值效应（如临床截断值）、小样本波动',
        'TERMINAL_DIGIT_NONUNIFORM': '可能原因: 仪器精度限制（如血压只记录偶数）、特定编码规则、四舍五入到5的倍数',
        'SUSPICIOUSLY_UNIFORM_VARIANCE': '可能原因: 标准化后的数据、同质性高的亚群、变量范围很窄',
        'TOO_PERFECT_NORMAL': '可能原因: 大样本中心极限定理效应、数据经过正态化转换',
        'DUPLICATE_ROWS': '可能原因: 时间序列中的稳态测量、分类数据的自然重复（有限组合）',
        'BLOCK_DUPLICATE': '可能原因: 数据录入系统的批量导入错误（非故意造假）',
        'PERFECT_CORRELATION': '可能原因: 未识别的数学派生关系、同一指标的不同单位',
    }
    return {f['flag']: alternatives.get(f['flag'], '需具体分析')
            for f in flags if f['flag'] in alternatives}

Step 3: 生成报告

# 数据鉴伪报告 (Data Forensics Report)

**数据文件**: {file_path}
**鉴定日期**: {date}
**样本量**: {n}
**变量数**: {n_vars}

---

## 综合评级: {rating_badge}

**风险评分**: {risk_score} / 10
**判定**: {verdict}

---

## 检验结果汇总

| 检验 | 适用变量数 | 通过 | 红旗 | 最高严重性 |
|------|-----------|------|------|-----------|
| T1: GRIM Test | {n1} | {pass1} | {flag1} | {sev1} |
| T2: SPRITE | {n2} | {pass2} | {flag2} | {sev2} |
| T3: Benford's Law | {n3} | {pass3} | {flag3} | {sev3} |
| T4: Terminal Digit | {n4} | {pass4} | {flag4} | {sev4} |
| T5: Variance Uniformity | {n5} | {pass5} | {flag5} | {sev5} |
| T6: Distribution Plausibility | {n6} | {pass6} | {flag6} | {sev6} |
| T7: Duplicate Pattern | {n7} | {pass7} | {flag7} | {sev7} |

---

## 红旗详情

### HIGH 级别
{high_flag_details}

### MEDIUM 级别
{medium_flag_details}

---

## 合理替代解释

{alternative_explanations}

**重要声明**: 统计检验只能识别异常模式，不能证明造假。RED 评级意味着"需要核实"而非"一定造假"。
最终判断应结合数据来源、采集过程、领域知识综合考虑。

---

## 建议行动

| 评级 | 行动 |
|------|------|
| GREEN | 正常推进分析流程 |
| YELLOW | 要求客户提供数据采集说明，解释可疑信号后再继续 |
| RED | 暂停接单，要求客户提供原始记录/伦理审批/数据采集 SOP |

管线集成

在 paper-pipeline 中的位置

Step 1:   data-profiler        → 数据字典
Step 1.5: data-forensics (NEW) → 鉴伪报告 + 风险评级
          |-- GREEN  → 继续
          |-- YELLOW → 警告 + 继续（报告附加到最终交付物）
          |-- RED    → 暂停，通知创始人决策
Step 2:   statistical-analysis → 分析报告
...
Step 7.5: research-integrity-audit → NTC 审计

独立使用（接单风控）

客户发来数据 → data-forensics → 风险评级
  |-- GREEN  → 正常报价接单
  |-- YELLOW → 接单，但合同注明"数据由客户负责"
  |-- RED    → 拒单，或要求客户先提供数据来源证明

与其他 Skill 的协作

关键原则: 只做鉴伪，不做其他。不生成数据字典，不做统计分析，不写稿件，不溯源。

局限性与伦理声明

1. 统计检验不等于证明: 红旗 ≠ 造假。每项异常都可能有合理解释。
2. 假阴性: 高水平的造假可能通过所有检验。本工具检测的是"统计指纹"，不是万能探测器。
3. 假阳性: 特殊领域的真实数据可能触发红旗（如受限范围变量不符合 Benford）。
4. 使用伦理: 本工具用于保护研究诚信，不用于恶意指控。RED 评级应触发对话和核实，而非直接定罪。
5. 领域适配: 当前检验方法主要适用于医学/社会科学的数值数据。基因组学、影像学等需要专门扩展。

学术参考

• Brown & Heathers (2017). The GRIM test. Social Psychological and Personality Science, 8(4), 363-369.
• Heathers et al. (2018). SPRITE: A response to "Escaping the Tyranny of the Mean". PsyArXiv.
• Nigrini (2012). Benford's Law: Applications for Forensic Accounting, Auditing, and Fraud Detection. Wiley.
• Simonsohn (2013). Just Post It: The Lesson from Two Cases of Fabricated Data. Psychological Science, 24(10), 1875-1888.
• Carlisle (2017). Data fabrication and other reasons for non-random sampling in 5087 randomised, controlled trials in anaesthetic and general medical journals. Anaesthesia, 72(8), 944-952.

红蓝对抗验证 (v1.1)

测试设计

5 个数据集 × 7 项检验的全矩阵对抗测试（tests/red_blue_battle.py）：

校准前问题（v1.0 → 真实数据 8 个红旗 = RED 假阳性）

校准后结果（v1.1 → 5/5 通过）

已知局限

1. L1 粗糙造假只到 YELLOW 不到 RED: 均匀分布虽然不符合 Benford，但 MAD 值未必极端到 NONCONFORMING 级别。如需更严格，可调低 Benford MAD 阈值。
2. L4 对抗造假的检出依赖隐藏相关: 如果对抗者不植入完美相关，仅用真实分布 + 微调参数，统计鉴伪很难检出。这是统计方法的根本极限——没有万能探测器。
3. SPRITE 适用范围窄: 校准后仅对 N<=50 且值域<=10 的离散变量有效。大样本、宽值域数据无法使用 SPRITE。

更新日志

• v1.1 (2026-03-06): 红蓝对抗校准
  • T2 SPRITE: 添加 N<=50 和 value_range<=10 护栏，放宽 SD 容差
  • T4 Terminal Digit: 新增精度效应检测（precision_effect），区分仪器精度 vs 造假
  • T5 Variance Uniformity: 固定阈值 → 动态阈值（按组数自适应）
  • 新增「红蓝对抗验证」章节，记录测试矩阵和校准过程
  • 校准结果: 真实数据 GREEN (0 旗), 4 种造假数据全部 YELLOW+ (检出)
• v1.0 (2026-03-06): 初始版本
  • 7 项统计鉴伪检验 (GRIM/SPRITE/Benford/Terminal Digit/Variance/Distribution/Duplicate)
  • 三级风险评级 (GREEN/YELLOW/RED)
  • 合理替代解释机制（避免冤枉诚实研究者）
  • paper-pipeline Step 1.5 门控集成
  • 独立接单风控模式