金融风控特征平台案例：实时性和可解释性要一起设计

# ❌ 坏做法：离线和线上各写一套逻辑
 
# 离线训练代码 (Spark)
def calc_txn_count_offline(user_id, date):
    """离线：自然日窗口，用清洗后宽表"""
    start = date - timedelta(days=7)
    end = date - timedelta(days=1)
    return spark.sql(f"""
        SELECT COUNT(*) as txn_count
        FROM cleaned_transaction_table
        WHERE user_id = '{user_id}'
          AND txn_date BETWEEN '{start}' AND '{end}'
    """).first()['txn_count']
 
# 线上推理代码 (Flink)
def calc_txn_count_online(user_id, current_time):
    """线上：滑动窗口，用实时流"""
    window_start = current_time - timedelta(hours=7*24)
    return flink_stream \
        .filter(lambda x: x.user_id == user_id) \
        .filter(lambda x: x.timestamp >= window_start) \
        .count()
# 问题：两套逻辑独立维护，口径偏差无法被自动发现

# ✅ 好做法：统一特征定义，自动生成离线和在线计算逻辑
 
# 特征注册中心：统一声明式定义
class TxnCountFeature(FeatureDefinition):
    """用一份定义同时生成离线和在线计算逻辑"""
    name = "user_txn_count_7d"
    description = "用户近7天交易次数"
    data_type = "INT"
    entity_key = "user_id"
 
    # 离线计算：批处理逻辑
    def batch_compute(self, context):
        return context.table("raw_transaction") \
            .filter(col("txn_time").between(
                context.point_in_time - interval("7 days"),
                context.point_in_time
            )) \
            .groupBy("user_id") \
            .agg(count("*").alias(self.name))
 
    # 在线计算：流处理逻辑
    def stream_compute(self, context):
        return context.stream("transaction_stream") \
            .key_by("user_id") \
            .window(sliding_window(size="7 days", slide="1 hour")) \
            .agg(count("*").alias(self.name))
 
    # 口径校验：离线和在线结果对比
    def consistency_check(self, offline_result, online_result):
        diff_rate = abs(offline_result - online_result) / max(offline_result, 1)
        return diff_rate < 0.05  # 允许5%以内偏差
# 好处：特征口径在一处定义，偏差有自动化检测

// ❌ 坏做法：所有实时特征走同一个 Redis 集群，无隔离
 
// 决策引擎里的特征获取
public RiskDecision evaluate(TransactionEvent event) {
    String deviceId = event.getDeviceId();
    String userId = event.getUserId();
 
    // 所有特征查同一个 Redis，热点 key 互相影响
    int deviceTxnCount = redis.get("device:txn_count:1h:" + deviceId);
    int userTxnAmount = redis.get("user:txn_amount:24h:" + userId);
    int ipTxnCount = redis.get("ip:txn_count:5m:" + event.getIp());
    // ... 还有十几个特征
 
    double riskScore = model.predict(assembleFeatures(deviceTxnCount, userTxnAmount, ipTxnCount));
    return applyRules(riskScore, event);
}
// 问题：一个热点特征拖慢整条链路，无降级方案

// ✅ 好做法：特征分级存储 + 超时降级 + 预计算缓存
 
public RiskDecision evaluate(TransactionEvent event) {
    // 1. 特征分级：核心特征走高性能存储，辅助特征走普通存储
    FeatureSet coreFeatures = coreFeatureStore.batchGet(
        List.of("user_txn_count_24h", "user_txn_amount_24h"),
        event.getEntityKeys(),
        Duration.ofMillis(10)  // 核心特征超时10ms
    );
 
    // 2. 非核心特征异步获取，设超时降级
    FeatureSet auxFeatures = auxFeatureStore.batchGetAsync(
        List.of("device_txn_count_1h", "ip_txn_count_5m"),
        event.getEntityKeys(),
        Duration.ofMillis(5)   // 辅助特征只给5ms
    ).orElse(FeatureSet.DEFAULT_VALUES);  // 超时则用默认值
 
    // 3. 预计算的高频特征走本地缓存（1分钟刷新）
    int globalFraudRateToday = localCache.get("global_fraud_rate_today");
 
    // 4. 组装特征并评分，记录每个特征的获取耗时
    RiskScore score = modelService.predict(
        coreFeatures.merge(auxFeatures).add(globalFraudRateToday),
        event.getTraceId()
    );
 
    return ruleEngine.execute(score, event);
}
// 好处：特征隔离、超时降级、全链路延迟可观测

# ❌ 坏做法：只记录最终分数，不记录决策依据
 
def make_decision(features, model, rules):
    risk_score = model.predict(features)
 
    if risk_score > 0.7:
        return Decision(status="REJECT")
    elif risk_score > 0.5:
        return Decision(status="REVIEW")
    else:
        return Decision(status="APPROVE")
 
    # 问题：没有任何解释信息，监管检查时无法回答

# ✅ 好做法：SHAP 值 + 规则命中 + 决策链路全记录
 
import shap
 
class ExplainableDecisionEngine:
    def __init__(self, model, rules, explainer=None):
        self.model = model
        self.rules = rules
        # 预计算 SHAP explainer，避免每次决策都初始化
        self.explainer = explainer or shap.TreeExplainer(model)
 
    def make_decision(self, features: dict, trace_id: str) -> Decision:
        risk_score = self.model.predict_proba([features])[0][1]
 
        # 1. 计算 SHAP 值：每个特征对预测结果的贡献
        shap_values = self.explainer.shap_values([features])[0]
        feature_contributions = self._rank_contributions(features, shap_values)
 
        # 2. 执行规则引擎，记录命中了哪些规则
        rule_hits = []
        for rule in self.rules:
            hit, reason = rule.evaluate(features)
            if hit:
                rule_hits.append(RuleHit(rule_id=rule.id, reason=reason))
 
        # 3. 综合决策
        final_status = self._determine_status(risk_score, rule_hits)
 
        # 4. 生成可读的拒绝/放行原因
        explanation = self._build_explanation(
            final_status, risk_score, feature_contributions, rule_hits
        )
 
        # 5. 全链路审计日志
        audit_log = AuditRecord(
            trace_id=trace_id,
            timestamp=datetime.utcnow(),
            status=final_status,
            risk_score=risk_score,
            top_contributors=feature_contributions[:5],  # 前5个最重要特征
            rule_hits=rule_hits,
            explanation=explanation,
            feature_snapshot=features,  # 完整特征快照，用于事后回溯
        )
        self.audit_store.save(audit_log)
 
        return Decision(status=final_status, explanation=explanation)
 
    def _rank_contributions(self, features, shap_values):
        """按 SHAP 绝对值排序，找出贡献最大的特征"""
        contributions = []
        for feat_name, shap_val in zip(features.keys(), shap_values):
            contributions.append(FeatureContribution(
                feature=feat_name,
                value=features[feat_name],
                shap_value=round(shap_val, 6),
                direction="positive" if shap_val > 0 else "negative"
            ))
        return sorted(contributions, key=lambda x: abs(x.shap_value), reverse=True)
 
    def _build_explanation(self, status, score, contributions, rule_hits):
        """生成人话版解释，给客服和监管看"""
        if status == "REJECT":
            if rule_hits:
                return f"命中规则拒绝：{rule_hits[0].reason}"
            top_reasons = [
                f"{c.feature}={c.value}（{'推高风险' if c.direction == 'positive' else '拉低风险'}）"
                for c in contributions[:3]
            ]
            return f"综合风险评分 {score:.3f}，主要因素：{'、'.join(top_reasons)}"
        return "通过"
# 好处：每笔决策都有完整的解释链路，可应对监管审查和用户咨询

# ✅ 策略编排配置示例：分层决策 + 降级兜底
 
decision_flow:
  name: "信贷申请风控决策"
  version: "2.1.0"
 
  stages:
    - name: "前置规则"
      type: "rule_engine"
      timeout_ms: 10
      rules:
        - id: "R001"
          condition: "user IN black_list"
          action: "REJECT"
          reason: "命中黑名单"
        - id: "R002"
          condition: "application_amount > 500000"
          action: "REVIEW"
          reason: "大额申请需人工审核"
 
    - name: "模型评分"
      type: "model_scoring"
      model_id: "xgb_fraud_v3"
      timeout_ms: 20
      fallback:  # 模型服务异常时的降级
        action: "USE_RULES_ONLY"
        default_score: 0.5
 
    - name: "综合决策"
      type: "decision_matrix"
      inputs: ["前置规则结果", "模型评分"]
      matrix:
        - condition: "前置规则 = REJECT"
          action: "REJECT"
          explanation_template: "命中规则：{rule_reason}"
        - condition: "模型评分 > 0.7"
          action: "REJECT"
          explanation_template: "风险评分 {score}，主要因素：{top_shap_reasons}"
        - condition: "模型评分 > 0.5 AND 模型评分 <= 0.7"
          action: "REVIEW"
          explanation_template: "风险评分 {score}，进入人工审核"
        - condition: "模型评分 <= 0.5"
          action: "APPROVE"
          explanation_template: "通过"

# ✅ 特征健康度监控示例
 
class FeatureHealthMonitor:
    """持续监控特征质量，发现异常自动告警"""
 
    def __init__(self, feature_registry, alert_manager):
        self.registry = feature_registry
        self.alerts = alert_manager
 
    def check_feature_drift(self, feature_name: str):
        """对比在线特征和离线基线的分布差异"""
        feature_def = self.registry.get(feature_name)
        online_dist = self._get_online_distribution(feature_name, window="1h")
        baseline_dist = self._get_baseline_distribution(feature_name)
 
        psi = self._calculate_psi(online_dist, baseline_dist)
 
        status = "HEALTHY"
        if psi > 0.2:
            status = "CRITICAL"
            self.alerts.send(
                level="P1",
                message=f"特征 {feature_name} PSI={psi:.4f}，严重漂移",
                action_required="立即排查，考虑模型回退"
            )
        elif psi > 0.1:
            status = "WARNING"
            self.alerts.send(
                level="P2",
                message=f"特征 {feature_name} PSI={psi:.4f}，轻微漂移",
                action_required="关注趋势，准备重训方案"
            )
 
        # 记录监控数据，用于趋势分析
        self._record_metric(feature_name, "psi", psi, status)
        return FeatureHealthReport(
            feature=feature_name,
            psi=psi,
            status=status,
            online_sample_count=len(online_dist),
            baseline_sample_count=len(baseline_dist),
        )
 
    def check_feature_completeness(self, feature_name: str):
        """监控特征缺失率和零值率"""
        recent_values = self._get_online_values(feature_name, window="5m")
        total = len(recent_values)
        missing = sum(1 for v in recent_values if v is None)
        zero_val = sum(1 for v in recent_values if v == 0)
 
        missing_rate = missing / total if total > 0 else 0
        zero_rate = zero_val / total if total > 0 else 0
 
        if missing_rate > 0.03:
            self.alerts.send(
                level="P1",
                message=f"特征 {feature_name} 缺失率 {missing_rate:.2%}",
                action_required="检查上游数据源和 Flink 任务状态"
            )
 
        return CompletenessReport(
            feature=feature_name,
            missing_rate=missing_rate,
            zero_rate=zero_rate,
            sample_count=total,
        )