CatBoost：类别特征处理专家

import pandas as pd
import numpy as np

# 构造含高基数类别特征的模拟数据
np.random.seed(42)
n_samples = 10000
n_cities = 500

df = pd.DataFrame({
    "city": np.random.choice([f"city_{i}" for i in range(n_cities)], n_samples),
    "category": np.random.choice(["A", "B", "C", "D"], n_samples),
    "age": np.random.randint(18, 70, n_samples),
})
# 目标变量：与城市和类别相关
df["target"] = (
    df["city"].map({f"city_{i}": np.random.randn() for i in range(n_cities)})
    + df["category"].map({"A": 1.2, "B": 0.5, "C": -0.3, "D": -1.0})
    + np.random.randn(n_samples) * 0.5
)

from sklearn.model_selection import train_test_split
from sklearn.preprocessing import OneHotEncoder
import category_encoders as ce

X = df[["city", "category", "age"]]
y = df["target"]

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

# One-Hot 编码：维度爆炸
ohe = OneHotEncoder(sparse_output=False, handle_unknown="ignore")
X_train_ohe = ohe.fit_transform(X_train)
print(f"One-Hot 后维度: {X_train_ohe.shape[1]}")

# Target Encoding：简单但有泄露风险
te = ce.TargetEncoder(cols=["city", "category"])
X_train_te = te.fit_transform(X_train, y_train)
X_test_te = te.transform(X_test)
print(f"Target Encoding 后维度: {X_train_te.shape[1]}")

import numpy as np

class OrderedTargetEncoder:
    """从零实现有序目标编码——理解 CatBoost 核心机制"""

    def __init__(self, prior_count=1.0, prior_value=None):
        self.prior_count = prior_count  # 先验样本数（平滑参数）
        self.prior_value = prior_value  # 先验值（全局均值）

    def fit_transform(self, X_col, y):
        """对单列类别特征进行有序目标编码"""
        n = len(y)
        self.prior_value = np.mean(y)
        encoded = np.zeros(n, dtype=np.float64)

        # 累计统计量
        count_sum = {}   # 类别 -> 出现次数累计
        target_sum = {}  # 类别 -> 目标值累计

        for i in range(n):
            cat = X_col.iloc[i]
            if cat not in count_sum:
                count_sum[cat] = 0
                target_sum[cat] = 0.0

            count = count_sum[cat]
            t_sum = target_sum[cat]

            # 有序编码 = 累计均值 + 先验平滑
            encoded[i] = (t_sum + self.prior_value * self.prior_count) /                          (count + self.prior_count)

            # 更新累计统计量（在当前样本编码之后！）
            count_sum[cat] += 1
            target_sum[cat] += y.iloc[i]

        return encoded

# 使用示例
X_train_copy = X_train.copy()
encoder = OrderedTargetEncoder(prior_count=1.0)
X_train_copy["city_encoded"] = encoder.fit_transform(X_train["city"], y_train)

# CatBoost 内置的有序编码（生产环境应直接使用）
from catboost import CatBoostRegressor, Pool

# 声明类别特征列
cat_features = ["city", "category"]

train_pool = Pool(
    data=X_train,
    label=y_train,
    cat_features=cat_features
)

model = CatBoostRegressor(
    iterations=500,
    learning_rate=0.05,
    depth=6,
    # CatBoost 自动对 cat_features 使用有序目标编码
    verbose=100,
    random_seed=42
)
model.fit(train_pool)

# 评估
test_pool = Pool(data=X_test, cat_features=cat_features)
preds = model.predict(test_pool)
mse = np.mean((preds - y_test) ** 2)
print(f"测试集 MSE: {mse:.4f}")
print(f"测试集 RMSE: {np.sqrt(mse):.4f}")

# 对比传统 Boosting vs 有序 Boosting 的残差计算
import numpy as np

class TraditionalBoostingResidual:
    """传统 Boosting：全量训练集计算残差"""

    def __init__(self, n_estimators=10):
        self.n_estimators = n_estimators

    def compute_residuals(self, y_true, y_preds_history):
        """第 t 轮残差 = y - 前 t-1 棵树的累积预测（使用全部训练数据）"""
        residuals_per_tree = []
        cumulative_pred = np.zeros(len(y_true))
        for t in range(self.n_estimators):
            # 传统方式：每棵树用全局累积预测计算残差
            residuals = y_true - cumulative_pred
            residuals_per_tree.append(residuals)
            # 模拟当前树的预测（实际由树模型给出）
            tree_pred = np.random.randn(len(y_true)) * 0.3
            cumulative_pred += tree_pred
        return residuals_per_tree

class OrderedBoostingResidual:
    """有序 Boosting：每个样本只用前面样本的信息"""

    def __init__(self, n_estimators=10):
        self.n_estimators = n_estimators

    def compute_residuals(self, y_true, permutation):
        n = len(y_true)
        residuals = np.zeros((self.n_estimators, n))

        for t in range(self.n_estimators):
            pred_before = np.zeros(n)
            for idx in range(n):
                pos = permutation[idx]
                if idx > 0:
                    # 只用前面样本的预测均值
                    pred_before[pos] = np.mean(pred_before[:idx])
            residuals[t] = y_true - pred_before
        return residuals

# CatBoost 的 boosting_type 参数对比
from catboost import CatBoostRegressor
from sklearn.metrics import mean_squared_error

train_pool = Pool(X_train, y_train, cat_features=["city", "category"])
test_pool = Pool(X_test, y_test, cat_features=["city", "category"])

# 默认：有序 Boosting（最安全）
model_ordered = CatBoostRegressor(
    iterations=300,
    learning_rate=0.03,
    depth=6,
    boosting_type="Ordered",  # 显式指定
    verbose=False,
    random_seed=42
)
model_ordered.fit(train_pool)

# Plain 模式（类似 XGBoost，更快但可能泄露）
model_plain = CatBoostRegressor(
    iterations=300,
    learning_rate=0.03,
    depth=6,
    boosting_type="Plain",
    verbose=False,
    random_seed=42
)
model_plain.fit(train_pool)

# 对比训练集和测试集性能，观察过拟合程度
for name, model in [("Ordered", model_ordered), ("Plain", model_plain)]:
    train_pred = model.predict(train_pool)
    test_pred = model.predict(test_pool)
    train_mse = mean_squared_error(y_train, train_pred)
    test_mse = mean_squared_error(y_test, test_pred)
    print(f"{name}: Train MSE={train_mse:.4f}, Test MSE={test_mse:.4f}, "
          f"Gap={test_mse - train_mse:.4f}")

import numpy as np

class SymmetricTreeDemo:
    """对称树 vs 非对称树的结构对比"""

    def __init__(self, depth=3):
        self.depth = depth

    def show_symmetric_structure(self):
        """对称树：每层同一分裂规则"""
        print("对称树结构（Oblivious Tree）:")
        for level in range(self.depth):
            n_nodes = 2  level
            # 该层所有节点使用相同特征和阈值
            feature = f"f_{level % 5}"
            threshold = round(np.random.uniform(-1, 1), 2)
            print(f"  Level {level}: {n_nodes} 个节点, "
                  f"分裂: {feature} <= {threshold}")
        print(f"  总分裂规则数: {self.depth}")
        print(f"  叶子节点数: {2  self.depth}")

    def show_asymmetric_structure(self):
        """非对称树：每个节点独立分裂"""
        print("\n非对称树结构（XGBoost/LightGBM）:")
        total_splits = 0
        for level in range(self.depth):
            n_nodes = 2  level
            for node in range(n_nodes):
                feature = f"f_{np.random.randint(0, 10)}"
                threshold = round(np.random.uniform(-1, 1), 2)
                print(f"  Level {level}, Node {node}: "
                      f"{feature} <= {threshold}")
                total_splits += 1
        print(f"  总分裂规则数: {total_splits}")
        print(f"  叶子节点数: 不确定（取决于剪枝）")

tree = SymmetricTreeDemo(depth=3)
tree.show_symmetric_structure()
tree.show_asymmetric_structure()

# CatBoost 中的对称树配置
from catboost import CatBoostClassifier

model = CatBoostClassifier(
    iterations=1000,
    learning_rate=0.03,
    # depth 控制对称树的深度（默认 6，范围 1-16）
    depth=6,
    # l2_leaf_reg 控制 L2 正则化强度
    l2_leaf_reg=3.0,
    # CatBoost 默认就是对称树，无法关闭
    # 这是与 XGBoost/LightGBM 的根本区别之一
    cat_features=["city", "category"],
    verbose=False,
    random_seed=42
)
model.fit(train_pool)

# 查看树的结构信息
print(f"树数量: {model.tree_count_}")
print(f"树深度: {model.get_param('depth')}")
print(f"L2 正则化: {model.get_param('l2_leaf_reg')}")

# 导出单棵树的可视化
# model.plot_tree(tree_idx=0, pool=train_pool)

# 三框架对比实验
import time
from catboost import CatBoostRegressor
from sklearn.metrics import mean_squared_error, r2_score

cat_features = ["city", "category"]

# ========== CatBoost ==========
t0 = time.time()
cb = CatBoostRegressor(
    iterations=500, learning_rate=0.05, depth=6,
    cat_features=cat_features, verbose=False, random_seed=42
)
cb.fit(X_train, y_train)
cb_time = time.time() - t0
cb_pred = cb.predict(X_test)
cb_mse = mean_squared_error(y_test, cb_pred)
cb_r2 = r2_score(y_test, cb_pred)

# ========== XGBoost（需手动编码） ==========
from xgboost import XGBRegressor
from sklearn.preprocessing import LabelEncoder

X_train_xgb = X_train.copy()
X_test_xgb = X_test.copy()
for col in cat_features:
    le = LabelEncoder()
    X_train_xgb[col] = le.fit_transform(X_train_xgb[col].astype(str))
    X_test_xgb[col] = le.transform(X_test_xgb[col].astype(str))

t0 = time.time()
xgb = XGBRegressor(
    n_estimators=500, learning_rate=0.05, max_depth=6,
    random_state=42, verbosity=0
)
xgb.fit(X_train_xgb, y_train)
xgb_time = time.time() - t0
xgb_pred = xgb.predict(X_test_xgb)
xgb_mse = mean_squared_error(y_test, xgb_pred)
xgb_r2 = r2_score(y_test, xgb_pred)

# ========== LightGBM ==========
from lightgbm import LGBMRegressor

X_train_lgb = X_train.copy()
X_test_lgb = X_test.copy()
for col in cat_features:
    X_train_lgb[col] = X_train_lgb[col].astype("category")
    X_test_lgb[col] = X_test_lgb[col].astype("category")

t0 = time.time()
lgb = LGBMRegressor(
    n_estimators=500, learning_rate=0.05, max_depth=6,
    random_state=42, verbose=-1
)
lgb.fit(X_train_lgb, y_train)
lgb_time = time.time() - t0
lgb_pred = lgb.predict(X_test_lgb)
lgb_mse = mean_squared_error(y_test, lgb_pred)
lgb_r2 = r2_score(y_test, lgb_pred)

print(f"{'模型':<12} {'时间(s)':>8} {'MSE':>10} {'R2':>8}")
print(f"{'CatBoost':<12} {cb_time:>8.2f} {cb_mse:>10.4f} {cb_r2:>8.4f}")
print(f"{'XGBoost':<12} {xgb_time:>8.2f} {xgb_mse:>10.4f} {xgb_r2:>8.4f}")
print(f"{'LightGBM':<12} {lgb_time:>8.2f} {lgb_mse:>10.4f} {lgb_r2:>8.4f}")

# 特征重要性分析对比
import matplotlib.pyplot as plt

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

# CatBoost 特征重要性
importances_cb = cb.get_feature_importance(train_pool)
feature_names = X_train.columns.tolist()
sorted_idx = np.argsort(importances_cb)
axes[0].barh([feature_names[i] for i in sorted_idx],
             importances_cb[sorted_idx])
axes[0].set_title("CatBoost Feature Importance")

# XGBoost 特征重要性
importances_xgb = xgb.feature_importances_
sorted_idx = np.argsort(importances_xgb)
axes[1].barh(feature_names, importances_xgb)
axes[1].set_title("XGBoost Feature Importance")

# LightGBM 特征重要性
importances_lgb = lgb.feature_importances_
sorted_idx = np.argsort(importances_lgb)
axes[2].barh(feature_names, importances_lgb)
axes[2].set_title("LightGBM Feature Importance")

plt.tight_layout()
plt.savefig("feature_importance_comparison.png", dpi=150)
plt.show()

from catboost import CatBoostRegressor, cv, Pool

train_pool = Pool(X_train, y_train, cat_features=["city", "category"])

# ========== 策略一：Early Stopping ==========
model_es = CatBoostRegressor(
    iterations=5000,           # 设一个很大的值
    learning_rate=0.03,
    depth=6,
    l2_leaf_reg=3.0,
    random_strength=1.0,
    early_stopping_rounds=50,  # 连续 50 轮无改善则停止
    eval_metric="RMSE",
    verbose=100,
    random_seed=42
)
model_es.fit(
    train_pool,
    eval_set=(X_test, y_test),  # 用验证集监控
    cat_features=["city", "category"]
)
print(f"最优迭代次数: {model_es.get_best_iteration()}")
print(f"最优验证 RMSE: {model_es.get_best_score()['validation']['RMSE']:.4f}")

# ========== 策略二：内置交叉验证 ==========
from catboost import cv

params = {
    "iterations": 300,
    "learning_rate": 0.05,
    "depth": 6,
    "l2_leaf_reg": 3.0,
    "loss_function": "RMSE",
    "random_seed": 42,
    "verbose": False
}

# CatBoost 原生 CV（支持 GPU 加速）
cv_data = cv(
    Pool(X_train, y_train, cat_features=["city", "category"]),
    params,
    fold_count=5,
    plot=False,
    stratified=False
)

# 输出每折的结果
print("CV 结果:")
for metric in cv_data.columns:
    mean_val = cv_data[metric].mean()
    std_val = cv_data[metric].std()
    print(f"  {metric}: {mean_val:.4f} +/- {std_val:.4f}")

# ========== 策略三：网格搜索 ==========
from catboost import CatBoostRegressor

grid = {
    "depth": [4, 6, 8],
    "learning_rate": [0.01, 0.05, 0.1],
    "l2_leaf_reg": [1, 3, 5],
    "random_strength": [0.5, 1.0, 2.0],
}

model_grid = CatBoostRegressor(
    iterations=300,
    cat_features=["city", "category"],
    verbose=False,
    random_seed=42
)
model_grid.grid_search(
    grid,
    train_pool,
    cv=3,
    plot=False,
    verbose=False
)
print(f"最优参数: {model_grid.get_params()}")

import pandas as pd
import numpy as np
from catboost import CatBoostClassifier, Pool, cv
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report, roc_auc_score
import matplotlib.pyplot as plt

# ========== Step 1: 数据准备 ==========
np.random.seed(42)
n = 20000
data = pd.DataFrame({
    "age": np.random.randint(18, 70, n),
    "income": np.random.lognormal(10, 0.5, n),
    "usage_freq": np.random.poisson(10, n),
    "days_since_active": np.random.exponential(30, n),
    "city": np.random.choice([f"city_{i}" for i in range(200)], n),
    "product_type": np.random.choice(["Basic", "Pro", "Enterprise"], n,
                                      p=[0.6, 0.3, 0.1]),
    "subscription": np.random.choice(["Monthly", "Yearly", "Lifetime"], n),
    "acquisition_channel": np.random.choice(
        ["Organic", "Paid", "Referral", "Social"], n,
        p=[0.4, 0.3, 0.2, 0.1]
    ),
})

# 目标变量：流失概率（与城市、产品类别、活跃度相关）
churn_prob = (
    1 / (1 + np.exp(-(
        -2.0
        + 0.02 * data["days_since_active"]
        - 0.03 * data["usage_freq"]
        + data["product_type"].map({"Basic": 0.8, "Pro": 0.0, "Enterprise": -0.5})
        + data["subscription"].map({"Monthly": 0.5, "Yearly": -0.3, "Lifetime": -0.8})
        + np.random.randn(n) * 0.3
    )))
)
data["churn"] = (np.random.rand(n) < churn_prob).astype(int)
print(f"流失率: {data['churn'].mean():.2%}")

# ========== Step 2: 模型训练 ==========
features = ["age", "income", "usage_freq", "days_since_active",
            "city", "product_type", "subscription", "acquisition_channel"]
cat_features = ["city", "product_type", "subscription", "acquisition_channel"]

X = data[features]
y = data["churn"]

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

train_pool = Pool(X_train, y_train, cat_features=cat_features)
test_pool = Pool(X_test, y_test, cat_features=cat_features)

model = CatBoostClassifier(
    iterations=1000,
    learning_rate=0.05,
    depth=6,
    l2_leaf_reg=3.0,
    random_strength=1.0,
    early_stopping_rounds=50,
    eval_metric="AUC",
    auto_class_weights="Balanced",
    verbose=200,
    random_seed=42
)
model.fit(train_pool, eval_set=test_pool)

# ========== Step 3: 评估 ==========
y_pred = model.predict(test_pool)
y_prob = model.predict_proba(test_pool)[:, 1]

print("\n=== 分类报告 ===")
print(classification_report(y_test, y_pred))
print(f"ROC-AUC: {roc_auc_score(y_test, y_prob):.4f}")
print(f"最优迭代: {model.get_best_iteration()}")

# ========== Step 4: 特征重要性 ==========
importances = model.get_feature_importance(train_pool)
feat_imp = pd.DataFrame({
    "feature": features,
    "importance": importances
}).sort_values("importance", ascending=False)

print("\n=== 特征重要性 Top 10 ===")
print(feat_imp.head(10).to_string(index=False))

plt.figure(figsize=(10, 6))
plt.barh(feat_imp["feature"], feat_imp["importance"])
plt.gca().invert_yaxis()
plt.xlabel("Feature Importance")
plt.title("CatBoost Feature Importance")
plt.tight_layout()
plt.savefig("catboost_feature_importance.png", dpi=150)

# ========== Step 5: 模型导出 ==========
model.save_model("churn_model.cbm")
model.save_model("churn_model.onnx", format="onnx")
print("\n模型已导出: churn_model.cbm (原生) / churn_model.onnx (通用)")