4. 特征分层方法论：基于信息量(IV)的分层、基于缺失率的分层、基于稳定性(PSI)的分层

在风控特征工程中，特征分层是连接特征加工与模型训练的关键桥梁。合理的分层策略能显著提升样本抽样的代表性，并降低因特征分布偏移导致的模型衰减风险。本节系统阐述三种主流分层方法：信息量(IV)分层、缺失率分层与稳定性(PSI)分层，并给出在 Spark 环境下的实现要点。

4.1 基于信息量(IV)的分层

信息量(IV, Information Value)是衡量特征对二分类目标变量区分能力的经典指标。IV 值越高，特征对好坏样本的区分度越强。基于 IV 分层，旨在将特征按预测能力划分为不同层级，用于指导特征筛选与抽样权重分配。

IV = Σ ( (Good_i / Good_total) - (Bad_i / Bad_total) ) * ln( (Good_i / Good_total) / (Bad_i / Bad_total) )

在 Spark 中，可通过 groupBy 与聚合函数实现 IV 的分布式计算。以下为基于分箱后的 IV 计算代码片段：

import org.apache.spark.sql.functions._

// 假设 df 包含字段: feature_bin, label (1=坏, 0=好)
val ivDF = df.groupBy("feature_bin")
  .agg(
    sum(when(col("label") === 0, 1).otherwise(0)).alias("good_cnt"),
    sum(when(col("label") === 1, 1).otherwise(0)).alias("bad_cnt")
  )
  .withColumn("good_dist", col("good_cnt") / sum("good_cnt").over())
  .withColumn("bad_dist", col("bad_cnt") / sum("bad_cnt").over())
  .withColumn("woe", log(col("good_dist") / col("bad_dist")))
  .withColumn("iv_contrib", (col("good_dist") - col("bad_dist")) * col("woe"))

val totalIV = ivDF.agg(sum("iv_contrib")).first().getDouble(0)

4.2 基于缺失率的分层

缺失率是衡量特征数据完整性的核心指标。高缺失率特征可能引入噪声或偏差，低缺失率特征则更可靠。基于缺失率分层，有助于在抽样时对特征质量进行差异化处理，例如对高缺失率特征进行标记或填充策略调整。

Spark 中计算特征缺失率的典型实现：

import org.apache.spark.sql.functions._

// 计算每列缺失率
val missingRateDF = df.select(
  df.columns.map(c => 
    (sum(when(col(c).isNull || col(c) === "" || col(c).isNaN, 1).otherwise(0)) / count("*")).alias(s"${c}_missing_rate")
  ): _*
)

4.3 基于稳定性(PSI)的分层

群体稳定性指标(PSI, Population Stability Index)用于衡量特征在时间维度上的分布偏移程度。PSI 越大，特征越不稳定，模型在跨时间样本上的表现越可能衰减。基于 PSI 分层，可识别出需要重点监控或重新建模的特征。

PSI = Σ ( (Actual_i / Actual_total) - (Expected_i / Expected_total) ) * ln( (Actual_i / Actual_total) / (Expected_i / Expected_total) )

Spark 中计算 PSI 的分布式实现：

import org.apache.spark.sql.functions._

// 假设 actualDF 为当前样本，expectedDF 为基准样本，均包含分箱字段 bin
val actualDist = actualDF.groupBy("bin").count().withColumnRenamed("count", "actual_cnt")
val expectedDist = expectedDF.groupBy("bin").count().withColumnRenamed("count", "expected_cnt")

val psiDF = actualDist.join(expectedDist, Seq("bin"), "outer")
  .na.fill(0)
  .withColumn("actual_pct", col("actual_cnt") / sum("actual_cnt").over())
  .withColumn("expected_pct", col("expected_cnt") / sum("expected_cnt").over())
  .withColumn("psi_contrib", (col("actual_pct") - col("expected_pct")) * log(col("actual_pct") / col("expected_pct")))

val totalPSI = psiDF.agg(sum("psi_contrib")).first().getDouble(0)

4.4 三种分层方法的综合对比