📖 现代表格数据方法¶

# pip install pytorch-tabnet
import torch
from pytorch_tabnet.tab_model import TabNetClassifier
import numpy as np

# 创建模型
clf = TabNetClassifier(
    n_d=64, n_a=64,           # 决策/注意力维度
    n_steps=5,                 # 注意力步数
    gamma=1.5,                 # 特征复用系数
    n_independent=2,           # 独立GLU层数
    n_shared=2,                # 共享GLU层数
    lambda_sparse=1e-4,        # 稀疏正则化
    optimizer_fn=torch.optim.Adam,
    optimizer_params=dict(lr=2e-2),
    scheduler_params={"step_size":50, "gamma":0.9},
    scheduler_fn=torch.optim.lr_scheduler.StepLR,
    mask_type='entmax',        # sparsemax 或 entmax
    verbose=10,
)

# 训练
clf.fit(
    X_train=X_train, y_train=y_train,
    eval_set=[(X_valid, y_valid)],
    eval_name=['valid'],
    eval_metric=['auc'],
    max_epochs=200,
    patience=20,
    batch_size=1024,
    virtual_batch_size=128,    # Ghost Batch Normalization
)

# 特征重要性（全局）
feature_importances = clf.feature_importances_

# 逐样本注意力掩码（局部可解释性）
explain_matrix, masks = clf.explain(X_test)

import torch
import torch.nn as nn
import math

class FeatureTokenizer(nn.Module):  # 继承nn.Module定义神经网络层
    """将表格特征转化为token嵌入"""
    def __init__(self, n_num_features, cat_cardinalities, d_token):  # __init__构造方法，创建对象时自动调用
        super().__init__()  # super()调用父类方法
        self.d_token = d_token

        # 数值特征：每个特征独立的线性嵌入
        if n_num_features > 0:
            self.num_embeddings = nn.Parameter(torch.randn(n_num_features, d_token))
            self.num_bias = nn.Parameter(torch.zeros(n_num_features, d_token))

        # 类别特征：每个特征独立的embedding table
        self.cat_embeddings = nn.ModuleList([
            nn.Embedding(card, d_token) for card in cat_cardinalities
        ])

        # [CLS] token
        self.cls_token = nn.Parameter(torch.randn(1, 1, d_token))

    def forward(self, x_num=None, x_cat=None):
        tokens = []

        if x_num is not None:
            # 数值特征嵌入: (batch, n_num, d_token)
            batch_size = x_num.shape[0]
            num_tokens = x_num.unsqueeze(-1) * self.num_embeddings.unsqueeze(0) + self.num_bias.unsqueeze(0)  # unsqueeze增加一个维度
            tokens.append(num_tokens)

        if x_cat is not None:
            cat_tokens = [
                emb(x_cat[:, i]) for i, emb in enumerate(self.cat_embeddings)  # enumerate同时获取索引和元素
            ]
            tokens.append(torch.stack(cat_tokens, dim=1))

        # 拼接所有特征token
        tokens = torch.cat(tokens, dim=1)

        # 添加[CLS] token
        cls = self.cls_token.expand(tokens.shape[0], -1, -1)
        tokens = torch.cat([cls, tokens], dim=1)

        return tokens

class FTTransformer(nn.Module):
    """FT-Transformer完整模型"""
    def __init__(self, n_num_features, cat_cardinalities, d_token=192,
                 n_layers=3, n_heads=8, d_ffn=256, dropout=0.1, n_classes=2):
        super().__init__()

        self.tokenizer = FeatureTokenizer(
            n_num_features, cat_cardinalities, d_token
        )

        # 标准Transformer Encoder
        encoder_layer = nn.TransformerEncoderLayer(
            d_model=d_token,
            nhead=n_heads,
            dim_feedforward=d_ffn,
            dropout=dropout,
            activation='gelu',
            batch_first=True,
        )
        self.transformer = nn.TransformerEncoder(encoder_layer, n_layers)

        # 分类头：使用[CLS] token的输出
        self.head = nn.Sequential(
            nn.LayerNorm(d_token),
            nn.ReLU(),
            nn.Linear(d_token, n_classes),
        )

    def forward(self, x_num=None, x_cat=None):
        tokens = self.tokenizer(x_num, x_cat)     # (B, 1+n_features, d)
        tokens = self.transformer(tokens)          # (B, 1+n_features, d)
        cls_output = tokens[:, 0]                  # (B, d) 取[CLS]输出
        return self.head(cls_output)               # (B, n_classes)

# === 使用示例 ===
model = FTTransformer(
    n_num_features=10,
    cat_cardinalities=[5, 12, 3],  # 3个类别特征，基数分别为5,12,3
    d_token=192,
    n_layers=3,
    n_heads=8,
    n_classes=2,
)

# 假设输入
x_num = torch.randn(32, 10)                    # 32样本，10个数值特征
x_cat = torch.randint(0, 5, (32, 3))           # 32样本，3个类别特征
logits = model(x_num, x_cat)                    # (32, 2)

# pip install tabpfn
from tabpfn import TabPFNClassifier
from sklearn.datasets import load_breast_cancer
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, roc_auc_score

# 加载数据
X, y = load_breast_cancer(return_X_y=True)
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

# TabPFN：无需训练，直接预测！
clf = TabPFNClassifier(device='cpu', N_ensemble_configurations=32)
clf.fit(X_train, y_train)  # 只是存储数据，不做梯度下降
y_pred = clf.predict(X_test)
y_prob = clf.predict_proba(X_test)

print(f"Accuracy: {accuracy_score(y_test, y_pred):.4f}")
print(f"ROC-AUC:  {roc_auc_score(y_test, y_prob[:, 1]):.4f}")

# 与XGBoost对比
from xgboost import XGBClassifier

xgb = XGBClassifier(n_estimators=100, max_depth=6, random_state=42)
xgb.fit(X_train, y_train)
y_pred_xgb = xgb.predict(X_test)
print(f"XGBoost Accuracy: {accuracy_score(y_test, y_pred_xgb):.4f}")

# 可视化KAN学到的激活函数
# 假设KAN学习到 y = sin(x₁) + x₂²

#   x₁ 的激活函数:        x₂ 的激活函数:
#    ↑ φ(x₁)              ↑ φ(x₂)
#    │  /\                 │     /
#    │ /  \                │    /
#    │/    \    → sin(x₁)  │   /     → x₂²
#    │      \              │  /
#  --│-------\→ x₁       --│/--------→ x₂
#    │                     │

# pip install pykan
from kan import KAN
import torch

# 创建 KAN 模型：2层，输入4维，隐藏[8,4]，输出1维
model = KAN(
    width=[4, 8, 4, 1],       # 网络结构
    grid=5,                     # B-spline网格点数
    k=3,                        # B-spline阶数
    seed=42,
)

# 生成合成数据: y = sin(x₁) + x₂² + x₃·x₄
def target_fn(x):
    return torch.sin(x[:, 0]) + x[:, 1]**2 + x[:, 2] * x[:, 3]

X_train = torch.randn(1000, 4)
y_train = target_fn(X_train).unsqueeze(1)

X_test = torch.randn(200, 4)
y_test = target_fn(X_test).unsqueeze(1)

# 训练
dataset = {
    'train_input': X_train, 'train_label': y_train,
    'test_input': X_test, 'test_label': y_test,
}

results = model.fit(
    dataset,
    opt='LBFGS',
    steps=100,
    lamb=0.01,        # 正则化
    lamb_entropy=2.0, # 熵正则化（鼓励简洁函数）
)

# 可视化学到的激活函数
model.plot(beta=3)

# 符号回归（尝试用数学公式解释）
model.auto_symbolic(lib=['sin', 'x^2', 'x*y'])
formula = model.symbolic_formula()[0][0]
print(f"发现的公式: {formula}")

# pip install autogluon
from autogluon.tabular import TabularPredictor

# 一行代码完成建模
predictor = TabularPredictor(
    label='target',
    eval_metric='roc_auc',
    problem_type='binary',
).fit(
    train_data=train_df,
    time_limit=600,          # 最多训练10分钟
    presets='best_quality',  # 最优质量（也可选 'medium_quality', 'optimize_for_deployment'）
)

# 预测
predictions = predictor.predict(test_df)

# 查看模型排行榜
leaderboard = predictor.leaderboard(test_df)
print(leaderboard)

# 特征重要性
importance = predictor.feature_importance(test_df)

# pip install flaml
from flaml import AutoML

automl = AutoML()
automl.fit(
    X_train, y_train,
    task='classification',
    metric='accuracy',
    time_budget=120,         # 预算120秒
    estimator_list=[
        'lgbm', 'xgboost', 'catboost',
        'rf', 'extra_tree',
    ],
    eval_method='cv',        # 交叉验证
    n_splits=5,
)

print(f"Best model: {automl.best_estimator}")
print(f"Best config: {automl.best_config}")
print(f"Best score: {automl.best_loss:.4f}")

# 预测
y_pred = automl.predict(X_test)

# pip install pycaret
from pycaret.classification import *

# 初始化实验
s = setup(
    data=train_df,
    target='target',
    session_id=42,
    normalize=True,
    transformation=True,
    remove_multicollinearity=True,
    multicollinearity_threshold=0.95,
)

# 比较所有模型
best_model = compare_models(sort='AUC', n_select=1)

# 调优最佳模型
tuned_model = tune_model(best_model, optimize='AUC', n_iter=50)

# 集成
ensemble_model = ensemble_model(tuned_model, method='Bagging')

# 生成最终Pipeline
final_model = finalize_model(ensemble_model)

# 保存
save_model(final_model, 'best_pipeline')

"""
现代表格数据方法：全方位对比实战
数据集: Adult Income (UCI) - 预测收入是否>50K
"""
import numpy as np
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, roc_auc_score, f1_score
from sklearn.preprocessing import LabelEncoder
import time
import warnings
warnings.filterwarnings('ignore')

# === 数据准备 ===
from sklearn.datasets import fetch_openml

data = fetch_openml('adult', version=2, as_frame=True)
df = data.frame

# 分离特征和标签
target_col = 'income'  # >50K / <=50K
y = (df[target_col] == '>50K').astype(int)
X = df.drop(columns=[target_col])

# 识别数值和类别特征
num_cols = X.select_dtypes(include=['number']).columns.tolist()
cat_cols = X.select_dtypes(include=['category', 'object']).columns.tolist()

# 编码类别特征（用于GBDT等）
le_dict = {}
X_encoded = X.copy()
for col in cat_cols:
    le = LabelEncoder()
    X_encoded[col] = le.fit_transform(X[col].astype(str))
    le_dict[col] = le

X_train, X_test, y_train, y_test = train_test_split(
    X_encoded, y, test_size=0.2, random_state=42, stratify=y
)

results = {}

def evaluate(name, y_true, y_pred, y_prob, elapsed):
    """记录评估结果"""
    results[name] = {
        'Accuracy': accuracy_score(y_true, y_pred),
        'ROC-AUC': roc_auc_score(y_true, y_prob),
        'F1': f1_score(y_true, y_pred),
        'Time(s)': elapsed,
    }
    print(f"\n{'='*40}")
    print(f"Model: {name}")
    for k, v in results[name].items():
        print(f"  {k}: {v:.4f}")

# === 1. XGBoost ===
from xgboost import XGBClassifier

start = time.time()
xgb = XGBClassifier(
    n_estimators=500,
    max_depth=6,
    learning_rate=0.1,
    subsample=0.8,
    colsample_bytree=0.8,
    reg_alpha=0.1,
    reg_lambda=1.0,
    random_state=42,
    eval_metric='logloss',
    early_stopping_rounds=50,
    enable_categorical=True,
)
xgb.fit(
    X_train, y_train,
    eval_set=[(X_test, y_test)],
    verbose=False,
)
y_pred_xgb = xgb.predict(X_test)
y_prob_xgb = xgb.predict_proba(X_test)[:, 1]
elapsed = time.time() - start
evaluate('XGBoost', y_test, y_pred_xgb, y_prob_xgb, elapsed)

# === 2. LightGBM ===
import lightgbm as lgb

start = time.time()
lgbm = lgb.LGBMClassifier(
    n_estimators=500,
    max_depth=6,
    learning_rate=0.1,
    subsample=0.8,
    colsample_bytree=0.8,
    reg_alpha=0.1,
    reg_lambda=1.0,
    random_state=42,
    verbose=-1,
)
lgbm.fit(
    X_train, y_train,
    eval_set=[(X_test, y_test)],
    callbacks=[lgb.early_stopping(50), lgb.log_evaluation(0)],
)
y_pred_lgb = lgbm.predict(X_test)
y_prob_lgb = lgbm.predict_proba(X_test)[:, 1]
elapsed = time.time() - start
evaluate('LightGBM', y_test, y_pred_lgb, y_prob_lgb, elapsed)

# === 3. CatBoost ===
from catboost import CatBoostClassifier

start = time.time()
cat = CatBoostClassifier(
    iterations=500,
    depth=6,
    learning_rate=0.1,
    l2_leaf_reg=3.0,
    random_seed=42,
    verbose=0,
    eval_metric='AUC',
    early_stopping_rounds=50,
)
cat.fit(
    X_train, y_train,
    eval_set=(X_test, y_test),
)
y_pred_cat = cat.predict(X_test).astype(int)  # 链式调用，连续执行多个方法
y_prob_cat = cat.predict_proba(X_test)[:, 1]
elapsed = time.time() - start
evaluate('CatBoost', y_test, y_pred_cat, y_prob_cat, elapsed)

# === 4. TabPFN（小数据子集演示） ===
from tabpfn import TabPFNClassifier

# TabPFN对数据量有限制，这里取子集演示
n_sub = 3000
idx = np.random.RandomState(42).choice(len(X_train), n_sub, replace=False)
X_train_sub = X_train.iloc[idx].values.astype(np.float32)
y_train_sub = y_train.iloc[idx].values

start = time.time()
tabpfn = TabPFNClassifier(device='cpu', N_ensemble_configurations=16)
tabpfn.fit(X_train_sub, y_train_sub)
y_pred_pfn = tabpfn.predict(X_test.values.astype(np.float32))
y_prob_pfn = tabpfn.predict_proba(X_test.values.astype(np.float32))[:, 1]
elapsed = time.time() - start
evaluate('TabPFN (3k subset)', y_test, y_pred_pfn, y_prob_pfn, elapsed)

# === 5. FT-Transformer (简化版训练流程) ===
import torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

# 数据准备
X_train_tensor = torch.FloatTensor(X_train.values).to(device)
y_train_tensor = torch.LongTensor(y_train.values).to(device)
X_test_tensor = torch.FloatTensor(X_test.values).to(device)

train_ds = TensorDataset(X_train_tensor, y_train_tensor)
train_loader = DataLoader(train_ds, batch_size=512, shuffle=True)  # DataLoader批量加载数据，支持shuffle和多进程

# 使用之前定义的FTTransformer类（简化版，仅数值特征）
class SimpleFTTransformer(nn.Module):
    def __init__(self, n_features, d_token=128, n_layers=3, n_heads=8, dropout=0.1):
        super().__init__()
        self.feature_embeddings = nn.ModuleList([
            nn.Linear(1, d_token) for _ in range(n_features)
        ])
        self.cls_token = nn.Parameter(torch.randn(1, 1, d_token))

        encoder_layer = nn.TransformerEncoderLayer(
            d_model=d_token, nhead=n_heads,
            dim_feedforward=d_token * 4, dropout=dropout,
            activation='gelu', batch_first=True,
        )
        self.transformer = nn.TransformerEncoder(encoder_layer, n_layers)
        self.head = nn.Sequential(
            nn.LayerNorm(d_token), nn.ReLU(), nn.Linear(d_token, 2)
        )

    def forward(self, x):
        # 每个特征独立嵌入
        tokens = [emb(x[:, i:i+1]) for i, emb in enumerate(self.feature_embeddings)]
        tokens = torch.stack(tokens, dim=1)  # (B, n_features, d)
        cls = self.cls_token.expand(x.shape[0], -1, -1)
        tokens = torch.cat([cls, tokens], dim=1)
        tokens = self.transformer(tokens)
        return self.head(tokens[:, 0])

n_features = X_train.shape[1]
model = SimpleFTTransformer(n_features, d_token=128, n_layers=3).to(device)
optimizer = torch.optim.AdamW(model.parameters(), lr=1e-4, weight_decay=1e-5)
criterion = nn.CrossEntropyLoss()

# 训练
start = time.time()
model.train()  # train()开启训练模式
for epoch in range(30):
    total_loss = 0
    for xb, yb in train_loader:
        optimizer.zero_grad()  # 清零梯度，防止梯度累积
        logits = model(xb)
        loss = criterion(logits, yb)
        loss.backward()  # 反向传播计算梯度
        optimizer.step()  # 根据梯度更新模型参数
        total_loss += loss.item()  # .item()将单元素张量转为Python数值

# 评估
model.eval()
with torch.no_grad():  # 禁用梯度计算，节省内存
    logits = model(X_test_tensor)
    y_prob_ft = torch.softmax(logits, dim=1)[:, 1].cpu().numpy()
    y_pred_ft = logits.argmax(dim=1).cpu().numpy()

elapsed = time.time() - start
evaluate('FT-Transformer', y_test, y_pred_ft, y_prob_ft, elapsed)

# === 6. AutoGluon ===
from autogluon.tabular import TabularPredictor

# 准备DataFrame格式
train_ag = X_train.copy()
train_ag['target'] = y_train.values
test_ag = X_test.copy()
test_ag['target'] = y_test.values

start = time.time()
predictor = TabularPredictor(
    label='target',
    eval_metric='roc_auc',
    problem_type='binary',
    verbosity=1,
).fit(
    train_data=train_ag,
    time_limit=300,                    # 5分钟预算
    presets='best_quality',
)

y_pred_ag = predictor.predict(test_ag.drop(columns=['target']))
y_prob_ag = predictor.predict_proba(test_ag.drop(columns=['target']))[1]
elapsed = time.time() - start
evaluate('AutoGluon', y_test, y_pred_ag.values, y_prob_ag.values, elapsed)

# 查看排行榜
print("\n=== AutoGluon Leaderboard ===")
print(predictor.leaderboard(test_ag, silent=True).to_string())

# === 汇总对比 ===
results_df = pd.DataFrame(results).T
results_df = results_df.sort_values('ROC-AUC', ascending=False)

print("\n" + "=" * 60)
print("📊 全方法对比结果")
print("=" * 60)
print(results_df.to_string(float_format='{:.4f}'.format))

# 可视化
import matplotlib.pyplot as plt

fig, axes = plt.subplots(1, 3, figsize=(15, 5))

metrics = ['Accuracy', 'ROC-AUC', 'F1']
for ax, metric in zip(axes, metrics):  # zip按位置配对多个可迭代对象
    values = results_df[metric].sort_values(ascending=True)
    colors = ['#2ecc71' if v == values.max() else '#3498db' for v in values]  # 列表推导式，简洁创建列表
    values.plot(kind='barh', ax=ax, color=colors)
    ax.set_title(metric, fontsize=14)
    ax.set_xlim(values.min() - 0.02, values.max() + 0.01)

plt.tight_layout()
plt.savefig('tabular_comparison.png', dpi=150, bbox_inches='tight')
plt.show()