智能风控：原理算法与工程实践 - 梅子行

本书基于 Python 全面介绍了机器学习在信贷风控领域的应用与实践，从原理、算法与工程实践三个维度全面展开，包含 21 种实用算法和 26 个解决方案。

关于作者

梅子行 是金融科技风控领域资深专家：

风控技术专家：多年金融机构风控实战经验
技术博客作者：在知乎、公众号等平台分享风控技术文章
开源贡献者：参与风控相关开源项目
培训师：为金融机构提供风控技术培训

梅子行以其"理论与实践并重"的写作风格著称，他不仅讲解风控算法的原理，更注重工程实践中的问题和解决方案，提供了大量来自真实业务场景的案例。

经典摘录

风控的本质是在风险和收益之间找到平衡点。

反欺诈是风控的第一道防线，信用评估是第二道防线。

好的风控模型是简单、稳定、可解释的。

特征工程决定了模型效果的上限，算法只是逼近这个上限。

风控模型需要平衡准确率和召回率，而不是单纯追求某一个指标。

样本不平衡是风控建模的常态，需要特殊处理。

模型监控和迭代与模型开发同等重要。

核心内容

1. 风控体系架构

智能风控整体架构

1. 三层风控体系

   第一层：反欺诈层
   ├── 规则引擎
   ├── 反欺诈评分卡
   ├── 知识图谱
   └── 设备指纹

   第二层：信用评估层
   ├── 申请评分卡 (A 卡)
   ├── 信用模型
   └── 定额定价模型

   第三层：贷后管理层
   ├── 行为评分卡 (B 卡)
   ├── 催收评分卡 (C 卡)
   └── 预警模型

2. 风控决策流程

   申请进件 → 反欺诈 → 信用评估 → 授信决策 → 贷后监控

   具体流程：

用户申请 ↓ 基本信息校验 (年龄、身份) ↓ 反欺诈检查 (黑名单、设备指纹) ↓ 征信查询 (央行征信、百行征信) ↓ 信用评分 (A 卡模型) ↓ 决策引擎 (规则 + 模型) ↓ 授信结果 (通过/拒绝/人工审核) ↓ 定额定价 (额度、利率)

3. 决策引擎

规则类型：
- 准入规则：硬性条件，不满足直接拒绝
- 否决规则：触发直接拒绝
- 通过规则：满足直接通过
- 评分规则：模型评分阈值

规则配置：
```python
rules = {
    '准入规则': {
        'age': {'min': 18, 'max': 60},
        'city': ['北京', '上海', '广州', '深圳', ...],
        'income': {'min': 3000}
    },
    '否决规则': {
        'blacklist': True,  # 黑名单
        'fraud_count': {'gt': 0},  # 欺诈记录
        'court_executed': True  # 法院执行
    },
    '通过规则': {
        'credit_score': {'gte': 750},
        'income': {'gte': 20000},
        'house': True
    }
}

风控指标体系

资产质量指标：
- 不良率 (NPL): 不良贷款余额/总贷款余额
- 逾期率：逾期贷款余额/总贷款余额
- 核销率：核销金额/平均贷款余额
风险指标：
- 逾期率 (DPD): DPD30+, DPD60+, DPD90+
- 首逾率 (FPD): 首次还款即逾期
- 滚动率：从 M0 滚动到 M1 的比例
效率指标：
- 通过率：通过申请数/总申请数
- 放款率：放款数/通过数
- 件均：放款金额/放款数

### 2. 反欺诈技术

反欺诈核心方法

规则引擎

规则类型：

# 黑名单规则
blacklist_rules = {
    '法院失信': 'court_dishonest',
    '法院执行': 'court_executed',
    '公安通缉': 'police_wanted',
    '金融黑名单': 'finance_blacklist'
}

# 多头借贷规则
multi_platform_rules = {
    '7 天申请次数': {'threshold': 5},
    '30 天申请次数': {'threshold': 15},
    '90 天申请次数': {'threshold': 30}
}

# 异常行为规则
abnormal_rules = {
    '设备更换': 'device_change_1d',
    '地点异常': 'location_abnormal',
    '时间异常': 'apply_time_night'
}

设备指纹

设备信息收集：

device_info = {
    # 硬件信息
    'device_id': 'xxx',
    'imei': 'xxx',
    'mac': 'xxx',
    'idfa': 'xxx',

    # 软件信息
    'os': 'iOS/Android',
    'os_version': '14.0',
    'app_version': '2.1.0',

    # 网络信息
    'ip': '192.168.1.1',
    'wifi_mac': 'xxx',
    'base_station': 'xxx',

    # 行为信息
    'gyroscope': [...],
    'accelerometer': [...],
    'touch_events': [...]
}

设备关联分析：

# 一度关联：直接共用设备
# 二度关联：通过中间人关联

# 关联网络构建
import networkx as nx

G = nx.Graph()
G.add_edge('user1', 'device1')
G.add_edge('user2', 'device1')
G.add_edge('user2', 'device2')
G.add_edge('user3', 'device2')

# 查找关联用户
neighbors = list(G.neighbors('device1'))
# ['user1', 'user2']

知识图谱反欺诈

图谱构建：

# 实体类型
entities = ['用户', '设备', '手机号', 'IP 地址', '银行卡', '联系人']

# 关系类型
relations = [
    ('用户', '使用', '设备'),
    ('用户', '绑定', '手机号'),
    ('用户', '绑定', '银行卡'),
    ('用户', '联系', '联系人'),
    ('设备', '连接', 'IP 地址'),
]

# 欺诈团伙识别
# 密集子图 = 可能的欺诈团伙

团伙欺诈检测：

import networkx as nx

def detect_fraud_ring(graph, threshold=5):
    """
    检测欺诈团伙
    基于共用关系的密集子图
    """
    # 查找一度关联
    one_ring = nx.algorithms.clique.find_cliques(graph)

    # 筛选大规模团伙
    fraud_rings = [ring for ring in one_ring if len(ring) >= threshold]

    return fraud_rings

反欺诈模型

特征工程：

# 个人特征
personal_features = [
    'age', 'gender', 'education', 'marital_status',
    'city_tier', 'has_house', 'has_car'
]

# 设备特征
device_features = [
    'device_change_count_7d',
    'device_change_count_30d',
    'is_rooted',
    'is_emulator',
    'app_install_count'
]

# 行为特征
behavior_features = [
    'apply_time_hour',
    'input_speed',
    'copy_paste_count',
    'form_edit_count'
]

# 关联特征
network_features = [
    'one_degree_count',
    'two_degree_count',
    'blacklist_connection_count',
    'ring_size'
]

### 3. 信用评分模型

评分卡建模流程

数据准备

样本选择：

# 观察点：申请时间
# 表现点：观察点后 12 个月
# 标签：是否逾期 90 天以上

# 好坏客户定义
# 好客户：从未逾期或逾期\<30 天
# 坏客户：逾期>=90 天
# 灰客户：介于两者之间（可选排除）

样本清洗：

# 排除样本
exclude_conditions = [
    'age \< 18',
    'age > 60',
    'income \<= 0',
    'city is null'
]

# 时间窗口
# 训练集：2020.01-2021.12
# 测试集：2022.01-2022.06
# OOT 样本外测试

特征分箱

决策树分箱：

from sklearn.tree import DecisionTreeClassifier

def decision_tree_binning(X, y, feature, max_bins=5):
    """基于决策树的分箱"""
    dt = DecisionTreeClassifier(
        max_leaf_nodes=max_bins,
        min_samples_leaf=0.05
    )
    dt.fit(X[[feature]], y)

    # 获取分割点
    thresholds = dt.tree_.threshold[
        dt.tree_.children_left != -1
    ]
    thresholds = sorted(thresholds)

    return thresholds

卡方分箱：

def chi_merge_binning(df, feature, target, max_bins=5):
    """卡方合并分箱"""
    # 初始每个值一个箱
    # 迭代合并卡方值最小的相邻箱
    # 直到箱数\<=max_bins
    pass

WOE 与 IV

WOE 计算：

def calculate_woe_iv(df, feature, target):
    """
    计算 WOE 和 IV 值
    WOE: Weight of Evidence
    IV: Information Value
    """
    grouped = df.groupby(feature)[target].agg(['sum', 'count'])
    grouped['good'] = grouped['count'] - grouped['sum']

    total_bad = df[target].sum()
    total_good = len(df) - total_bad

    grouped['bad_pct'] = grouped['sum'] / total_bad
    grouped['good_pct'] = grouped['good'] / total_good

    # WOE = ln(good_pct / bad_pct)
    grouped['woe'] = np.log(
        grouped['good_pct'] / grouped['bad_pct']
    )

    # IV = Σ(good_pct - bad_pct) * WOE
    iv = np.sum(
        (grouped['good_pct'] - grouped['bad_pct']) *
        grouped['woe']
    )

    return grouped['woe'].to_dict(), iv

IV 值解读：

IV < 0.02: 无预测力
0.02 <= IV < 0.1: 较弱
0.1 <= IV < 0.3: 中等
0.3 <= IV < 0.5: 较强
IV >= 0.5: 过强 (可能有问题)

逻辑回归建模

from sklearn.linear_model import LogisticRegression

# WOE 编码后的数据
X_woe = transform_to_woe(X, woe_mappings)

# 训练模型
model = LogisticRegression(
    C=1.0,
    penalty='l2',
    class_weight='balanced',
    max_iter=1000
)
model.fit(X_woe, y)

# 转换为评分卡
# 分数 = 基准分 + Σ(特征 WOE × 系数 × 刻度)
base_score = 600
pdo = 50  # 几率翻倍时的分数变化
factor = pdo / np.log(2)
offset = base_score - factor * np.log(20)  # 基准几率 20:1

# 各特征得分
def calculate_score(x_woe, coefficients):
    scores = {}
    for feat, woe in x_woe.items():
        scores[feat] = woe * coefficients[feat] * factor
    total_score = offset + sum(scores.values())
    return total_score

### 4. 机器学习算法

风控常用算法

XGBoost 实战

import xgboost as xgb
from sklearn.model_selection import cross_val_score

# 处理样本不平衡
scale_pos_weight = (y == 0).sum() / (y == 1).sum()

params = {
    'max_depth': 5,           # 树的最大深度
    'learning_rate': 0.01,    # 学习率
    'n_estimators': 1000,     # 树的数量
    'scale_pos_weight': scale_pos_weight,  # 正负样本平衡
    'subsample': 0.8,         # 样本采样
    'colsample_bytree': 0.8,  # 特征采样
    'lambda': 1,              # L2 正则
    'alpha': 0,               # L1 正则
    'eval_metric': ['auc', 'logloss']
}

model = xgb.XGBClassifier(**params)

# 早停
model.fit(
    X_train, y_train,
    eval_set=[(X_valid, y_valid)],
    early_stopping_rounds=50,
    verbose=True
)

LightGBM 实战

import lightgbm as lgb

params = {
    'objective': 'binary',
    'metric': 'auc',
    'boosting_type': 'gbdt',
    'num_leaves': 31,
    'learning_rate': 0.01,
    'feature_fraction': 0.8,
    'bagging_fraction': 0.8,
    'bagging_freq': 5,
    'scale_pos_weight': scale_pos_weight,
    'min_child_samples': 100,
    'reg_alpha': 0.1,
    'reg_lambda': 1
}

train_data = lgb.Dataset(X_train, label=y_train)
valid_data = lgb.Dataset(X_valid, label=y_valid)

model = lgb.train(
    params,
    train_data,
    num_boost_round=1000,
    valid_sets=[train_data, valid_data],
    early_stopping_rounds=50,
    verbose_eval=50
)

模型融合

Stacking:

from sklearn.ensemble import StackingClassifier

# 基模型
base_models = [
    ('xgb', xgb.XGBClassifier(**xgb_params)),
    ('lgb', lgb.LGBMClassifier(**lgb_params)),
    ('rf', RandomForestClassifier())
]

# 元模型
meta_model = LogisticRegression()

# Stacking
stacking = StackingClassifier(
    estimators=base_models,
    final_estimator=meta_model,
    cv=5
)
stacking.fit(X_train, y_train)

Voting:

from sklearn.ensemble import VotingClassifier

voting = VotingClassifier(
    estimators=[
        ('xgb', xgb_model),
        ('lgb', lgb_model),
        ('lr', lr_model)
    ],
    voting='soft'  # 概率投票
)
voting.fit(X_train, y_train)

### 5. 模型监控与迭代

模型监控体系

稳定性监控

PSI 计算：

def calculate_psi(expected, actual, bins=10):
    """
    计算群体稳定性指标 PSI
    PSI \< 0.1: 稳定
    0.1 \<= PSI \< 0.25: 轻微变化
    PSI >= 0.25: 显著变化
    """
    # 分箱
    bins_expected = np.histogram(expected, bins=bins, density=True)[0]
    bins_actual = np.histogram(actual, bins=bins, density=True)[0]

    # 避免除零
    bins_actual = np.where(bins_actual == 0, 0.0001, bins_actual)
    bins_expected = np.where(bins_expected == 0, 0.0001, bins_expected)

    psi = np.sum((bins_actual - bins_expected) *
                 np.log(bins_actual / bins_expected))
    return psi

特征稳定性监控：

# 每日计算各特征的 PSI
feature_psi = {}
for feature in feature_cols:
    psi = calculate_psi(
        train_data[feature],
        online_data[feature]
    )
    feature_psi[feature] = psi

# 告警
unstable_features = [
    f for f, psi in feature_psi.items()
    if psi > 0.1
]

性能监控

区分度监控：

def daily_ks_auc(df, date_col, score_col, target_col):
    """每日 KS 和 AUC 追踪"""
    daily_metrics = df.groupby(date_col).apply(
        lambda x: {
            'ks': calculate_ks(x[target_col], x[score_col]),
            'auc': roc_auc_score(x[target_col], x[score_col])
        }
    )
    return daily_metrics

准确率监控：

# 不同分数段的逾期率
def score_bucket_analysis(df, score_col, target_col, n_buckets=10):
    df['score_bucket'] = pd.qcut(df[score_col], q=n_buckets)
    bucket_report = df.groupby('score_bucket')[target_col].agg([
        'count', 'sum', 'mean'
    ])
    bucket_report.columns = ['count', 'bad', 'bad_rate']
    return bucket_report

模型迭代

迭代触发条件：

PSI > 0.25: 客群显著变化
KS 下降>20%: 区分度下降
业务规则变化
新产品上线

迭代策略：

小迭代 (月度):
├── 更新特征 WOE 映射
├── 调整阈值
└── 规则优化

中迭代 (季度):
├── 模型重训练
├── 特征工程更新
└── 参数调优

大迭代 (年度):
├── 重新建模
├── 新数据源引入
└── 架构升级

### 6. 工程实践

风控系统工程

实时决策系统

架构设计：

请求入口 (API Gateway)
    ↓
特征计算引擎
├── 实时特征 (设备、位置)
├── 缓存特征 (用户画像)
└── 离线特征 (征信数据)
    ↓
规则引擎
├── 准入规则
├── 反欺诈规则
└── 信用规则
    ↓
模型引擎
├── 反欺诈模型
└── 信用评分模型
    ↓
决策引擎
├── 决策树
└── 结果输出

性能要求：

响应时间：< 200ms
可用性：99.99%
QPS: 1000+

特征平台

特征存储：

# Redis 缓存实时特征
import redis

r = redis.Redis()

# 存储用户特征
def store_user_features(user_id, features):
    key = f'user_features:{user_id}'
    r.hmset(key, features)
    r.expire(key, 3600)  # 1 小时过期

# 获取用户特征
def get_user_features(user_id):
    key = f'user_features:{user_id}'
    return r.hgetall(key)

特征计算：

# 实时特征计算
def calculate_features(user_id, event):
    features = {}

    # 7 天内申请次数
    features['apply_count_7d'] = get_apply_count(
        user_id, days=7
    )

    # 设备更换次数
    features['device_change_30d'] = get_device_change_count(
        user_id, days=30
    )

    # 多头借贷指数
    features['multi_platform_index'] = calculate_multi_platform(
        user_id
    )

    return features

模型部署

模型服务化：

from fastapi import FastAPI
import joblib

app = FastAPI()
model = joblib.load('model.pkl')

@app.post('/predict')
async def predict(data: RequestData):
    # 特征处理
    features = await extract_features(data)

    # 模型预测
    score = model.predict_proba([features])[0, 1]

    # 决策
    result = 'approve' if score \< 0.3 else 'reject'

    return {
        'score': float(score),
        'result': result
    }

## 读书心得

《智能风控：原理 算法与工程实践》是一本全面讲解风控技术的书籍。作者从原理、算法、工程三个维度展开，既有理论深度，又有实践价值。

**风控体系架构**部分让我对整体框架有了清晰认识。三层风控体系——反欺诈层、信用评估层、贷后管理层，各司其职又相互配合。决策引擎将规则和模型有机组合，形成完整的决策流程。

**反欺诈技术**是本书的亮点。规则引擎、设备指纹、知识图谱，这些技术构成了立体的反欺诈防线。特别是知识图谱在团伙欺诈检测中的应用，通过关联分析可以发现隐蔽的欺诈网络。

**评分卡建模**流程讲解详细。从样本选择、特征分箱、WOE 编码到逻辑回归建模，每一步都有清晰的代码实现。评分卡的优势在于可解释性强，符合监管要求。

**模型监控**部分很实用。PSI 监控稳定性，KS/AUC 监控区分度，特征分布监控数据质量。模型上线只是开始，持续监控和迭代才能保证长期效果。

对于从事风控工作的同行，这本书提供了：
- 完整的知识体系
- 实用的算法代码
- 工程实践的经验
- 行业最佳实践

推荐给所有对智能风控感兴趣的同行阅读。

关于作者​

经典摘录​

核心内容​

1. 风控体系架构​

关于作者

经典摘录

核心内容

1. 风控体系架构