3. 数据质量评估指标体系

在风控模型沙盘推演中，数据质量评估是模型优化的起点。本章构建一套可量化的数据质量评分卡，并详细拆解四个核心指标的计算方法。通过这套体系，推演者能够快速定位数据短板，为后续清洗与特征工程提供决策依据。

3.1 数据质量评分卡框架

评分卡采用多维度加权评分机制，将数据质量抽象为四个一级维度：完整性、准确性、一致性与时效性。每个维度下包含若干可计算的关键指标，最终汇总为百分制总分。

3.2 核心指标计算方法

3.2.1 缺失率（Missing Rate）

缺失率衡量字段中空值、NULL或业务定义占位符（如-999、'Unknown'）的比例。计算公式如下：

缺失率 = (缺失记录数 / 总记录数) × 100%

计算步骤：

1. 确定目标字段，如 income（收入）。
2. 统计该字段中值为 NULL、空字符串或业务占位符的记录数。
3. 除以数据集总记录数，得到百分比。

3.2.2 异常率（Anomaly Rate）

异常率检测字段值是否超出业务定义的合理范围或违反逻辑约束。常见异常类型包括：

• 范围异常：年龄 < 0 或 > 120；收入为负数。
• 逻辑异常：申请日期 > 审批日期；性别字段出现非M/F值。
• 统计异常：通过IQR或Z-score方法识别离群点。

异常率 = (异常记录数 / 总记录数) × 100%

3.2.3 重复率（Duplicate Rate）

重复率衡量数据集中主键或业务键重复的记录比例。在风控模型中，重复数据会导致模型过拟合或样本偏差。

重复率 = (重复记录数 / 总记录数) × 100%

计算步骤：

1. 定义唯一标识键，如 user_id 或 application_id。
2. 统计该键出现次数大于1的记录数。
3. 除以总记录数，得到重复率。

3.2.4 时效偏差（Timeliness Deviation）

时效偏差衡量数据时间戳与标准参考时间之间的平均延迟。在实时风控场景中，高时效偏差会导致模型决策滞后。

时效偏差 = Σ(|数据时间戳 - 标准时间|) / 总记录数

单位：通常为秒、分钟或小时，根据业务场景确定。

3.3 数据质量评分卡计算流程

下图展示了从原始数据到最终评分的完整推演流程，包含四个核心指标的计算节点与加权汇总逻辑。

3.4 评分卡应用与阈值设定

在沙盘推演中，数据质量评分卡不仅用于评估，还用于触发质量预警与优化决策。建议设定三级阈值：

3.5 指标计算代码模板

以下Python代码模板可直接用于沙盘推演中的数据质量评估环节：

import pandas as pd
import numpy as np

def calculate_missing_rate(df, column, placeholder=[-999, 'Unknown']):
    """计算缺失率"""
    missing = df[column].isnull().sum()
    for p in placeholder:
        missing += (df[column] == p).sum()
    return missing / len(df) * 100

def calculate_anomaly_rate(df, column, lower_bound, upper_bound):
    """计算范围异常率"""
    anomalies = ((df[column] < lower_bound) | (df[column] > upper_bound)).sum()
    return anomalies / len(df) * 100

def calculate_duplicate_rate(df, key_column):
    """计算重复率"""
    duplicates = df.duplicated(subset=[key_column], keep=False).sum()
    return duplicates / len(df) * 100

def calculate_timeliness_deviation(df, time_column, reference_time):
    """计算时效偏差（分钟）"""
    df[time_column] = pd.to_datetime(df[time_column])
    deviation = (reference_time - df[time_column]).dt.total_seconds().abs() / 60
    return deviation.mean()

# 使用示例
df = pd.read_csv('risk_data.csv')
missing_rate = calculate_missing_rate(df, 'income')
anomaly_rate = calculate_anomaly_rate(df, 'age', 18, 100)
duplicate_rate = calculate_duplicate_rate(df, 'user_id')
timeliness_dev = calculate_timeliness_deviation(df, 'event_time', pd.Timestamp.now())

print(f"缺失率: {missing_rate:.2f}%")
print(f"异常率: {anomaly_rate:.2f}%")
print(f"重复率: {duplicate_rate:.2f}%")
print(f"时效偏差: {timeliness_dev:.2f}分钟")