02. 监督学习算法详解

import numpy as np

# 正规方程实现
class LinearRegressionNormal:
    def fit(self, X, y):
        X_b = np.c_[np.ones((X.shape[0], 1)), X]  # 添加偏置列
        self.w = np.linalg.inv(X_b.T @ X_b) @ X_b.T @ y
    
    def predict(self, X):
        X_b = np.c_[np.ones((X.shape[0], 1)), X]
        return X_b @ self.w

# 梯度下降实现
class LinearRegressionGD:
    def __init__(self, lr=0.01, epochs=1000):
        self.lr = lr
        self.epochs = epochs
    
    def fit(self, X, y):
        n, m = X.shape
        self.w = np.zeros(m + 1)
        X_b = np.c_[np.ones((n, 1)), X]
        
        for _ in range(self.epochs):
            grad = (2/n) * X_b.T @ (X_b @ self.w - y)
            self.w -= self.lr * grad
    
    def predict(self, X):
        X_b = np.c_[np.ones((X.shape[0], 1)), X]
        return X_b @ self.w

# sklearn 实际使用解析解（小规模）或 SGD（大规模）
from sklearn.linear_model import LinearRegression, SGDRegressor

# 解析解
model = LinearRegression().fit(X_train, y_train)

# 大规模数据用 SGD
model = SGDRegressor(max_iter=1000, tol=1e-3, penalty='l2', alpha=0.0001)
model.fit(X_train, y_train)

from sklearn.linear_model import LogisticRegression
from sklearn.preprocessing import StandardScaler

# 逻辑回归需先标准化
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)

model = LogisticRegression(
    penalty='l2',        # 正则化类型
    C=1.0,               # 正则化强度倒数，越大正则越弱
    solver='lbfgs',      # 优化算法
    max_iter=1000,
    class_weight='balanced'  # 处理类别不平衡
)
model.fit(X_train_scaled, y_train)

# 预测概率
proba = model.predict_proba(X_test_scaled)[:, 1]

# softmax 多分类
model = LogisticRegression(
    multi_class='multinomial',  # softmax 而非 one-vs-rest
    solver='lbfgs',
    max_iter=1000
)

from sklearn.tree import DecisionTreeClassifier, export_text, plot_tree
import matplotlib.pyplot as plt

model = DecisionTreeClassifier(
    criterion='gini',      # 'gini' 或 'entropy'
    max_depth=5,           # 最大深度，防过拟合
    min_samples_split=20,  # 分裂最小样本数
    min_samples_leaf=10,   # 叶子最小样本数
    max_features='sqrt',   # 每次分裂考虑的特征数
    random_state=42
)
model.fit(X_train, y_train)

# 可视化
tree_rules = export_text(model, feature_names=list(X.columns))
print(tree_rules)

plt.figure(figsize=(20, 10))
plot_tree(model, feature_names=X.columns, filled=True, rounded=True)
plt.savefig('tree.png')

importances = pd.Series(model.feature_importances_, index=X.columns)
importances.sort_values(ascending=False).plot(kind='barh')

from sklearn.ensemble import RandomForestClassifier, RandomForestRegressor

# 随机森林分类
rf = RandomForestClassifier(
    n_estimators=200,        # 树的数量
    max_depth=10,
    min_samples_split=5,
    max_features='sqrt',     # 每棵树随机选 sqrt(n_features) 个特征
    bootstrap=True,          # 有放回抽样
    oob_score=True,          # 使用 out-of-bag 样本评估
    n_jobs=-1,               # 并行
    random_state=42
)
rf.fit(X_train, y_train)
print(f"OOB Score: {rf.oob_score_:.3f}")

# 特征重要性
importances = pd.DataFrame({
    'feature': X.columns,
    'importance': rf.feature_importances_
}).sort_values('importance', ascending=False)

from sklearn.ensemble import AdaBoostClassifier

ada = AdaBoostClassifier(
    estimator=DecisionTreeClassifier(max_depth=1),  # 弱分类器（决策树桩）
    n_estimators=100,
    learning_rate=1.0
)
ada.fit(X_train, y_train)

from sklearn.ensemble import GradientBoostingClassifier

gb = GradientBoostingClassifier(
    n_estimators=100,
    learning_rate=0.1,       # 收缩系数，越小越稳健
    max_depth=3,             # 树深度，通常 3-8
    subsample=0.8,           # 每次迭代抽样比例（随机梯度提升）
    random_state=42
)
gb.fit(X_train, y_train)

import xgboost as xgb

# 原生接口
dtrain = xgb.DMatrix(X_train, label=y_train)
dtest = xgb.DMatrix(X_test, label=y_test)

params = {
    'objective': 'binary:logistic',  # 或 'multi:softmax', 'reg:squarederror'
    'max_depth': 6,
    'learning_rate': 0.1,
    'subsample': 0.8,
    'colsample_bytree': 0.8,
    'reg_alpha': 0.1,        # L1
    'reg_lambda': 1.0,       # L2
    'eval_metric': 'auc'
}

model = xgb.train(
    params,
    dtrain,
    num_boost_round=100,
    evals=[(dtrain, 'train'), (dtest, 'test')],
    early_stopping_rounds=10
)

# sklearn 风格接口
from xgboost import XGBClassifier

xgb_model = XGBClassifier(
    n_estimators=100,
    max_depth=6,
    learning_rate=0.1,
    subsample=0.8,
    colsample_bytree=0.8,
    reg_alpha=0.1,
    random_state=42,
    n_jobs=-1
)
xgb_model.fit(X_train, y_train, 
              eval_set=[(X_test, y_test)],
              early_stopping_rounds=10,
              verbose=False)

import lightgbm as lgb

lgb_model = lgb.LGBMClassifier(
    n_estimators=100,
    max_depth=-1,           # 无限制，由 min_child_samples 控制
    learning_rate=0.1,
    num_leaves=31,          # 叶子节点数，控制复杂度
    subsample=0.8,
    colsample_bytree=0.8,
    reg_alpha=0.1,
    random_state=42
)
lgb_model.fit(X_train, y_train,
              eval_set=[(X_test, y_test)],
              callbacks=[lgb.early_stopping(10, verbose=False)])

from sklearn.svm import SVC, LinearSVC

# 线性 SVM（大规模数据）
linear_svc = LinearSVC(C=1.0, max_iter=5000)
linear_svc.fit(X_train_scaled, y_train)

# 核 SVM
rbf_svc = SVC(
    kernel='rbf',         # 'linear', 'poly', 'rbf', 'sigmoid'
    C=1.0,                # 惩罚系数
    gamma='scale',        # 核函数系数
    probability=True      # 启用概率输出（训练 Platt Scaling，较耗时）
)
rbf_svc.fit(X_train_scaled, y_train)

# 预测概率
proba = rbf_svc.predict_proba(X_test_scaled)

from sklearn.neighbors import KNeighborsClassifier

knn = KNeighborsClassifier(
    n_neighbors=5,
    weights='distance',  # 'uniform' 等权重，'distance' 按距离加权
    metric='minkowski',  # 距离度量
    p=2                  # p=2 为欧氏距离，p=1 为曼哈顿
)
knn.fit(X_train_scaled, y_train)  # KNN 实际是惰性学习，fit 只存储数据

from sklearn.naive_bayes import GaussianNB, MultinomialNB

# 连续特征用高斯分布
model = GaussianNB()

# 离散计数特征用多项式分布（如词频）
model = MultinomialNB(alpha=1.0)  # Laplace 平滑
model.fit(X_train, y_train)

from sklearn.model_selection import GridSearchCV, RandomizedSearchCV
from scipy.stats import uniform, randint

# 网格搜索
param_grid = {
    'max_depth': [3, 5, 7],
    'learning_rate': [0.01, 0.1, 0.3],
    'n_estimators': [100, 200]
}
grid = GridSearchCV(
    XGBClassifier(random_state=42),
    param_grid,
    cv=5,
    scoring='f1_macro',
    n_jobs=-1,
    verbose=1
)
grid.fit(X_train, y_train)
print(f"最优参数: {grid.best_params_}")

# 随机搜索（更高效）
param_dist = {
    'max_depth': randint(3, 10),
    'learning_rate': uniform(0.01, 0.3),
    'subsample': uniform(0.6, 0.4),
    'colsample_bytree': uniform(0.6, 0.4)
}
random = RandomizedSearchCV(
    XGBClassifier(random_state=42),
    param_dist,
    n_iter=50,  # 采样 50 组参数
    cv=5,
    scoring='f1_macro',
    n_jobs=-1,
    random_state=42
)

import optuna

def objective(trial):
    params = {
        'max_depth': trial.suggest_int('max_depth', 3, 10),
        'learning_rate': trial.suggest_float('learning_rate', 0.01, 0.3),
        'n_estimators': trial.suggest_int('n_estimators', 50, 300),
        'subsample': trial.suggest_float('subsample', 0.6, 1.0),
        'colsample_bytree': trial.suggest_float('colsample_bytree', 0.6, 1.0),
        'reg_alpha': trial.suggest_float('reg_alpha', 0, 1),
        'reg_lambda': trial.suggest_float('reg_lambda', 0, 1)
    }
    
    model = XGBClassifier(**params, random_state=42, n_jobs=-1)
    scores = cross_val_score(model, X_train, y_train, cv=5, scoring='f1_macro')
    return scores.mean()

study = optuna.create_study(direction='maximize')
study.optimize(objective, n_trials=100, show_progress_bar=True)
print(f"最优参数: {study.best_params}")
print(f"最优得分: {study.best_value:.3f}")