[数据处理] 7. 数据质量评估

ayiya_Oese

2879人浏览 · 2025-05-09 08:09:43

ayiya_Oese · 2025-05-09 08:09:43 发布

👋 你好！这里有实用干货与深度分享✨✨ 若有帮助，欢迎：
👍 点赞 | ⭐ 收藏 | 💬 评论 | ➕ 关注，解锁更多精彩！
📁 收藏专栏即可第一时间获取最新推送🔔。
📖后续我将持续带来更多优质内容，期待与你一同探索知识，携手前行，共同进步🚀。

人工智能

数据质量评估

数据质量是影响深度学习模型性能与可靠性的重要因素。高质量的数据不仅能提升模型的泛化能力，还能降低过拟合风险。本文将系统介绍数据质量的核心维度、评估方法、监控机制及优化实践。

一、数据质量的核心维度

1. 完整性

样本完整性：是否包含足够数量和代表性的数据样本。
特征完整性：各个特征值是否存在缺失。
标签完整性：目标变量是否完整且准确标注。

2. 准确性

数值准确性：数值型数据是否在合理范围内。
类别准确性：分类变量是否符合预设的取值范围。
标签准确性：目标标签是否存在错误或标注偏差。

3. 一致性

格式一致性：字段格式是否标准统一（如日期、数字、文本格式等）。
单位一致性：不同字段或数据源中是否使用相同的度量单位。
编码一致性：类别变量或标签是否使用统一的编码方式。

二、数据质量评估方法

1. 基础统计分析

import pandas as pd

def analyze_data_quality(data):
    stats = {
        '样本数量': len(data),
        '特征数量': data.shape[1],
        '缺失率': data.isnull().mean(),
        '重复项数量': data.duplicated().sum(),
        '特征唯一值数量': data.nunique()
    }
    return pd.DataFrame(stats)

功能：快速评估样本量、缺失比例、重复项数量以及特征唯一性。

示例

import seaborn as sns

# 加载示例数据
iris = sns.load_dataset('iris')

# 数据质量分析
print(analyze_data_quality(iris))

2. 异常值检测

import numpy as np

# 异常值检测（Z-score 或 IQR）
def detect_outliers(data, column, method='zscore'):
    if method == 'zscore':
        z_scores = (data[column] - data[column].mean()) / data[column].std()
        return abs(z_scores) > 3
    elif method == 'iqr':
        Q1 = data[column].quantile(0.25)
        Q3 = data[column].quantile(0.75)
        IQR = Q3 - Q1
        return (data[column] < (Q1 - 1.5 * IQR)) | (data[column] > (Q3 + 1.5 * IQR))

功能：通过 Z-score 或 IQR 方法检测数值型数据中的离群值。

示例

# 检测异常值
outliers = detect_outliers(iris, 'sepal_length', method='iqr')
print(iris[outliers])

三、数据质量监控机制

1. 数据分布漂移监控

from scipy.stats import ks_2samp

# 监控分布漂移
def monitor_distribution(reference_data, current_data, column):
    ks_statistic, p_value = ks_2samp(reference_data[column], current_data[column])
    return {
        'ks_statistic': ks_statistic,
        'p_value': p_value,
        'distribution_shift': p_value < 0.05
    }

功能：通过 KS 检验监控新数据与参考数据的分布差异。

示例

# 模拟分布漂移
current_iris = iris.sample(frac=1.0)
print(monitor_distribution(iris, current_iris, 'sepal_length'))

2. 标签分布监控

from collections import Counter

# 标签分布监控
def monitor_labels(labels):
    label_dist = Counter(labels)
    balance_ratio = min(label_dist.values()) / max(label_dist.values())
    return {
        '标签分布': label_dist,
        '平衡比': balance_ratio
    }

功能：监控类别标签的分布与平衡性，识别数据倾斜问题。

示例

print(monitor_labels(iris['species']))

四、数据质量报告生成

1. 结构化报告生成

# 生成数据质量报告
def generate_quality_report(data):
   report = {
       'basic_info': {
           'rows': len(data),
           'columns': data.shape[1],
           'memory_usage_MB': data.memory_usage().sum() / 1024**2
       },
       'missing_info': {
           'total_missing': data.isnull().sum().sum(),
           'missing_by_column': data.isnull().sum().to_dict()
       },
       'duplicate_info': {
           'total_duplicates': data.duplicated().sum(),
           'duplicate_ratio': data.duplicated().mean()
       },
       'dtype_info': data.dtypes.to_dict()
   }
   return report

示例

print(generate_quality_report(iris))

2. 可视化缺失值分布

import matplotlib.pyplot as plt
import seaborn as sns

# 可视化缺失值分布
def plot_quality_report(report):
    plt.figure(figsize=(12, 6))
    missing_data = pd.Series(report['missing_info']['missing_by_column'])
    sns.barplot(x=missing_data.index, y=missing_data.values)
    plt.title('Missing Values by Column')
    plt.xticks(rotation=45)
    plt.show()

五、数据质量改进策略

常用清洗方法

def clean_data(data, strategies):
   cleaned_data = data.copy()
   for column, strategy in strategies.items():
       if strategy == 'drop':
           cleaned_data = cleaned_data.dropna(subset=[column])
       elif strategy == 'mean':
           cleaned_data[column].fillna(cleaned_data[column].mean(), inplace=True)
       elif strategy == 'mode':
           cleaned_data[column].fillna(cleaned_data[column].mode()[0], inplace=True)
   return cleaned_data

示例

strategies = {'sepal_length': 'mean', 'sepal_width': 'mode'}
cleaned_iris = clean_data(iris, strategies)
print(cleaned_iris.head())

数据验证规则

def validate_data(data, rules):
   validation_results = []
   for rule in rules:
       result = {
           'rule_name': rule['name'],
           'passed': eval(rule['condition']),
           'message': rule['message']
       }
       validation_results.append(result)
   return validation_results

示例

rules = [
   {'name': 'Sepal Length Range', 'condition': 'cleaned_iris["sepal_length"].between(4.0, 8.0)', 'message': 'Sepal length should be between 4.0 and 8.0'},
   {'name': 'Sepal Width Range', 'condition': 'cleaned_iris["sepal_width"].between(2.0, 4.5)','message': 'Sepal width should be between 2.0 and 4.5'}
]
print(validate_data(cleaned_iris, rules))

六、最佳实践建议

标准化质量评估流程

def quality_assessment_pipeline(data):
   # 1. 基础质量检查
   basic_quality = analyze_data_quality(data)
   # 2. 异常值检测
   outliers = {col: detect_outliers(data, col) for col in data.select_dtypes('number').columns}
   # 3. 分布监控
   distribution_checks = {
       col: monitor_distribution(reference_data, data, col)
       for col in data.columns if col in reference_data.columns
   }
   # 4. 质量报告生成
   report = generate_quality_report(data)
   return report