3、数据质量探查：缺失率、异常值、唯一值比例、跨表一致性校验的自动化脚本设计

数据质量是风控模型的基石。在特征工程启动前，必须对原始数据进行系统性探查，识别缺失、异常、冗余和不一致问题。本节设计一套自动化脚本，覆盖四个核心维度：缺失率、异常值、唯一值比例、跨表一致性校验。

3.1 探查维度与业务含义

3.2 自动化脚本架构设计

脚本采用模块化设计，每个探查维度对应一个独立函数，最终由主调度函数统一输出探查报告。以下为架构图：

3.3 核心代码实现

以下为四个探查模块的 Python 实现，基于 Pandas 和 NumPy，可直接集成到风控建模流水线中。

3.3.1 缺失率探查

def check_missing(df, threshold_high=0.5, threshold_medium=0.3):
    """
    探查各字段缺失率，返回缺失率及处理建议。
    """
    missing_series = df.isnull().mean()
    result = []
    for col, rate in missing_series.items():
        if rate > threshold_high:
            suggestion = '建议剔除'
        elif rate > threshold_medium:
            suggestion = '需分析缺失机制，考虑填充或标记'
        else:
            suggestion = '可填充或保留'
        result.append({
            '字段名': col,
            '缺失率': round(rate, 4),
            '建议操作': suggestion
        })
    return pd.DataFrame(result).sort_values('缺失率', ascending=False)

3.3.2 异常值探查（IQR方法）

def check_outlier(df, numeric_cols, method='iqr', multiplier=1.5):
    """
    对数值型字段使用IQR方法探查异常值比例。
    """
    result = []
    for col in numeric_cols:
        if col not in df.columns:
            continue
        series = df[col].dropna()
        if len(series) == 0:
            continue
        Q1 = series.quantile(0.25)
        Q3 = series.quantile(0.75)
        IQR = Q3 - Q1
        lower = Q1 - multiplier * IQR
        upper = Q3 + multiplier * IQR
        outlier_ratio = ((series < lower) | (series > upper)).mean()
        result.append({
            '字段名': col,
            '异常值占比': round(outlier_ratio, 4),
            '下界': round(lower, 2),
            '上界': round(upper, 2)
        })
    return pd.DataFrame(result).sort_values('异常值占比', ascending=False)

3.3.3 唯一值比例探查

def check_unique(df, low_threshold=0.01, high_threshold=0.95):
    """
    计算各字段唯一值比例，识别低区分度和类主键字段。
    """
    n = len(df)
    result = []
    for col in df.columns:
        unique_ratio = df[col].nunique() / n
        if unique_ratio < low_threshold:
            suggestion = '低区分度，建议合并或剔除'
        elif unique_ratio > high_threshold:
            suggestion = '类主键字段，建模时谨慎使用'
        else:
            suggestion = '正常'
        result.append({
            '字段名': col,
            '唯一值比例': round(unique_ratio, 4),
            '建议操作': suggestion
        })
    return pd.DataFrame(result).sort_values('唯一值比例', ascending=False)

3.3.4 跨表一致性校验

def check_consistency(df_left, df_right, key_col, check_cols):
    """
    基于主键关联两张表，检查指定字段的值一致性。
    返回不一致率及不一致样本示例。
    """
    merged = df_left.merge(df_right, on=key_col, suffixes=('_left', '_right'), how='inner')
    inconsistent_records = []
    for col in check_cols:
        left_col = col + '_left'
        right_col = col + '_right'
        if left_col not in merged.columns or right_col not in merged.columns:
            continue
        mismatch = merged[left_col] != merged[right_col]
        mismatch_rate = mismatch.mean()
        sample = merged[mismatch][[key_col, left_col, right_col]].head(3).to_dict('records')
        inconsistent_records.append({
            '字段名': col,
            '不一致率': round(mismatch_rate, 4),
            '不一致样本': sample
        })
    return pd.DataFrame(inconsistent_records).sort_values('不一致率', ascending=False)

3.4 主调度函数与报告整合

def run_quality_check(df, df_secondary=None, key_col=None, 
                      numeric_cols=None, check_cols=None):
    """
    主调度函数：依次执行四个探查模块，整合为一份完整报告。
    """
    report_parts = []
    
    # 1. 缺失率
    missing_report = check_missing(df)
    missing_report['探查维度'] = '缺失率'
    report_parts.append(missing_report)
    
    # 2. 异常值
    if numeric_cols is None:
        numeric_cols = df.select_dtypes(include=['int64', 'float64']).columns.tolist()
    outlier_report = check_outlier(df, numeric_cols)
    outlier_report['探查维度'] = '异常值'
    report_parts.append(outlier_report)
    
    # 3. 唯一值比例
    unique_report = check_unique(df)
    unique_report['探查维度'] = '唯一值比例'
    report_parts.append(unique_report)
    
    # 4. 跨表一致性
    if df_secondary is not None and key_col is not None and check_cols is not None:
        consistency_report = check_consistency(df, df_secondary, key_col, check_cols)
        consistency_report['探查维度'] = '跨表一致性'
        report_parts.append(consistency_report)
    
    final_report = pd.concat(report_parts, ignore_index=True)
    return final_report

3.5 探查结果可视化辅助

为快速定位问题字段，可对探查结果进行可视化。以下为缺失率与异常值占比的联合散点图示意：

3.6 探查结果决策矩阵

根据四个维度的综合结果，可制定如下处理策略：