4. 数据准备与清洗：风险数据源梳理、数据质量检查、缺失值与异常值处理、数据标准化

数据准备与清洗是组合风险归因分析中最耗时、最关键的环节。原始风险数据通常来自多个异构系统，包含大量噪声、缺失值和格式不一致问题。本节将系统梳理风险数据源、建立数据质量检查框架，并给出缺失值、异常值处理及标准化的实战方法。

4.1 风险数据源梳理

组合风险归因所需的数据源通常涵盖以下四类：

4.2 数据质量检查框架

数据质量检查应覆盖完整性、准确性、一致性和时效性四个维度。以下为推荐的质量检查清单：

• 完整性检查：检查关键字段是否存在空值，时间序列是否存在断点，证券代码是否在参考表中存在。
• 准确性检查：验证数值字段是否在合理范围内（如持仓数量 > 0，价格 > 0），计算逻辑是否与源系统一致（如市值 = 持仓数量 × 收盘价）。
• 一致性检查：跨系统比对同一字段（如交易系统与托管行的持仓市值差异 < 0.1%），检查行业分类代码是否与标准分类一致。
• 时效性检查：确认数据日期是否为最新交易日，延迟数据是否已标记，历史数据回溯是否完整。

4.3 缺失值与异常值处理

4.3.1 缺失值处理策略

风险数据中的缺失值通常出现在行情数据（停牌、节假日）、因子暴露（新上市证券）或持仓数据（数据延迟）中。根据缺失机制和业务场景，推荐以下处理策略：

4.3.2 异常值检测与处理

异常值在风险数据中常见于价格错误、因子暴露极端值或持仓数据录入错误。推荐使用以下方法进行检测：

• Z-Score 方法：计算每个字段的 Z-Score，绝对值大于 3 的标记为异常值。适用于近似正态分布的数据。
• IQR 方法：计算四分位距（IQR），超出 Q1 - 1.5×IQR 或 Q3 + 1.5×IQR 的标记为异常值。适用于偏态分布。
• 业务规则法：根据业务常识设定硬阈值，如股票价格不能为负、持仓数量不能超过总股本等。
• 时间序列方法：对日频数据使用滚动窗口的均值和标准差，检测突变点。

异常值处理策略：

• 修正：如果能确认正确值（如人工核查），直接替换。
• 截尾：将异常值替换为阈值上限或下限（Winsorization）。
• 剔除：在分析中排除该记录，但需记录剔除原因和数量。

4.4 数据标准化

数据标准化是确保不同量纲、不同分布的数据能够进行统一分析的关键步骤。在组合风险归因中，主要涉及以下标准化操作：

4.5 数据清洗实战代码示例

以下 Python 代码展示了数据清洗的核心流程，包括缺失值处理、异常值检测和标准化：

import pandas as pd
import numpy as np
from scipy import stats

# 加载原始风险数据
df = pd.read_csv('risk_data_raw.csv', parse_dates=['date'])

# ========== 1. 缺失值处理 ==========
# 前向填充行情数据（最多连续填充3天）
df['close_price'] = df.groupby('security_id')['close_price'] \
    .transform(lambda x: x.fillna(method='ffill', limit=3))

# 行业均值填充因子暴露
df['factor_exposure'] = df.groupby('industry')['factor_exposure'] \
    .transform(lambda x: x.fillna(x.mean()))

# 标记仍为缺失的记录
df['is_imputed'] = df['factor_exposure'].isna().astype(int)

# ========== 2. 异常值检测 ==========
# Z-Score 方法检测价格异常
z_scores = np.abs(stats.zscore(df['close_price'].dropna()))
df.loc[df['close_price'].notna(), 'price_outlier'] = (z_scores > 3).astype(int)

# IQR 方法检测持仓异常
Q1 = df['holding_value'].quantile(0.25)
Q3 = df['holding_value'].quantile(0.75)
IQR = Q3 - Q1
df['holding_outlier'] = ((df['holding_value'] < Q1 - 1.5*IQR) | 
                         (df['holding_value'] > Q3 + 1.5*IQR)).astype(int)

# 截尾处理异常持仓
upper_limit = Q3 + 1.5*IQR
lower_limit = Q1 - 1.5*IQR
df['holding_value_winsorized'] = df['holding_value'].clip(lower_limit, upper_limit)

# ========== 3. 数据标准化 ==========
# Z-Score 标准化因子暴露
df['factor_exposure_z'] = (df['factor_exposure'] - df['factor_exposure'].mean()) / \
                           df['factor_exposure'].std()

# Min-Max 标准化风险贡献
df['risk_contribution_norm'] = (df['risk_contribution'] - df['risk_contribution'].min()) / \
                                (df['risk_contribution'].max() - df['risk_contribution'].min())

# 输出清洗后的数据质量报告
quality_report = {
    'total_records': len(df),
    'missing_after_fill': df['factor_exposure'].isna().sum(),
    'price_outliers': df['price_outlier'].sum(),
    'holding_outliers': df['holding_outlier'].sum(),
    'standardized_fields': ['factor_exposure_z', 'risk_contribution_norm']
}
print("数据质量报告:", quality_report)