import pandas as pd
import numpy as np
# 1. 构造包含脏数据的示范数据集 (模拟 UCI Adult / Titanic 风格)
data = {
'age': [22, 35, 28, 42, 150, 29, 35, np.nan], # 包含异常值 150 和缺失值
'income': [5000, 8000, np.nan, 12000, 9500, 8000, 8000, 6000],
'education': ['Bachelors', 'HS-grad', 'Bachelors', 'Masters', 'HS-grad', 'HS-grad', 'HS-grad', 'Doctorate'],
'target': [0, 0, 0, 1, 1, 0, 0, 1]
}
df = pd.DataFrame(data)
# 2. 自动化质量审计函数
def audit_dataset(df: pd.DataFrame) -> pd.DataFrame:
audit_rows = []
for col in df.columns:
col_data = df[col]
dtype = col_data.dtype
null_count = col_data.isnull().sum()
null_pct = (null_count / len(df)) * 100
# 异常值检测 (针对数值列)
outliers_count = 0
if np.issubdtype(dtype, np.number):
q1 = col_data.quantile(0.25)
q3 = col_data.quantile(0.75)
iqr = q3 - q1
upper = q3 + 1.5 * iqr
lower = q1 - 1.5 * iqr
outliers_count = ((col_data < lower) | (col_data > upper)).sum()
audit_rows.append({
'字段名': col,
'数据类型': str(dtype),
'缺失记录数': null_count,
'缺失占比(%)': f"{null_pct:.1f}%",
'IQR异常值数': outliers_count,
'唯一值个数': col_data.nunique()
})
return pd.DataFrame(audit_rows)
report_df = audit_dataset(df)
duplicates = df.duplicated().sum()
print("================ 数据质量审计报告 (Data Audit) ================")
print(report_df.to_string(index=False))
print("---------------------------------------------------------------")
print(f"全表总行数: {len(df)} | 重复记录行数: {duplicates}")