3. 数据校验与异常检测
任务目标
编写一个数据校验脚本，对输入数据集执行三类检查：缺失值检测、字段类型校验、分布偏移检测（基于参考数据集），最终生成包含各项检查结果与汇总结论的HTML报告。
环境依赖
需要 pandas、numpy、scipy。若需生成PDF版报告，可额外安装 weasyprint（本案例仅生成HTML）。
核心实现
import pandas as pd
import numpy as np
from scipy.stats import ks_2samp
import json
from datetime import datetime
import html
class DataValidator:
def __init__(self, current_df, reference_df=None, numeric_threshold=0.05, cat_threshold=0.05):
"""
current_df: 待校验数据集
reference_df: 参考数据集（用于分布偏移检测），可为None
numeric_threshold: 数值列KS检验p值阈值，低于则判定偏移
cat_threshold: 分类列分布偏移阈值（最大比例差值）
"""
self.current_df = current_df
self.reference_df = reference_df
self.num_th = numeric_threshold
self.cat_th = cat_threshold
self.results = {
"timestamp": datetime.now().isoformat(),
"missing": {},
"type_errors": {},
"distribution_shift": {},
"summary": {}
}
def check_missing(self):
"""检测每列缺失值数量与比例"""
missing_counts = self.current_df.isnull().sum()
total_rows = len(self.current_df)
for col in self.current_df.columns:
count = int(missing_counts[col])
ratio = round(count / total_rows, 4) if total_rows else 0
self.results["missing"][col] = {
"count": count,
"ratio": ratio,
"status": "PASS" if ratio == 0 else "WARN" if ratio < 0.05 else "FAIL"
}
def check_types(self):
"""检查每列实际类型与声明类型是否一致"""
# 声明类型映射：实际使用时可按需修改或从外部配置读取
declared_types = {}
for col in self.current_df.columns:
# 默认推断：int64/float64视为numeric，object视为string，datetime64视为datetime
if pd.api.types.is_numeric_dtype(self.current_df[col]):
declared_types[col] = "numeric"
elif pd.api.types.is_datetime64_any_dtype(self.current_df[col]):
declared_types[col] = "datetime"
else:
declared_types[col] = "string"
for col in self.current_df.columns:
actual_series = self.current_df[col].dropna()
if len(actual_series) == 0:
self.results["type_errors"][col] = {"status": "WARN", "detail": "列全为空，无法校验类型"}
continue
declared = declared_types[col]
errors = []
if declared == "numeric":
# 检查是否能转为数值（非空值中）
coerced = pd.to_numeric(actual_series, errors='coerce')
error_mask = coerced.isnull()
if error_mask.any():
sample_vals = actual_series[error_mask].head(3).tolist()
errors = [f"非数值值: {v}" for v in sample_vals]
elif declared == "datetime":
coerced = pd.to_datetime(actual_series, errors='coerce')
error_mask = coerced.isnull()
if error_mask.any():
sample_vals = actual_series[error_mask].head(3).tolist()
errors = [f"非日期值: {v}" for v in sample_vals]
else: # string
# 所有非空值均可视为字符串，但检查是否混入明显非字符串类型
non_str_mask = actual_series.apply(lambda x: not isinstance(x, str))
if non_str_mask.any():
sample_vals = actual_series[non_str_mask].head(3).tolist()
errors = [f"非字符串值: {v} (type={type(v).__name__})" for v in sample_vals]
self.results["type_errors"][col] = {
"declared": declared,
"error_count": len(errors),
"samples": errors[:3],
"status": "PASS" if len(errors) == 0 else "FAIL"
}
def check_distribution_shift(self):
"""与参考集对比分布偏移"""
if self.reference_df is None:
self.results["distribution_shift"]["note"] = "未提供参考数据集，跳过分布偏移检测"
return
common_cols = [c for c in self.current_df.columns if c in self.reference_df.columns]
if not common_cols:
self.results["distribution_shift"]["note"] = "当前集与参考集无共同列，跳过检测"
return
for col in common_cols:
cur = self.current_df[col].dropna()
ref = self.reference_df[col].dropna()
if len(cur) == 0 or len(ref) == 0:
self.results["distribution_shift"][col] = {"status": "SKIP", "detail": "某一侧数据为空"}
continue
# 数值列用KS检验
if pd.api.types.is_numeric_dtype(cur) and pd.api.types.is_numeric_dtype(ref):
stat, p_value = ks_2samp(cur, ref)
shift = p_value < self.num_th
self.results["distribution_shift"][col] = {
"test": "ks",
"p_value": round(p_value, 6),
"status": "FAIL" if shift else "PASS"
}
# 分类列用比例差
elif cur.dtype == object or ref.dtype == object:
cur_dist = cur.value_counts(normalize=True)
ref_dist = ref.value_counts(normalize=True)
all_cats = set(cur_dist.index) | set(ref_dist.index)
max_diff = 0.0
for cat in all_cats:
diff = abs(cur_dist.get(cat, 0) - ref_dist.get(cat, 0))
max_diff = max(max_diff, diff)
shift = max_diff > self.cat_th
self.results["distribution_shift"][col] = {
"test": "category_ratio_diff",
"max_diff": round(max_diff, 4),
"status": "FAIL" if shift else "PASS"
}
else:
self.results["distribution_shift"][col] = {"status": "SKIP", "detail": "类型不匹配或无法自动检测"}
def generate_report(self, output_path="validation_report.html"):
"""生成HTML报告"""
# 汇总统计
missing_fail = sum(1 for v in self.results["missing"].values() if v["status"] == "FAIL")
type_fail = sum(1 for v in self.results["type_errors"].values() if v["status"] == "FAIL")
shift_fail = sum(1 for v in self.results["distribution_shift"].values() if isinstance(v, dict) and v.get("status") == "FAIL")
total_checks = len(self.results["missing"]) + len(self.results["type_errors"]) + \
sum(1 for v in self.results["distribution_shift"].values() if isinstance(v, dict) and v.get("status") in ("PASS", "FAIL"))
total_fail = missing_fail + type_fail + shift_fail
self.results["summary"] = {
"total_checks": total_checks,
"failed_checks": total_fail,
"overall_status": "FAIL" if total_fail > 0 else "PASS"
}
# 构建HTML
def render_table(data_dict, columns):
rows = ""
for col, info in data_dict.items():
if not isinstance(info, dict):
continue
cells = ""
for c in columns:
val = info.get(c, "")
if isinstance(val, list):
val = "; ".join(str(x) for x in val)
cells += f"<td>{html.escape(str(val))}</td>"
rows += f"<tr><td>{html.escape(str(col))}</td>{cells}</tr>"
return rows
missing_rows = render_table(self.results["missing"], ["count", "ratio", "status"])
type_rows = render_table(self.results["type_errors"], ["declared", "error_count", "samples", "status"])
shift_rows = render_table(self.results["distribution_shift"], ["test", "p_value", "max_diff", "status"])
html_content = f"""<!DOCTYPE html>
<html><head><meta charset="utf-8"><title>数据校验报告</title></head><body>
<h3>数据校验报告</h3>
<p>生成时间: {self.results['timestamp']}</p>
<p>总体状态: <strong>{self.results['summary']['overall_status']}</strong> (通过 {self.results['summary']['total_checks'] - self.results['summary']['failed_checks']}/{self.results['summary']['total_checks']} 项检查)</p>
<h4>1. 缺失值检测</h4>
<table border="1" cellpadding="4"><tr><th>列名</th><th>缺失数</th><th>缺失比例</th><th>状态</th></tr>{missing_rows}</table>
<h4>2. 类型错误检测</h4>
<table border="1" cellpadding="4"><tr><th>列名</th><th>声明类型</th><th>错误数</th><th>错误样例</th><th>状态</th></tr>{type_rows}</table>
<h4>3. 分布偏移检测</h4>
<table border="1" cellpadding="4"><tr><th>列名</th><th>检测方法</th><th>p值/最大差异</th><th>状态</th></tr>{shift_rows}</table>
</body></html>"""
with open(output_path, "w", encoding="utf-8") as f:
f.write(html_content)
return output_path
def run_all(self, output_path="validation_report.html"):
"""执行全部检查并生成报告"""
self.check_missing()
self.check_types()
self.check_distribution_shift()
report_path = self.generate_report(output_path)
return self.results, report_path
# ========== 最小可运行示例 ==========
if __name__ == "__main__":
# 构造当前数据集（含缺失、类型错误、分布偏移）
current_data = {
"age": [25, 30, None, 40, "unknown", 35, 28, 45],
"income": [50000, 60000, 55000, None, 70000, 65000, 48000, 72000],
"city": ["北京", "上海", "广州", "北京", "深圳", "上海", "北京", "广州"],
"signup_date": ["2023-01-01", "2023-02-15", "2023-03-20", "2023-04-10", "2023-05-05", "2023-06-01", "2023-07-12", "2023-08-30"]
}
current_df = pd.DataFrame(current_data)
# 将signup_date转为datetime
current_df["signup_date"] = pd.to_datetime(current_df["signup_date"])
# 构造参考数据集（分布与当前集不同）
ref_data = {
"age": [35, 40, 38, 42, 45, 39, 41, 37, 36, 43],
"income": [80000, 90000, 75000, 85000, 95000, 88000, 92000, 78000, 86000, 91000],
"city": ["北京", "北京", "上海", "上海", "广州", "北京", "深圳", "深圳", "北京", "上海"],
"signup_date": pd.date_range("2022-01-01", periods=10, freq="M")
}
reference_df = pd.DataFrame(ref_data)
validator = DataValidator(current_df, reference_df)
results, report_path = validator.run_all("validation_report.html")
print(f"报告已生成: {report_path}")
print(f"总体状态: {results['summary']['overall_status']}")运行方式
# 安装依赖
pip install pandas numpy scipy
# 运行脚本（直接执行示例）
python data_validator.py验证方法
运行后观察控制台输出与生成的 validation_report.html：
- 报告应包含三个检查区块，每列均有明确状态（PASS/WARN/FAIL/SKIP）；
- 示例数据中
age列含缺失值和字符串"unknown"，应分别触发缺失WARN和类型FAIL； income列含一个缺失值，应触发缺失WARN；city列分布与参考集不同（参考集中北京占比更高），应触发分布偏移FAIL；- 总体状态应为
FAIL（因存在FAIL项）。
判定通过条件：脚本无异常执行完毕，生成的HTML报告可在浏览器中正常打开，且上述各项检查结果与预期一致。