4. 数据质量检查：缺失值、异常值、重复值检测与可视化报告

在金融特征回放评审中，数据质量直接决定后续建模与回测的可靠性。缺失值、异常值与重复值是三类最常见的“脏数据”来源。本章构建一套基于 Python 的自动化检测与可视化报告工具链，帮助分析师快速定位问题、量化影响并生成可复用的质量报告。

4.1 三类质量问题的定义与金融影响

4.2 检测流程架构

下图展示了从原始特征表到质量报告的核心检测流水线。每个阶段均输出中间统计量，支持逐层下钻排查。

4.3 核心检测函数实现

以下代码封装了三个独立的检测模块，每个模块返回统计 DataFrame，便于后续合并与可视化。

import pandas as pd
import numpy as np

def detect_missing(df, threshold=0.0):
    """检测缺失值，返回缺失比率 > threshold 的列"""
    missing_ratio = df.isnull().mean()
    missing_df = missing_ratio[missing_ratio > threshold].reset_index()
    missing_df.columns = ['column', 'missing_ratio']
    missing_df['missing_count'] = (df.isnull().sum()[missing_df['column']]).values
    return missing_df.sort_values('missing_ratio', ascending=False)

def detect_outliers_iqr(df, columns, multiplier=1.5):
    """基于 IQR 检测异常值，返回每列异常记录数"""
    result = []
    for col in columns:
        if df[col].dtype not in ['float64', 'int64']:
            continue
        Q1 = df[col].quantile(0.25)
        Q3 = df[col].quantile(0.75)
        IQR = Q3 - Q1
        lower = Q1 - multiplier * IQR
        upper = Q3 + multiplier * IQR
        outlier_mask = (df[col] < lower) | (df[col] > upper)
        result.append({
            'column': col,
            'outlier_count': outlier_mask.sum(),
            'outlier_ratio': outlier_mask.mean(),
            'lower_bound': lower,
            'upper_bound': upper
        })
    return pd.DataFrame(result)

def detect_duplicates(df, subset=None):
    """检测完全重复或指定子集重复"""
    total_dup = df.duplicated(subset=subset, keep=False).sum()
    first_dup = df.duplicated(subset=subset, keep='first').sum()
    return {
        'total_duplicate_rows': total_dup,
        'first_occurrence_removed': first_dup,
        'duplicate_groups': df.duplicated(subset=subset, keep=False).sum()
    }

4.4 可视化报告生成

将检测结果整合为一张综合报告图，包含缺失热力图、异常值箱线图与重复记录分布。以下 SVG 展示报告模板的布局结构。

4.5 一键生成综合报告

将检测与可视化合并为 generate_quality_report() 函数，输出包含表格与图表的 HTML 报告。

def generate_quality_report(df, numeric_cols, output_path='quality_report.html'):
    """生成数据质量综合报告"""
    # 1. 执行检测
    missing_df = detect_missing(df, threshold=0.0)
    outlier_df = detect_outliers_iqr(df, numeric_cols)
    dup_info = detect_duplicates(df)

    # 2. 构建 HTML 片段
    html_parts = []
    html_parts.append("<h2>数据质量报告</h2>")

    # 缺失值表格
    html_parts.append("<h3>缺失值统计</h3>")
    html_parts.append(missing_df.to_html(classes='table table-striped'))

    # 异常值表格
    html_parts.append("<h3>异常值统计 (IQR)</h3>")
    html_parts.append(outlier_df.to_html(classes='table table-striped'))

    # 重复值摘要
    html_parts.append("<h3>重复值摘要</h3>")
    html_parts.append(f"<p>总重复行数: {dup_info['total_duplicate_rows']}</p>")
    html_parts.append(f"<p>去重后保留首行: {dup_info['first_occurrence_removed']}</p>")

    # 3. 写入文件
    with open(output_path, 'w', encoding='utf-8') as f:
        f.write('\n'.join(html_parts))
    print(f"报告已生成: {output_path}")
    return output_path

4.6 质量检查清单

• 缺失值：检查每列缺失比率；对时序数据检查缺失是否连续；对关键特征（如收盘价）要求缺失率 < 1%。
• 异常值：使用 IQR 或 Z-Score 标记；对金融数据建议使用 3σ 或 5σ 阈值；检查异常值是否集中在特定时间段。
• 重复值：按主键（时间 + 代码）去重；记录重复组数；检查是否因数据源合并导致重复。
• 报告输出：每次回放前自动生成质量报告；将报告纳入特征版本管理；设置质量阈值自动阻断流水线。