第3章：数据预处理与质量控制

做了这么多年的参数调优，我得说一句大实话：数据预处理比模型调参重要十倍。
很多同学一上来就盯着 learning rate、batch size 猛调，结果模型死活不收敛。
回头一看，数据里有空值、有异常点、两个批次的数据分布根本不搭界。
说白了，脏数据进去，再好的算法也白搭。

我个人习惯，拿到原始数据后的第一件事不是跑模型，而是先做一遍彻底的质量检查。
这一步会帮我省下后面至少一半的 debug 时间。你想想看，批次效应校准本身就是在消除技术噪声，
要是输入数据本身就有质量问题，那校准出来的结果你敢信吗？

3.1 数据检查：先摸清家底

数据预处理的第一步，是搞清楚你手上东西到底什么样。
我会先检查下面几项：

• 缺失值比例：每个特征缺失了多少？超过 30% 的特征我一般会直接砍掉。
• 重复样本：同一个样本在多个批次里出现了吗？这在实践里很常见。
• 数据类型：该是数值的列是不是被读成了字符串？该是分类的列是不是被当成了连续值？
• 数值范围：不同批次的同一个特征，极差和方差是不是差了几个数量级？

我在项目中遇到过最坑的一次，是两个批次的表达量数据量级差了 100 倍——
一个是 counts 值，另一个是 log-transformed 值，放在一起直接做批次校准，结果一塌糊涂。

3.2 缺失值处理：别急着删

很多人碰到缺失值就顺手删掉一行。
但这样做，如果样本量本身不大，损失的信息可能比你想的更多。

我常用的策略是按缺失比例分级处理：

1. 比例 < 5%：用均值或中位数填充。
   连续值用中位数（鲁棒性好），分类值用众数。
2. 比例 5% - 20%：用 KNN 插补或更高级的预测模型。
   利用其他特征来估计缺失值，效果比简单填充好很多。
3. 比例 20% - 30%：建立缺失指示变量，比如新增一列 is_missing。
   这样模型可以自己学会缺失模式。
4. 比例 > 30%：直接删除该特征。不值得浪费模型容量。

from sklearn.impute import SimpleImputer
import numpy as np

# 用一个统一的 imputer 拟合所有批次
imputer = SimpleImputer(strategy='median')
X_all = np.vstack([X_batch1, X_batch2, X_batch3])
imputer.fit(X_all)

# 分别对每个批次做 transform
X_b1_clean = imputer.transform(X_batch1)
X_b2_clean = imputer.transform(X_batch2)

3.3 异常值检测：别一刀切

异常值在批次校准里是个烫手山芋。
有些异常值确实是测量错误，但有些反而是真实的生物学变异。
我个人习惯分两步走：

• 第一步：用 Z-score 或 IQR 方法标记潜在异常点。Z-score 绝对值超过 3 的，我会打个标签。
• 第二步：不是直接删除，而是看这个异常点在所有批次里是否一致出现。
  如果只在某个批次里异常，那可能是批次效应；如果在所有批次里都异常，那可能是真实的。

异常值的检测代码，我一般会写一个辅助函数：

def flag_outliers(data, method='zscore', threshold=3):
    if method == 'zscore':
        z = np.abs((data - data.mean()) / data.std())
        return z > threshold
    elif method == 'iqr':
        q1, q3 = np.percentile(data, [25, 75])
        iqr = q3 - q1
        lower = q1 - 1.5 * iqr
        upper = q3 + 1.5 * iqr
        return (data < lower) | (data > upper)

# 用法示例
outlier_mask = flag_outliers(X_batch1[:, 0], method='zscore')
print(f'Batch 1 特征0 异常值数量：{outlier_mask.sum()}')

3.4 数据标准化：统一尺度

批次校准的算法，比如 ComBat 或 Harmony，对数据的尺度非常敏感。
不同批次的数据可能使用不同的测量平台或者不同的实验室操作流程，
导致数据分布存在整体偏移或缩放。

我一般会先做一步简单的标准化，让所有批次的特征都落到差不多的范围里：

我个人更倾向用 RobustScaler。为什么？因为它不受异常值的影响。
一个极端的离群点可能会把均值和标准差拉偏好几个数量级，但中位数和 IQR 几乎纹丝不动。

from sklearn.preprocessing import RobustScaler

# 同样，用一个统一的 scaler 拟合所有批次
scaler = RobustScaler()
scaler.fit(X_all)

# 对每个批次分别做 transform
X_b1_scaled = scaler.transform(X_b1_clean)
X_b2_scaled = scaler.transform(X_b2_clean)

3.5 质量控制评分：量化你的数据质量

光凭肉眼判断数据好不好，太主观了。
我习惯在预处理结束后，给数据算一个质量评分。这样后面跑参数调优时，
我可以明确知道：数据质量从 0.6 提升到 0.85 之后，校准效果提高了多少。

实现起来不复杂：

from scipy.stats import ks_2samp

def quality_score(X_batches, batch_labels):
    total_n = sum(len(b) for b in X_batches)
    missing_rate = np.isnan(np.vstack(X_batches)).sum() / (total_n * X_batches[0].shape[1])
    
    outlier_count = 0
    for b in X_batches:
        z = np.abs((b - b.mean(axis=0)) / b.std(axis=0))
        outlier_count += (z > 3).sum()
    outlier_rate = outlier_count / (total_n * X_batches[0].shape[1])
    
    # 批次间分布差异：取所有两两批次 KS 检验 p 值的均值
    ks_pvals = []
    for i in range(len(X_batches)):
        for j in range(i+1, len(X_batches)):
            for k in range(X_batches[i].shape[1]):
                stat, p = ks_2samp(X_batches[i][:, k], X_batches[j][:, k])
                ks_pvals.append(p)
    batch_diff = 1 - np.mean(ks_pvals)
    
    Q = 1 - (missing_rate * 0.3 + outlier_rate * 0.3 + batch_diff * 0.4)
    return Q

score = quality_score([X_b1_clean, X_b2_clean], ['batch1', 'batch2'])
print(f'数据质量评分：{score:.3f}')

3.6 避坑指南：我踩过的几个坑

做数据预处理这么多年，有些坑我反复踩过。今天一并写出来，帮你们省点时间。

• 别把批次变量当作普通特征标准化：标准化时不要把 batch ID 或者其他元数据一起处理。这些是分类变量，标准化以后毫无意义。
• 小心数据泄露：填充缺失值、标准化时，一定要用所有批次联合拟合，再分别转换。如果每个批次各自拟合，相当于人为放大了批次差异——我在早期项目里就吃过这个亏。
• 不要过度清洗：记住数据预处理的目的是保留真实信号，去除技术噪声。如果你为了追求“干净”把 20% 的样本都删了，那你的结论基本可以扔了。
• 日志记录很重要：每一次处理，都记录你做了什么、参数是什么、删除了多少样本。我曾在一次调优复盘时，因为没做记录，花了整整两天才弄清楚数据被怎么处理过。

做完这整套流程，你的数据基本就准备好了。
下一步就是把干净的数据送进 ComBat 或者 Harmony 做批次校准。
记住一句话：预处理做好了，校准效果至少提升 30%。