225 lines
8.8 KiB
Python
225 lines
8.8 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
数据质量检查模块 - 自动评估数据质量并提供改进建议
|
||
"""
|
||
|
||
import pandas as pd
|
||
import numpy as np
|
||
from typing import Dict, List, Tuple, Any
|
||
from dataclasses import dataclass
|
||
|
||
|
||
@dataclass
|
||
class QualityIssue:
|
||
"""数据质量问题"""
|
||
column: str
|
||
issue_type: str # missing, duplicate, outlier, type_mismatch等
|
||
severity: str # high, medium, low
|
||
description: str
|
||
suggestion: str
|
||
|
||
|
||
class DataQualityChecker:
|
||
"""数据质量检查器"""
|
||
|
||
def __init__(self, df: pd.DataFrame):
|
||
self.df = df
|
||
self.issues: List[QualityIssue] = []
|
||
self.quality_score: float = 100.0
|
||
|
||
def check_all(self) -> Dict[str, Any]:
|
||
"""执行所有质量检查"""
|
||
self.check_missing_values()
|
||
self.check_duplicates()
|
||
self.check_data_types()
|
||
self.check_outliers()
|
||
self.check_consistency()
|
||
|
||
return self.generate_report()
|
||
|
||
def check_missing_values(self) -> None:
|
||
"""检查缺失值"""
|
||
for col in self.df.columns:
|
||
missing_count = self.df[col].isnull().sum()
|
||
missing_ratio = (missing_count / len(self.df)) * 100
|
||
|
||
if missing_ratio > 50:
|
||
severity = "high"
|
||
self.quality_score -= 10
|
||
elif missing_ratio > 20:
|
||
severity = "medium"
|
||
self.quality_score -= 5
|
||
elif missing_ratio > 0:
|
||
severity = "low"
|
||
self.quality_score -= 2
|
||
else:
|
||
continue
|
||
|
||
issue = QualityIssue(
|
||
column=col,
|
||
issue_type="missing",
|
||
severity=severity,
|
||
description=f"列 '{col}' 存在 {missing_count} 个缺失值 ({missing_ratio:.1f}%)",
|
||
suggestion=self._suggest_missing_handling(col, missing_ratio)
|
||
)
|
||
self.issues.append(issue)
|
||
|
||
def check_duplicates(self) -> None:
|
||
"""检查重复数据"""
|
||
duplicate_count = self.df.duplicated().sum()
|
||
if duplicate_count > 0:
|
||
duplicate_ratio = (duplicate_count / len(self.df)) * 100
|
||
|
||
severity = "high" if duplicate_ratio > 10 else "medium"
|
||
self.quality_score -= 5 if severity == "high" else 3
|
||
|
||
issue = QualityIssue(
|
||
column="全表",
|
||
issue_type="duplicate",
|
||
severity=severity,
|
||
description=f"发现 {duplicate_count} 行重复数据 ({duplicate_ratio:.1f}%)",
|
||
suggestion="建议使用 df.drop_duplicates() 删除重复行，或检查是否为合理的重复记录"
|
||
)
|
||
self.issues.append(issue)
|
||
|
||
def check_data_types(self) -> None:
|
||
"""检查数据类型一致性"""
|
||
for col in self.df.columns:
|
||
# 检查是否有数值列被识别为object
|
||
if self.df[col].dtype == 'object':
|
||
try:
|
||
# 尝试转换为数值
|
||
pd.to_numeric(self.df[col].dropna(), errors='raise')
|
||
|
||
issue = QualityIssue(
|
||
column=col,
|
||
issue_type="type_mismatch",
|
||
severity="medium",
|
||
description=f"列 '{col}' 当前为文本类型，但可以转换为数值类型",
|
||
suggestion=f"建议使用 df['{col}'] = pd.to_numeric(df['{col}']) 转换类型"
|
||
)
|
||
self.issues.append(issue)
|
||
self.quality_score -= 3
|
||
except:
|
||
pass
|
||
|
||
def check_outliers(self) -> None:
|
||
"""检查数值列的异常值"""
|
||
numeric_cols = self.df.select_dtypes(include=[np.number]).columns
|
||
|
||
for col in numeric_cols:
|
||
q1 = self.df[col].quantile(0.25)
|
||
q3 = self.df[col].quantile(0.75)
|
||
iqr = q3 - q1
|
||
|
||
lower_bound = q1 - 3 * iqr
|
||
upper_bound = q3 + 3 * iqr
|
||
|
||
outliers = self.df[(self.df[col] < lower_bound) | (self.df[col] > upper_bound)]
|
||
outlier_count = len(outliers)
|
||
|
||
if outlier_count > 0:
|
||
outlier_ratio = (outlier_count / len(self.df)) * 100
|
||
|
||
if outlier_ratio > 5:
|
||
severity = "medium"
|
||
self.quality_score -= 3
|
||
else:
|
||
severity = "low"
|
||
self.quality_score -= 1
|
||
|
||
issue = QualityIssue(
|
||
column=col,
|
||
issue_type="outlier",
|
||
severity=severity,
|
||
description=f"列 '{col}' 存在 {outlier_count} 个异常值 ({outlier_ratio:.1f}%)",
|
||
suggestion=f"建议检查 {lower_bound:.2f} 以下和 {upper_bound:.2f} 以上的值是否合理"
|
||
)
|
||
self.issues.append(issue)
|
||
|
||
def check_consistency(self) -> None:
|
||
"""检查数据一致性"""
|
||
# 检查时间列的时序性
|
||
datetime_cols = self.df.select_dtypes(include=['datetime64']).columns
|
||
|
||
for col in datetime_cols:
|
||
if not self.df[col].is_monotonic_increasing:
|
||
issue = QualityIssue(
|
||
column=col,
|
||
issue_type="consistency",
|
||
severity="medium",
|
||
description=f"时间列 '{col}' 不是单调递增的，可能存在乱序",
|
||
suggestion=f"建议使用 df.sort_values('{col}') 进行排序"
|
||
)
|
||
self.issues.append(issue)
|
||
self.quality_score -= 3
|
||
|
||
def _suggest_missing_handling(self, col: str, missing_ratio: float) -> str:
|
||
"""建议缺失值处理方法"""
|
||
if missing_ratio > 70:
|
||
return f"缺失比例过高，建议删除列 '{col}'"
|
||
elif missing_ratio > 30:
|
||
return f"建议填充或删除缺失值：使用中位数/众数填充或删除含缺失值的行"
|
||
else:
|
||
if pd.api.types.is_numeric_dtype(self.df[col]):
|
||
return f"建议使用均值/中位数填充：df['{col}'].fillna(df['{col}'].median())"
|
||
else:
|
||
return f"建议使用众数填充：df['{col}'].fillna(df['{col}'].mode()[0])"
|
||
|
||
def generate_report(self) -> Dict[str, Any]:
|
||
"""生成质量报告"""
|
||
# 确保质量分数在0-100之间
|
||
self.quality_score = max(0, min(100, self.quality_score))
|
||
|
||
# 按严重程度分类
|
||
high_issues = [i for i in self.issues if i.severity == "high"]
|
||
medium_issues = [i for i in self.issues if i.severity == "medium"]
|
||
low_issues = [i for i in self.issues if i.severity == "low"]
|
||
|
||
return {
|
||
"quality_score": round(self.quality_score, 2),
|
||
"total_issues": len(self.issues),
|
||
"high_severity": len(high_issues),
|
||
"medium_severity": len(medium_issues),
|
||
"low_severity": len(low_issues),
|
||
"issues": self.issues,
|
||
"summary": self._generate_summary()
|
||
}
|
||
|
||
def _generate_summary(self) -> str:
|
||
"""生成可读的摘要"""
|
||
summary = f"## 数据质量报告\n\n"
|
||
summary += f"**质量评分**: {self.quality_score:.1f}/100\n\n"
|
||
|
||
if self.quality_score >= 90:
|
||
summary += "[OK] **评级**: 优秀 - 数据质量很好\n\n"
|
||
elif self.quality_score >= 75:
|
||
summary += "[WARN] **评级**: 良好 - 存在一些小问题\n\n"
|
||
elif self.quality_score >= 60:
|
||
summary += "[WARN] **评级**: 一般 - 需要处理多个问题\n\n"
|
||
else:
|
||
summary += "[ERROR] **评级**: 差 - 数据质量问题严重\n\n"
|
||
|
||
summary += f"**问题统计**: 共 {len(self.issues)} 个质量问题\n"
|
||
summary += f"- [RED] 高严重性: {len([i for i in self.issues if i.severity == 'high'])} 个\n"
|
||
summary += f"- [YELLOW] 中严重性: {len([i for i in self.issues if i.severity == 'medium'])} 个\n"
|
||
summary += f"- [GREEN] 低严重性: {len([i for i in self.issues if i.severity == 'low'])} 个\n\n"
|
||
|
||
if self.issues:
|
||
summary += "### 主要问题:\n\n"
|
||
# 只显示高和中严重性的问题
|
||
for issue in self.issues:
|
||
if issue.severity in ["high", "medium"]:
|
||
emoji = "[RED]" if issue.severity == "high" else "[YELLOW]"
|
||
summary += f"{emoji} **{issue.column}** - {issue.description}\n"
|
||
summary += f" [TIP] {issue.suggestion}\n\n"
|
||
|
||
return summary
|
||
|
||
|
||
def quick_quality_check(df: pd.DataFrame) -> str:
|
||
"""快速数据质量检查"""
|
||
checker = DataQualityChecker(df)
|
||
report = checker.check_all()
|
||
return report['summary']
|