未记录事实核验 · 状态：维护中
记录仅说明核验时间与维护状态，不保证适用于所有环境；使用前请留意正文中的版本、适用条件与风险提示。
warning 适用边界与维护说明
本文适合已有Python、概率统计和机器学习基础的读者学习分类算法推导、手工实现与合成数据建模流程；本次仅作全文结构审阅，未运行代码或进行外部事实核验。不平衡策略比较仍在数据切分前执行SMOTE，且未使用共同的原始留出集；客户流失流程的特征工程使用全量月消费均值，并复用测试集比较模型、选模和计算情景收益，因此相关输出不能作为公平策略比较或独立泛化评估。客户流失模型将名义类别编码与连续特征一同交给普通SMOTE，插值结果可能失去类别语义；默认SMOTE邻居参数要求每个被过采样类别至少有6个样本。手工随机森林训练要求非负整数标签，ROC/PR示例按0/1标签及概率第二列对应正类编写，决策边界绘图限于二维特征，其中SVM散点着色还依赖前半类别0、后半类别1的排列。客户流失关系由代码预设，收益仅基于客户价值1000、挽留成本100和成功率0.3等假设，重采样后的概率及风险阈值不能直接用于业务决策。监控示例未交付建模所用scaler，也未保存真实训练分布基线，仅可作为接口思路，不是可直接部署的完整系统；算法选型表应作为候选方案提示，而非仅凭样本量即可确定最优算法的规则。
分类算法深度解析
abstract 本章概览
分类算法是监督学习的核心任务之一，用于预测离散类别标签。本文深入解析主流分类算法的数学原理、技术实现和优化策略。
核心内容：
- 分类算法的数学基础与决策边界理论
- 逻辑回归的概率估计与梯度下降优化
- 决策树的信息增益与剪枝策略
- 集成学习：Bagging与Boosting机制
- 分类性能评估：混淆矩阵、ROC曲线与AUC指标
- 类别不平衡问题的解决方案
- 基于合成数据的客户流失预测项目示例
分类算法的数学基础
决策边界理论
分类算法的核心是在特征空间中构建决策边界，将不同类别的样本分离。对于d维特征空间中的二分类问题，决策边界可表示为：
$$h(\mathbf{x}) = \begin{cases}
1, & \text{if } f(\mathbf{x}) \geq \theta \\
0, & \text{if } f(\mathbf{x}) < \theta
\end{cases}$$其中$f(\mathbf{x})$是判别函数，$\theta$是决策阈值。
概率估计与贝叶斯决策
分类问题可转化为条件概率估计：
$$P(y=k|\mathbf{x}) = \frac{P(\mathbf{x}|y=k)P(y=k)}{P(\mathbf{x})}$$最优贝叶斯分类器选择后验概率最大的类别：
$$\hat{y} = \arg\max_k P(y=k|\mathbf{x})$$分类问题类型
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.linear_model import LogisticRegression
from sklearn.tree import DecisionTreeClassifier
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.svm import SVC
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.metrics import (
accuracy_score, precision_score, recall_score, f1_score,
confusion_matrix, classification_report, roc_curve, auc,
precision_recall_curve
)
from sklearn.preprocessing import StandardScaler
from sklearn.utils.class_weight import compute_class_weight
from imblearn.over_sampling import SMOTE
# 分类问题的数学表示
def classification_fundamentals():
"""分类问题基础理论"""
print("分类问题类型:")
print("1. 二分类: y ∈ {0, 1}")
print("2. 多分类: y ∈ {1, 2, ..., K}")
print("3. 多标签: y ∈ {0, 1}^K")
print("\n核心算法特性:")
print("• 线性分类器: 线性决策边界")
print("• 非线性分类器: 非线性决策边界")
print("• 概率分类器: 输出类别概率")
print("• 判别模型: 直接建模P(y|x)")
print("• 生成模型: 建模P(x|y)和P(y)")
逻辑回归：线性分类的数学基础
Sigmoid函数与最大似然估计
逻辑回归通过Sigmoid函数将线性组合映射到概率值：
$$\sigma(z) = \frac{1}{1 + e^{-z}}$$其中$z = \mathbf{w}^T\mathbf{x} + b$是线性决策函数。
参数估计通过最大似然估计（MLE）：
$$L(\mathbf{w}) = \prod_{i=1}^{n} [\sigma(\mathbf{w}^T\mathbf{x}_i)]^{y_i} [1-\sigma(\mathbf{w}^T\mathbf{x}_i)]^{1-y_i}$$对数似然函数为：
$$\ell(\mathbf{w}) = \sum_{i=1}^{n} [y_i \log(\sigma(\mathbf{w}^T\mathbf{x}_i)) + (1-y_i) \log(1-\sigma(\mathbf{w}^T\mathbf{x}_i))]$$def logistic_regression_theory():
"""逻辑回归理论与实现"""
# 生成线性可分的二分类数据
np.random.seed(42)
n_samples = 1000
# 特征生成
X = np.random.randn(n_samples, 2)
# 真实决策边界: w1*x1 + w2*x2 + b = 0
true_w = np.array([1.5, -2.0])
true_b = 0.5
# 线性组合
linear_combination = X @ true_w + true_b
# 添加噪声
noise = np.random.normal(0, 0.2, n_samples)
y_prob = 1 / (1 + np.exp(-(linear_combination + noise)))
y = (y_prob > 0.5).astype(int)
return X, y, true_w, true_b
梯度下降优化算法
逻辑回归可采用梯度上升最大化前文定义的对数似然ℓ，或等价地采用梯度下降最小化负对数似然；以下公式采用梯度上升，后续代码采用负对数似然的梯度下降：
$$\mathbf{w}_{t+1} = \mathbf{w}_t + \alpha \nabla_{\mathbf{w}} \ell(\mathbf{w}_t)$$梯度公式：
$$\frac{\partial \ell}{\partial \mathbf{w}} = \sum_{i=1}^{n} (y_i - \sigma(\mathbf{w}^T\mathbf{x}_i)) \mathbf{x}_i$$def train_logistic_regression(X, y, learning_rate=0.01, max_iter=1000):
"""手工实现逻辑回归训练"""
def sigmoid(z):
# 稳定sigmoid函数，避免数值上溢
exp_neg_abs = np.exp(-np.abs(z))
return np.where(z >= 0,
1 / (1 + exp_neg_abs),
exp_neg_abs / (1 + exp_neg_abs))
def log_likelihood(w, b, X, y):
z = X @ w + b
return np.sum(y * z - np.logaddexp(0, z))
n_samples, n_features = X.shape
w = np.zeros(n_features)
b = 0.0
losses = []
for i in range(max_iter):
# 前向传播
z = X @ w + b
predictions = sigmoid(z)
# 计算损失
loss = -log_likelihood(w, b, X, y) / n_samples
losses.append(loss)
# 计算梯度
dw = (X.T @ (predictions - y)) / n_samples
db = np.mean(predictions - y)
# 更新参数
w -= learning_rate * dw
b -= learning_rate * db
if i % 100 == 0:
print(f"Iteration {i}, Loss: {loss:.4f}")
return w, b, losses
# sklearn实现对比
def sklearn_logistic_regression(X, y):
"""使用sklearn训练逻辑回归"""
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=42, stratify=y
)
# 标准化特征
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)
# 训练模型
model = LogisticRegression(random_state=42, max_iter=1000)
model.fit(X_train_scaled, y_train)
# 预测与评估
y_pred_proba = model.predict_proba(X_test_scaled)
y_pred = model.predict(X_test_scaled)
print(f"权重系数: {model.coef_[0]}")
print(f"偏置项: {model.intercept_[0]:.3f}")
print(f"测试准确率: {accuracy_score(y_test, y_pred):.3f}")
return model, scaler, (X_test_scaled, y_test, y_pred, y_pred_proba)
决策树：信息理论与递归分割
信息增益与熵的计算
决策树通过递归分割构建，每次选择最优特征和分割点。衡量指标包括：
熵(Entropy)： $$H(S) = -\sum_{i=1}^{c} p_i \log_2(p_i)$$
信息增益(Information Gain)： $$IG(S, A) = H(S) - \sum_{v \in Values(A)} \frac{|S_v|}{|S|} H(S_v)$$
基尼系数(Gini Index)： $$Gini(S) = 1 - \sum_{i=1}^{c} p_i^2$$
def decision_tree_fundamentals():
"""决策树理论实现"""
def entropy(y):
"""计算熵"""
_, counts = np.unique(y, return_counts=True)
probabilities = counts / len(y)
return -np.sum(probabilities * np.log2(probabilities + 1e-10))
def gini_impurity(y):
"""计算基尼系数"""
_, counts = np.unique(y, return_counts=True)
probabilities = counts / len(y)
return 1 - np.sum(probabilities**2)
def information_gain(X, y, feature_idx, threshold):
"""计算信息增益"""
# 分割数据
left_mask = X[:, feature_idx] <= threshold
right_mask = ~left_mask
if np.sum(left_mask) == 0 or np.sum(right_mask) == 0:
return 0
# 计算加权熵
n_samples = len(y)
left_entropy = entropy(y[left_mask])
right_entropy = entropy(y[right_mask])
weighted_entropy = (np.sum(left_mask) / n_samples * left_entropy +
np.sum(right_mask) / n_samples * right_entropy)
return entropy(y) - weighted_entropy
def find_best_split(X, y):
"""找到最优分割"""
best_gain = 0
best_feature = None
best_threshold = None
n_features = X.shape[1]
for feature_idx in range(n_features):
thresholds = np.unique(X[:, feature_idx])
for threshold in thresholds:
gain = information_gain(X, y, feature_idx, threshold)
if gain > best_gain:
best_gain = gain
best_feature = feature_idx
best_threshold = threshold
return best_feature, best_threshold, best_gain
# 生成非线性可分数据
np.random.seed(42)
n_samples = 800
X = np.random.randn(n_samples, 2)
# XOR问题
y = ((X[:, 0] > 0) ^ (X[:, 1] > 0)).astype(int)
return X, y, find_best_split
决策树训练与剪枝策略
def comprehensive_decision_tree_analysis(X, y):
"""全面的决策树分析"""
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=42, stratify=y
)
# 不同参数的决策树
tree_configs = {
'Unrestricted': {'max_depth': None, 'min_samples_split': 2},
'Depth Limited': {'max_depth': 5, 'min_samples_split': 2},
'Min Samples': {'max_depth': None, 'min_samples_split': 20},
'Pruned': {'max_depth': 10, 'min_samples_split': 10, 'min_samples_leaf': 5}
}
results = {}
for name, config in tree_configs.items():
# 训练模型
model = DecisionTreeClassifier(random_state=42, **config)
model.fit(X_train, y_train)
# 预测
train_pred = model.predict(X_train)
test_pred = model.predict(X_test)
# 评估指标
train_acc = accuracy_score(y_train, train_pred)
test_acc = accuracy_score(y_test, test_pred)
# 模型复杂度
tree_depth = model.get_depth()
n_leaves = model.get_n_leaves()
results[name] = {
'model': model,
'train_accuracy': train_acc,
'test_accuracy': test_acc,
'tree_depth': tree_depth,
'n_leaves': n_leaves,
'feature_importance': model.feature_importances_
}
print(f"{name}:")
print(f" 训练准确率: {train_acc:.3f}")
print(f" 测试准确率: {test_acc:.3f}")
print(f" 树深度: {tree_depth}")
print(f" 叶节点数: {n_leaves}")
print(f" 过拟合程度: {train_acc - test_acc:.3f}\n")
return results, (X_test, y_test)
def analyze_decision_boundaries(results, test_data):
"""分析决策边界"""
X_test, y_test = test_data
fig, axes = plt.subplots(2, 2, figsize=(12, 10))
axes = axes.ravel()
for i, (name, result) in enumerate(results.items()):
model = result['model']
# 创建网格
x_min, x_max = X_test[:, 0].min() - 0.5, X_test[:, 0].max() + 0.5
y_min, y_max = X_test[:, 1].min() - 0.5, X_test[:, 1].max() + 0.5
xx, yy = np.meshgrid(np.linspace(x_min, x_max, 100),
np.linspace(y_min, y_max, 100))
# 预测网格点
Z = model.predict(np.c_[xx.ravel(), yy.ravel()])
Z = Z.reshape(xx.shape)
# 绘制决策边界
axes[i].contourf(xx, yy, Z, alpha=0.3, cmap=plt.cm.RdYlBu)
scatter = axes[i].scatter(X_test[:, 0], X_test[:, 1], c=y_test,
cmap=plt.cm.RdYlBu, edgecolors='black')
axes[i].set_title(f'{name}\nTest Acc: {result["test_accuracy"]:.3f}')
axes[i].set_xlabel('Feature 1')
axes[i].set_ylabel('Feature 2')
plt.tight_layout()
plt.show()
集成学习：Bagging与Bootstrap理论
随机森林的数学基础
随机森林结合两种随机化策略：
- Bootstrap采样：从原始数据集中有放回地抽取n个样本
- 特征随机化：在每个节点分割时随机选择$\sqrt{p}$个特征
最终预测通过多数投票：
$$\hat{y} = \text{mode}\{h_1(\mathbf{x}), h_2(\mathbf{x}), ..., h_T(\mathbf{x})\}$$方差的偏差-方差分解：
$$\text{Error} = \text{Bias}^2 + \text{Variance} + \text{Noise}$$随机森林通过集成降低方差，同时保持较低的偏差。
def ensemble_learning_theory():
"""集成学习理论与实现"""
def bootstrap_sample(X, y, random_state=None):
"""生成Bootstrap样本"""
if random_state is not None:
np.random.seed(random_state)
n_samples = X.shape[0]
bootstrap_indices = np.random.choice(n_samples, n_samples, replace=True)
return X[bootstrap_indices], y[bootstrap_indices], bootstrap_indices
def random_forest_from_scratch(X, y, n_trees=10, max_depth=5, min_samples_split=2):
"""从零实现随机森林"""
class SimpleTree:
def __init__(self, max_depth, min_samples_split):
self.max_depth = max_depth
self.min_samples_split = min_samples_split
self.feature_subset_size = int(np.sqrt(X.shape[1]))
def fit(self, X, y, depth=0):
self.n_samples = X.shape[0]
self.n_features = X.shape[1]
# 停止条件
if (depth >= self.max_depth or
self.n_samples < self.min_samples_split or
len(np.unique(y)) == 1):
self.prediction = np.bincount(y).argmax()
return self
# 随机选择特征子集
feature_indices = np.random.choice(
self.n_features, self.feature_subset_size, replace=False
)
# 找到最优分割
best_gain = 0
for feature_idx in feature_indices:
thresholds = np.unique(X[:, feature_idx])
for threshold in thresholds:
gain = self._information_gain(X, y, feature_idx, threshold)
if gain > best_gain:
best_gain = gain
self.feature_idx = feature_idx
self.threshold = threshold
if best_gain == 0:
self.prediction = np.bincount(y).argmax()
return self
# 分割数据
left_mask = X[:, self.feature_idx] <= self.threshold
right_mask = ~left_mask
self.left = SimpleTree(self.max_depth, self.min_samples_split)
self.right = SimpleTree(self.max_depth, self.min_samples_split)
self.left.fit(X[left_mask], y[left_mask], depth + 1)
self.right.fit(X[right_mask], y[right_mask], depth + 1)
return self
def _information_gain(self, X, y, feature_idx, threshold):
parent_entropy = self._entropy(y)
left_mask = X[:, feature_idx] <= threshold
right_mask = ~left_mask
if np.sum(left_mask) == 0 or np.sum(right_mask) == 0:
return 0
n_samples = len(y)
left_entropy = self._entropy(y[left_mask])
right_entropy = self._entropy(y[right_mask])
child_entropy = (np.sum(left_mask) / n_samples * left_entropy +
np.sum(right_mask) / n_samples * right_entropy)
return parent_entropy - child_entropy
def _entropy(self, y):
_, counts = np.unique(y, return_counts=True)
probabilities = counts / len(y)
return -np.sum(probabilities * np.log2(probabilities + 1e-10))
def predict(self, X):
if hasattr(self, 'prediction'):
return np.full(X.shape[0], self.prediction)
predictions = np.zeros(X.shape[0])
left_mask = X[:, self.feature_idx] <= self.threshold
right_mask = ~left_mask
if np.sum(left_mask) > 0:
predictions[left_mask] = self.left.predict(X[left_mask])
if np.sum(right_mask) > 0:
predictions[right_mask] = self.right.predict(X[right_mask])
return predictions
# 训练多棵树
trees = []
for i in range(n_trees):
# Bootstrap采样
X_bootstrap, y_bootstrap, _ = bootstrap_sample(X, y, random_state=i)
# 训练单棵树
tree = SimpleTree(max_depth, min_samples_split)
tree.fit(X_bootstrap, y_bootstrap)
trees.append(tree)
return trees
def predict_forest(trees, X):
"""随机森林预测"""
predictions = np.zeros((len(trees), X.shape[0]))
for i, tree in enumerate(trees):
predictions[i] = tree.predict(X)
# 多数投票
final_predictions = np.array([np.bincount(predictions[:, i].astype(np.int64)).argmax()
for i in range(X.shape[0])])
return final_predictions
return bootstrap_sample, random_forest_from_scratch, predict_forest
集成学习性能分析
def ensemble_performance_analysis(X, y):
"""集成学习性能分析"""
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=42, stratify=y
)
# 对比不同的集成方法
ensemble_models = {
'Single Tree': DecisionTreeClassifier(max_depth=10, random_state=42),
'Random Forest': RandomForestClassifier(n_estimators=100, max_depth=10, random_state=42),
'Gradient Boosting': GradientBoostingClassifier(n_estimators=100, max_depth=3, random_state=42)
}
results = {}
for name, model in ensemble_models.items():
model.fit(X_train, y_train)
train_pred = model.predict(X_train)
test_pred = model.predict(X_test)
# 计算多种指标
train_acc = accuracy_score(y_train, train_pred)
test_acc = accuracy_score(y_test, test_pred)
# 交叉验证
cv_scores = cross_val_score(model, X_train, y_train, cv=5)
results[name] = {
'model': model,
'train_accuracy': train_acc,
'test_accuracy': test_acc,
'cv_mean': cv_scores.mean(),
'cv_std': cv_scores.std(),
'overfitting': train_acc - test_acc
}
print(f"{name}:")
print(f" 训练准确率: {train_acc:.3f}")
print(f" 测试准确率: {test_acc:.3f}")
print(f" CV平均分: {cv_scores.mean():.3f} (±{cv_scores.std():.3f})")
print(f" 过拟合程度: {train_acc - test_acc:.3f}\n")
return results, (X_test, y_test)
def feature_importance_analysis(results):
"""特征重要性分析"""
fig, axes = plt.subplots(1, 3, figsize=(15, 5))
for i, (name, result) in enumerate(results.items()):
model = result['model']
if hasattr(model, 'feature_importances_'):
importances = model.feature_importances_
feature_names = [f'Feature {i}' for i in range(len(importances))]
axes[i].bar(feature_names, importances)
axes[i].set_title(f'{name} - Feature Importance')
axes[i].set_ylabel('Importance')
axes[i].tick_params(axis='x', rotation=45)
plt.tight_layout()
plt.show()
支持向量机：间隔最大化理论
SVM的数学原理
支持向量机目标是找到间隔最大的分离超平面。对于线性可分情况，以下硬间隔公式要求标签编码为$y_i \in \{-1,+1\}$；后续sklearn示例可使用0/1标签，由库内部处理：
目标函数： $$\min_{\mathbf{w}, b} \frac{1}{2} \|\mathbf{w}\|^2$$
约束条件： $$y_i(\mathbf{w}^T\mathbf{x}_i + b) \geq 1, \quad i = 1, 2, ..., n$$
对于非线性情况，使用核函数$K(\mathbf{x}_i, \mathbf{x}_j)$将数据映射到高维空间：
常用核函数：
- RBF核：$K(\mathbf{x}_i, \mathbf{x}_j) = \exp(-\gamma \|\mathbf{x}_i - \mathbf{x}_j\|^2)$
- 多项式核：$K(\mathbf{x}_i, \mathbf{x}_j) = (\mathbf{x}_i^T\mathbf{x}_j + c)^d$
- Sigmoid核：$K(\mathbf{x}_i, \mathbf{x}_j) = \tanh(\alpha \mathbf{x}_i^T\mathbf{x}_j + c)$
def svm_comprehensive_analysis():
"""全面的SVM分析"""
# 生成非线性可分数据
np.random.seed(42)
n_samples = 300
# 同心圆数据集
theta = np.random.uniform(0, 2*np.pi, n_samples)
r1 = np.random.normal(2, 0.3, n_samples//2) # 内圈
r2 = np.random.normal(4, 0.3, n_samples//2) # 外圈
# 内圈（类别0）
x1_inner = r1[:n_samples//2] * np.cos(theta[:n_samples//2])
x2_inner = r1[:n_samples//2] * np.sin(theta[:n_samples//2])
# 外圈（类别1）
x1_outer = r2 * np.cos(theta[n_samples//2:])
x2_outer = r2 * np.sin(theta[n_samples//2:])
X = np.vstack([
np.column_stack([x1_inner, x2_inner]),
np.column_stack([x1_outer, x2_outer])
])
y = np.hstack([np.zeros(n_samples//2), np.ones(n_samples//2)])
return X, y
SVM核函数比较与调优
def svm_kernel_comparison(X, y):
"""比较不同SVM核函数"""
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=42, stratify=y
)
# 标准化数据
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)
# 不同核函数的SVM
kernels = {
'Linear': SVC(kernel='linear', random_state=42),
'RBF': SVC(kernel='rbf', gamma='scale', random_state=42),
'Polynomial': SVC(kernel='poly', degree=3, random_state=42),
'RBF (C=0.1)': SVC(kernel='rbf', C=0.1, gamma='scale', random_state=42),
'RBF (C=10)': SVC(kernel='rbf', C=10, gamma='scale', random_state=42)
}
results = {}
for name, model in kernels.items():
# 训练模型
model.fit(X_train_scaled, y_train)
# 预测
train_pred = model.predict(X_train_scaled)
test_pred = model.predict(X_test_scaled)
# 评估指标
train_acc = accuracy_score(y_train, train_pred)
test_acc = accuracy_score(y_test, test_pred)
# 支持向量信息
n_support = model.n_support_
results[name] = {
'model': model,
'train_accuracy': train_acc,
'test_accuracy': test_acc,
'n_support_vectors': np.sum(n_support),
'support_vector_ratio': np.sum(n_support) / len(X_train_scaled)
}
print(f"{name}:")
print(f" 训练准确率: {train_acc:.3f}")
print(f" 测试准确率: {test_acc:.3f}")
print(f" 支持向量数: {np.sum(n_support)}")
print(f" 支持向量比例: {np.sum(n_support) / len(X_train_scaled):.3f}\n")
return results, (X_test_scaled, y_test, scaler)
def visualize_svm_decision_boundaries(results, test_data, X_original):
"""可视化SVM决策边界"""
X_test_scaled, y_test, scaler = test_data
fig, axes = plt.subplots(2, 3, figsize=(18, 12))
axes = axes.ravel()
model_names = list(results.keys())
for i, (name, result) in enumerate(results.items()):
if i >= 6: # 最多显示6个模型
break
model = result['model']
# 创建网格
x_min, x_max = X_original[:, 0].min() - 1, X_original[:, 0].max() + 1
y_min, y_max = X_original[:, 1].min() - 1, X_original[:, 1].max() + 1
xx, yy = np.meshgrid(np.linspace(x_min, x_max, 100),
np.linspace(y_min, y_max, 100))
# 标准化网格点
grid_points = np.c_[xx.ravel(), yy.ravel()]
grid_points_scaled = scaler.transform(grid_points)
# 预测网格点
Z = model.predict(grid_points_scaled)
Z = Z.reshape(xx.shape)
# 绘制决策边界
axes[i].contourf(xx, yy, Z, alpha=0.3, cmap=plt.cm.RdYlBu)
# 绘制原始数据点
scatter = axes[i].scatter(X_original[:, 0], X_original[:, 1],
c=np.hstack([np.zeros(len(X_original)//2),
np.ones(len(X_original)//2)]),
cmap=plt.cm.RdYlBu, edgecolors='black', alpha=0.7)
# 高亮支持向量
if hasattr(model, 'support_vectors_'):
support_vectors_original = scaler.inverse_transform(model.support_vectors_)
axes[i].scatter(support_vectors_original[:, 0], support_vectors_original[:, 1],
s=100, facecolors='none', edgecolors='red', linewidth=2)
axes[i].set_title(f'{name}\nTest Acc: {result["test_accuracy"]:.3f}')
axes[i].set_xlabel('Feature 1')
axes[i].set_ylabel('Feature 2')
plt.tight_layout()
plt.show()
分类性能评估体系
混淆矩阵与评估指标
对于二分类问题，混淆矩阵提供完整的性能图像：
| 预测正例 | 预测负例 | |
|---|---|---|
| 实际正例 | TP | FN |
| 实际负例 | FP | TN |
核心评估指标：
- 准确率(Accuracy)：$\frac{TP + TN}{TP + TN + FP + FN}$
- 精确率(Precision)：$\frac{TP}{TP + FP}$
- 召回率(Recall)：$\frac{TP}{TP + FN}$
- F1分数：$\frac{2 \cdot Precision \cdot Recall}{Precision + Recall}$
ROC曲线与AUC指标
ROC曲线绘制真正例率(TPR)与假正例率(FPR)的关系：
- TPR (Sensitivity)：$\frac{TP}{TP + FN}$
- FPR：$\frac{FP}{FP + TN}$
AUC(Area Under Curve)衡量分类器的整体性能，范围为[0,1]，越接近1越好。
def comprehensive_classification_evaluation(y_true, y_pred, y_proba=None):
"""全面的分类评估"""
# 基础指标
accuracy = accuracy_score(y_true, y_pred)
precision = precision_score(y_true, y_pred, average='weighted')
recall = recall_score(y_true, y_pred, average='weighted')
f1 = f1_score(y_true, y_pred, average='weighted')
print("分类性能评估:")
print(f"准确率: {accuracy:.3f}")
print(f"精确率: {precision:.3f}")
print(f"召回率: {recall:.3f}")
print(f"F1分数: {f1:.3f}")
# 混淆矩阵
cm = confusion_matrix(y_true, y_pred)
# 可视化
fig, axes = plt.subplots(1, 3, figsize=(15, 5))
# 1. 混淆矩阵
sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', ax=axes[0])
axes[0].set_title('混淆矩阵')
axes[0].set_xlabel('预测类别')
axes[0].set_ylabel('实际类别')
# 2. ROC曲线
if y_proba is not None and len(np.unique(y_true)) == 2:
fpr, tpr, _ = roc_curve(y_true, y_proba[:, 1])
roc_auc = auc(fpr, tpr)
axes[1].plot(fpr, tpr, color='darkorange', lw=2,
label=f'ROC curve (AUC = {roc_auc:.2f})')
axes[1].plot([0, 1], [0, 1], color='navy', lw=2, linestyle='--')
axes[1].set_xlim([0.0, 1.0])
axes[1].set_ylim([0.0, 1.05])
axes[1].set_xlabel('假正例率')
axes[1].set_ylabel('真正例率')
axes[1].set_title('ROC曲线')
axes[1].legend(loc="lower right")
# 3. Precision-Recall曲线
precision_vals, recall_vals, _ = precision_recall_curve(y_true, y_proba[:, 1])
pr_auc = auc(recall_vals, precision_vals)
axes[2].plot(recall_vals, precision_vals, color='blue', lw=2,
label=f'PR curve (AUC = {pr_auc:.2f})')
axes[2].set_xlabel('召回率')
axes[2].set_ylabel('精确率')
axes[2].set_title('Precision-Recall曲线')
axes[2].legend()
plt.tight_layout()
plt.show()
return {
'accuracy': accuracy,
'precision': precision,
'recall': recall,
'f1': f1,
'confusion_matrix': cm
}
类别不平衡问题解决方案
SMOTE过采样技术
类别不平衡是实际应用中的常见问题。SMOTE(Synthetic Minority Oversampling Technique)通过插值生成少数类样本：
def handle_imbalanced_data(X, y, strategy='smote'):
"""处理类别不平衡数据"""
print(f"原始数据类别分布:")
unique, counts = np.unique(y, return_counts=True)
for label, count in zip(unique, counts):
print(f" 类别 {label}: {count} ({count/len(y)*100:.1f}%)")
if strategy == 'smote':
# SMOTE过采样
smote = SMOTE(random_state=42)
X_resampled, y_resampled = smote.fit_resample(X, y)
elif strategy == 'class_weight':
# 计算类别权重
class_weights = compute_class_weight('balanced', classes=unique, y=y)
weight_dict = dict(zip(unique, class_weights))
print(f"类别权重: {weight_dict}")
return X, y, weight_dict
print(f"\n{strategy}后类别分布:")
unique_new, counts_new = np.unique(y_resampled, return_counts=True)
for label, count in zip(unique_new, counts_new):
print(f" 类别 {label}: {count} ({count/len(y_resampled)*100:.1f}%)")
return X_resampled, y_resampled, None
def compare_imbalanced_strategies(X, y):
"""比较不同的不平衡处理策略"""
strategies = {
'Original': (X, y, None),
'SMOTE': handle_imbalanced_data(X, y, 'smote'),
'Class Weight': handle_imbalanced_data(X, y, 'class_weight')
}
results = {}
for name, (X_data, y_data, weights) in strategies.items():
X_train, X_test, y_train, y_test = train_test_split(
X_data, y_data, test_size=0.2, random_state=42, stratify=y_data
)
# 使用随机森林
if weights:
model = RandomForestClassifier(class_weight=weights, random_state=42)
else:
model = RandomForestClassifier(random_state=42)
model.fit(X_train, y_train)
y_pred = model.predict(X_test)
# 计算各类别的F1分数
f1_scores = f1_score(y_test, y_pred, average=None)
results[name] = {
'model': model,
'f1_scores': f1_scores,
'macro_f1': f1_score(y_test, y_pred, average='macro'),
'weighted_f1': f1_score(y_test, y_pred, average='weighted')
}
print(f"{name}:")
print(f" 各类别F1: {f1_scores}")
print(f" 宏平均F1: {results[name]['macro_f1']:.3f}")
print(f" 加权平均F1: {results[name]['weighted_f1']:.3f}\n")
return results
客户流失预测项目示例：合成数据建模与业务情景分析
业务场景与数据准备
电信客户流失预测是典型的二分类问题，具有高业务价值和技术挑战性：
def create_customer_churn_dataset():
"""创建客户流失数据集"""
np.random.seed(42)
n_customers = 5000
# 客户基本特征
data = {
'tenure': np.random.exponential(24, n_customers), # 在网时长(月)
'monthly_charges': np.random.gamma(2, 35, n_customers), # 月消费
'total_charges': np.random.exponential(1200, n_customers), # 总消费
'age': np.random.normal(45, 15, n_customers), # 年龄
'contract_type': np.random.choice([0, 1, 2], n_customers, p=[0.5, 0.3, 0.2]), # 合同类型
'payment_method': np.random.choice([0, 1, 2, 3], n_customers), # 支付方式
'internet_service': np.random.choice([0, 1, 2], n_customers), # 网络服务
'tech_support': np.random.choice([0, 1], n_customers), # 技术支持
'online_security': np.random.choice([0, 1], n_customers), # 在线安全
'device_protection': np.random.choice([0, 1], n_customers), # 设备保护
}
df = pd.DataFrame(data)
# 确保合理范围
df['tenure'] = np.clip(df['tenure'], 1, 72)
df['monthly_charges'] = np.clip(df['monthly_charges'], 20, 150)
df['age'] = np.clip(df['age'], 18, 80)
# 构建流失概率（基于业务逻辑）
churn_logit = (
-2.0 + # 基础截距
-0.05 * df['tenure'] + # 在网时间越长越不容易流失
0.02 * df['monthly_charges'] + # 费用越高越容易流失
-0.0001 * df['total_charges'] + # 总消费高的客户价值高
-0.01 * df['age'] + # 年龄大的相对稳定
0.8 * (df['contract_type'] == 0) + # 月付客户易流失
0.3 * (df['payment_method'] == 0) + # 电子支票易流失
-0.5 * df['tech_support'] + # 有技术支持不易流失
-0.3 * df['online_security'] # 有安全服务不易流失
)
# 添加随机噪声
churn_logit += np.random.normal(0, 0.5, n_customers)
# 转换为概率并生成标签
churn_prob = 1 / (1 + np.exp(-churn_logit))
df['churn'] = np.random.binomial(1, churn_prob)
print(f"数据集信息:")
print(f" 客户数量: {len(df)}")
print(f" 流失率: {df['churn'].mean():.2%}")
print(f" 特征数量: {len(df.columns) - 1}")
return df
def advanced_feature_engineering(df):
"""高级特征工程"""
df_processed = df.copy()
# 1. 组合特征
df_processed['charges_per_tenure'] = df_processed['total_charges'] / (df_processed['tenure'] + 1)
df_processed['monthly_charges_ratio'] = df_processed['monthly_charges'] / df_processed['monthly_charges'].mean()
# 2. 分箱特征
df_processed['tenure_group'] = pd.cut(df_processed['tenure'],
bins=[0, 12, 24, 48, 100],
labels=['<1年', '1-2年', '2-4年', '>4年'])
df_processed['charges_group'] = pd.cut(df_processed['monthly_charges'],
bins=[0, 50, 80, 120, 200],
labels=['低', '中低', '中高', '高'])
# 3. 服务组合特征
service_cols = ['tech_support', 'online_security', 'device_protection']
df_processed['total_services'] = df_processed[service_cols].sum(axis=1)
# 4. 风险评分
risk_factors = [
(df_processed['contract_type'] == 0) * 2, # 月付合同
(df_processed['tenure'] < 12) * 1.5, # 新客户
(df_processed['monthly_charges'] > 100) * 1, # 高消费
(df_processed['total_services'] == 0) * 1, # 无增值服务
]
df_processed['risk_score'] = sum(risk_factors)
print("特征工程完成:")
print(f" 原始特征数: {len(df.columns) - 1}")
print(f" 处理后特征数: {len([col for col in df_processed.columns if col != 'churn'])}")
return df_processed
def comprehensive_churn_modeling(df):
"""全面的客户流失建模"""
# 特征选择
feature_cols = ['tenure', 'monthly_charges', 'total_charges', 'age',
'contract_type', 'payment_method', 'internet_service',
'tech_support', 'online_security', 'device_protection',
'charges_per_tenure', 'monthly_charges_ratio', 'total_services', 'risk_score']
X = df[feature_cols]
y = df['churn']
# 数据分割
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=42, stratify=y
)
# 特征标准化
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)
# 处理类别不平衡
smote = SMOTE(random_state=42)
X_train_balanced, y_train_balanced = smote.fit_resample(X_train_scaled, y_train)
# 模型训练与比较
models = {
'Logistic Regression': LogisticRegression(random_state=42, max_iter=1000),
'Random Forest': RandomForestClassifier(n_estimators=100, random_state=42),
'Gradient Boosting': GradientBoostingClassifier(n_estimators=100, random_state=42),
'SVM': SVC(kernel='rbf', probability=True, random_state=42)
}
results = {}
for name, model in models.items():
# 训练模型
model.fit(X_train_balanced, y_train_balanced)
# 预测
y_pred = model.predict(X_test_scaled)
y_pred_proba = model.predict_proba(X_test_scaled)[:, 1] if hasattr(model, 'predict_proba') else None
# 评估指标
accuracy = accuracy_score(y_test, y_pred)
precision = precision_score(y_test, y_pred)
recall = recall_score(y_test, y_pred)
f1 = f1_score(y_test, y_pred)
if y_pred_proba is not None:
fpr, tpr, _ = roc_curve(y_test, y_pred_proba)
roc_auc = auc(fpr, tpr)
else:
roc_auc = None
results[name] = {
'model': model,
'accuracy': accuracy,
'precision': precision,
'recall': recall,
'f1': f1,
'auc': roc_auc,
'predictions': y_pred,
'probabilities': y_pred_proba
}
print(f"{name}:")
print(f" 准确率: {accuracy:.3f}")
print(f" 精确率: {precision:.3f}")
print(f" 召回率: {recall:.3f}")
print(f" F1分数: {f1:.3f}")
if roc_auc:
print(f" AUC: {roc_auc:.3f}")
print()
return results, (X_test_scaled, y_test), feature_cols
def business_impact_analysis(results, test_data, feature_cols):
"""业务影响分析"""
X_test, y_test = test_data
# 选择最佳模型（基于F1分数）
best_model_name = max(results.keys(), key=lambda k: results[k]['f1'])
best_model = results[best_model_name]['model']
best_predictions = results[best_model_name]['predictions']
best_probabilities = results[best_model_name]['probabilities']
print(f"最佳模型: {best_model_name}")
# 业务指标计算
tp = np.sum((y_test == 1) & (best_predictions == 1)) # 正确识别的流失客户
fp = np.sum((y_test == 0) & (best_predictions == 1)) # 误判的非流失客户
fn = np.sum((y_test == 1) & (best_predictions == 0)) # 漏掉的流失客户
# 假设业务参数
avg_customer_value = 1000 # 平均客户价值
retention_cost = 100 # 挽留成本
retention_success_rate = 0.3 # 挽留成功率
# 计算业务价值
value_saved = tp * retention_success_rate * avg_customer_value
cost_incurred = (tp + fp) * retention_cost
value_lost = fn * avg_customer_value
net_benefit = value_saved - cost_incurred
print(f"\n业务影响分析:")
print(f" 正确识别流失客户: {tp}")
print(f" 误判非流失客户: {fp}")
print(f" 漏掉流失客户: {fn}")
print(f" 预计挽回价值: ¥{value_saved:,.0f}")
print(f" 挽留成本: ¥{cost_incurred:,.0f}")
print(f" 损失价值: ¥{value_lost:,.0f}")
print(f" 净收益: ¥{net_benefit:,.0f}")
# 特征重要性分析
if hasattr(best_model, 'feature_importances_'):
feature_importance = pd.DataFrame({
'feature': feature_cols,
'importance': best_model.feature_importances_
}).sort_values('importance', ascending=False)
print(f"\n关键特征排序:")
for idx, row in feature_importance.head(10).iterrows():
print(f" {row['feature']}: {row['importance']:.3f}")
# 风险分层
if best_probabilities is not None:
risk_groups = pd.cut(best_probabilities,
bins=[0, 0.3, 0.7, 1.0],
labels=['低风险', '中风险', '高风险'])
print(f"\n客户风险分层:")
for risk_level in ['低风险', '中风险', '高风险']:
mask = risk_groups == risk_level
actual_churn_rate = y_test[mask].mean() if mask.sum() > 0 else 0
print(f" {risk_level}: {mask.sum()} 客户, 实际流失率: {actual_churn_rate:.1%}")
return best_model_name, net_benefit
模型部署与监控
def create_model_monitoring_framework():
"""创建模型监控框架"""
class ChurnModelMonitor:
def __init__(self, model, scaler, feature_cols, baseline_metrics):
self.model = model
self.scaler = scaler
self.feature_cols = feature_cols
self.baseline_metrics = baseline_metrics
self.prediction_log = []
def predict_and_log(self, X_new, y_true=None):
"""预测并记录"""
# 预处理
X_scaled = self.scaler.transform(X_new)
# 预测
predictions = self.model.predict(X_scaled)
probabilities = self.model.predict_proba(X_scaled)[:, 1]
# 记录预测结果
log_entry = {
'timestamp': pd.Timestamp.now(),
'n_samples': len(X_new),
'predictions': predictions,
'probabilities': probabilities,
'churn_rate_predicted': predictions.mean()
}
if y_true is not None:
log_entry.update({
'y_true': y_true,
'accuracy': accuracy_score(y_true, predictions),
'precision': precision_score(y_true, predictions),
'recall': recall_score(y_true, predictions),
'f1': f1_score(y_true, predictions)
})
self.prediction_log.append(log_entry)
return predictions, probabilities
def check_data_drift(self, X_new, threshold=0.1):
"""检测数据漂移"""
drift_detected = []
for i, col in enumerate(self.feature_cols):
# 简单的统计检验（实际应用中可用KS检验等）
baseline_mean = 0 # 应该存储训练时的统计量
baseline_std = 1
current_mean = X_new[:, i].mean()
current_std = X_new[:, i].std()
mean_shift = abs(current_mean - baseline_mean) / baseline_std
std_shift = abs(current_std - baseline_std) / baseline_std
if mean_shift > threshold or std_shift > threshold:
drift_detected.append({
'feature': col,
'mean_shift': mean_shift,
'std_shift': std_shift
})
return drift_detected
def generate_monitoring_report(self):
"""生成监控报告"""
if not self.prediction_log:
print("暂无预测记录")
return
recent_logs = self.prediction_log[-10:] # 最近10次预测
print("模型监控报告:")
print(f" 预测次数: {len(self.prediction_log)}")
print(f" 最近预测流失率: {recent_logs[-1]['churn_rate_predicted']:.1%}")
if 'accuracy' in recent_logs[-1]:
recent_accuracy = np.mean([log['accuracy'] for log in recent_logs if 'accuracy' in log])
print(f" 最近准确率: {recent_accuracy:.3f}")
print(f" 基线准确率: {self.baseline_metrics.get('accuracy', 'N/A')}")
return ChurnModelMonitor
最佳实践与技术总结
算法选择决策树
def classification_algorithm_selector():
"""分类算法选择指南"""
decision_tree = {
"数据规模": {
"小数据集 (<1000)": ["SVM", "逻辑回归"],
"中等数据集 (1k-100k)": ["随机森林", "梯度提升"],
"大数据集 (>100k)": ["逻辑回归", "线性SVM"]
},
"数据特性": {
"线性可分": ["逻辑回归", "线性SVM"],
"非线性": ["随机森林", "RBF SVM", "梯度提升"],
"高维稀疏": ["逻辑回归", "线性SVM"],
"类别不平衡": ["集成方法 + 重采样", "调整类别权重"]
},
"业务需求": {
"可解释性强": ["逻辑回归", "决策树"],
"预测准确性": ["随机森林", "梯度提升"],
"训练速度快": ["逻辑回归", "朴素贝叶斯"],
"预测速度快": ["逻辑回归", "决策树"]
},
"技术约束": {
"内存限制": ["在线学习算法", "增量学习"],
"实时预测": ["线性模型", "浅层决策树"],
"分布式计算": ["可并行化算法"],
"模型更新": ["增量学习算法"]
}
}
print("分类算法选择指南:")
for category, subcategories in decision_tree.items():
print(f"\n{category}:")
for scenario, algorithms in subcategories.items():
print(f" {scenario}: {', '.join(algorithms)}")
性能优化策略
def performance_optimization_strategies():
"""性能优化策略"""
strategies = {
"数据层面": [
"特征选择：移除不相关特征",
"特征缩放：标准化或归一化",
"特征工程：构造有意义的组合特征",
"数据清洗：处理异常值和噪声",
"降维：PCA、特征选择算法"
],
"算法层面": [
"超参数调优：网格搜索、贝叶斯优化",
"集成方法：Bagging、Boosting、Stacking",
"正则化：L1/L2正则化防止过拟合",
"交叉验证：估计模型泛化表现，结果依赖数据切分与评估设计",
"早停：避免过度训练"
],
"工程层面": [
"并行计算：利用多核CPU",
"GPU加速：深度学习框架",
"内存优化：批处理、数据流",
"模型压缩：剪枝、量化",
"缓存策略：预计算常用结果"
]
}
print("性能优化策略:")
for category, items in strategies.items():
print(f"\n{category}:")
for item in items:
print(f" • {item}")
def common_pitfalls_and_solutions():
"""常见陷阱与解决方案"""
pitfalls = {
"数据泄露": {
"问题": "使用未来信息预测过去",
"解决": "严格的时间切分，特征工程时避免未来信息"
},
"过拟合": {
"问题": "模型在训练集上表现好，测试集上差",
"解决": "正则化、交叉验证、增加数据、简化模型"
},
"类别不平衡": {
"问题": "少数类被忽略，准确率高但无意义",
"解决": "重采样、调整类别权重、使用F1等平衡指标"
},
"特征缩放": {
"问题": "不同尺度特征影响模型性能",
"解决": "标准化、归一化，注意测试集使用训练集参数"
},
"评估偏差": {
"问题": "错误的评估方式导致模型选择失误",
"解决": "使用合适的评估指标，避免数据泄露"
}
}
print("常见陷阱与解决方案:")
for pitfall, details in pitfalls.items():
print(f"\n{pitfall}:")
print(f" 问题: {details['问题']}")
print(f" 解决: {details['解决']}")
技术发展趋势
自动机器学习(AutoML)
def automl_classification_example():
"""AutoML分类示例"""
print("AutoML在分类中的应用:")
print("• 自动特征工程：特征选择、变换、组合")
print("• 自动模型选择：算法选择、超参数优化")
print("• 自动集成：多模型融合策略")
print("• 神经架构搜索：自动设计网络结构")
print("• 可解释AI：自动生成模型解释")
print("\n主流AutoML工具:")
tools = [
"H2O AutoML: 开源企业级平台",
"Auto-sklearn: 基于sklearn的自动化",
"TPOT: 基于遗传编程的优化",
"AutoKeras: 深度学习AutoML",
"Google AutoML: 云端AI服务"
]
for tool in tools:
print(f" • {tool}")
深度学习分类器
def deep_learning_classification_trends():
"""深度学习分类发展趋势"""
trends = {
"Transformer架构": "注意力机制在表格数据上的应用",
"图神经网络": "处理结构化关系数据的分类",
"元学习": "Few-shot学习，快速适应新类别",
"对比学习": "自监督表示学习",
"联邦学习": "隐私保护的分布式训练",
"可解释AI": "神经网络的可解释性增强",
"边缘部署": "移动端、IoT设备上的推理优化"
}
print("深度学习分类发展趋势:")
for trend, description in trends.items():
print(f" • {trend}: {description}")
总结
分类算法是机器学习的核心技术之一，本文深入解析了主流分类算法的数学原理和工程实践：
核心算法掌握：
- 逻辑回归：线性分类的数学基础
- 决策树：信息理论驱动的递归分割
- 随机森林：集成学习的经典代表
- SVM：间隔最大化的几何直觉
技术要点：
- 决策边界理论与概率估计
- 信息增益、熵与基尼系数
- Bootstrap采样与特征随机化
- 核函数与高维映射
工程实践：
- 性能评估体系：混淆矩阵、ROC曲线、AUC指标
- 类别不平衡处理：SMOTE、类别权重调整
- 特征工程与模型调优
- 合成数据项目流程与假设业务情景分析
发展趋势：
- AutoML自动化机器学习
- 深度学习分类器
- 可解释AI与模型透明度
掌握这些核心概念和实践技巧，能够在实际项目中选择合适的分类算法，构建高效可靠的分类系统。
相关链接：
参考资源：
- Scikit-learn官方文档
- 《统计学习方法》- 李航
- 《机器学习》- 周志华
- Kaggle分类竞赛案例
创建时间：2024-06-29
最后更新：2024-09-02
字数统计：约7800字
阅读时长：50-60分钟
#机器学习 #分类算法 #逻辑回归 #决策树 #随机森林 #SVM #集成学习 #客户流失预测