变量分箱与自然分组实践总结

在数据分析与建模中，我们经常遇到两类变量处理难题：

• 连续变量（如距离、时长、金额、温度）：取值范围连续，直接使用可能导致模型过拟合，且难以输出可解释的业务结论。例如”距离7km”不如”中途（15-30km）”直观。
• 高基数类别变量（如小时、地理编码、用户ID前缀、线路编号）：类别数从几十到上万不等，直接 one-hot 编码会产生极度稀疏的特征矩阵，且无法体现类别间的”相似性”。

变量分箱（Binning） 的核心目标是将这两类变量转化为有限的、有意义的分组，使得：

• 组内同质、组间异质— 同一组内的样本在目标指标上表现相近，不同组之间有显著差异
• 可解释性— 分组结果可以用业务语言描述（如”短途/中途/长途”、”高峰/平峰/夜间”）
• 特征工程— 为下游模型提供高质量的分箱特征，减少噪声干扰

本文介绍两种互补的分箱方法，分别针对连续变量和类别变量：

决策树分箱（连续变量）

决策树分箱利用了决策树天然的最优分裂特性：决策树在训练时会自动寻找使信息增益最大的分裂点，因此树的结构天然定义了一组分箱边界。具体做法是：将连续变量作为唯一特征、目标变量作为标签训练决策树，然后从训练好的树结构中提取分裂阈值，这些阈值排序后即构成分箱边界。

传统等频/等距分箱不考虑目标变量，分组边界是”盲目的”。决策树分箱是有监督的——它直接以目标变量为优化目标，自动找到”目标率发生显著变化”的切分点。例如对于距离变量，如果目标率在 5.5km 和 30.5km 处发生跳跃，决策树会自动在这些位置分裂，而等距分箱可能会把关键的 5.5km 切分点劈成两半。

关键参数

步骤详解：

• 遍历 max_leaf_nodes：在[min_bins, max_bins] 范围内，每个值训练一棵 DecisionTreeClassifier
• 控制叶节点最小样本：min_samples_leaf=0.01保证每个分箱至少包含 1% 的样本，避免出现只有几条样本的”碎片箱”
• 交叉验证评估：对每个max_leaf_nodes，用 K 折交叉验证计算验证集 AUC（均值 ± 标准差）
• 选择最优分组数：按strategy 参数（cv / elbow / one_se / train）选择最优 max_leaf_nodes
• 提取分裂阈值：从决策树的threshold 属性直接提取分裂点
• 构建分箱边界：阈值排序后加上 ±∞ 构成完整边界，用cut 分配标签

分箱边界构建示例

边界 = [-inf] + [排序后的分裂阈值] + [+inf]
 
示例 (距离变量, n=5):
  [-inf, 5.5, 15.5, 30.5, 60.0, +inf]
 
  箱0: [0,    5.5)   → 短途
  箱1: [5.5,  15.5)  → 中短途
  箱2: [15.5, 30.5)  → 中途
  箱3: [30.5, 60.0)  → 长途
  箱4: [60.0, +inf)  → 超长途

设计亮点：从决策树 tree_.threshold 属性直接提取分裂点，保证分箱边界与模型决策完全一致。这不是”近似”决策树的分裂，而是精确提取。

分组数选择策略：四种策略深度对比

使用中遇到的最大问题：训练集 AUC 永远随 max_leaf_nodes 单调递增。直接选 train_AUC 最大 = 永远选最大分组数，失去选择意义。

这是一个经典的偏差-方差权衡问题：分组数越多，模型对训练数据的拟合越好（偏差降低），但泛化能力可能下降（方差升高）。我们需要交叉验证来评估泛化性能。

四种策略对比

策略选择逻辑：

• one_se在防过拟合、简洁性、鲁棒性三个维度上表现均衡，是综合最优选择
• cv性能最优但简洁性不足，且计算成本最高
• train速度最快但防过拟合能力极差，不推荐生产使用
• elbow简洁性突出但性能和鲁棒性略逊于 one_se

cv 策略的自动回退机制

代码实现中有一个重要的改进：当 cv 策略选中 max_bins（即 CV-AUC 仍在单调递增）时，会自动应用 one-se’ 规则回退到更简洁的分组数。这避免了”CV-AUC 还在涨但差异已在噪声范围内”时无意义地选择最大分组数。

# cv 策略选中 max_bins 时自动应用 1-SE 回退
if best_n == self.max_bins and best_idx > 0:
    best_score = scores[best_idx]
    best_std = stds[best_idx]
    threshold = best_score - best_std
    for i in range(len(n_list)):
        if scores[i] >= threshold and n_list[i] < best_n:
            # 回退到更简洁的分组数
            return n_list[i]

优缺点分析

优点：

• 有监督：直接以目标变量为优化目标，分箱结果具有业务区分度
• 自动选点：无需人工指定切分点，决策树自动寻找最优分裂阈值
• 交叉验证防过拟合：用验证集 AUC 替代训练集 AUC
• 多策略可选：cv / elbow / one_se 三种策略适应不同业务需求
• 稳定性：min_samples_leaf保证每箱有足够样本

缺点：

• 计算成本：cv 策略需要训练 K 倍模型，大数据集下较慢
• 边界可能不直观：分裂点可能是非整数（如5、17.5）
• 仅适用连续变量：对类别型变量不适用
• 单变量：每次只考虑一个变量的分裂，不捕捉交互效应

关键知识点：1-SE 规则详解

1-SE 规则（One Standard Error Rule）是统计学中经典的模型选择方法，源自 Hastie & Tibshirani 的《The Elements of Statistical Learning》。它的核心思想是：在最优性能的一个标准差范围内，选择最简单的模型。

图解 1-SE 规则

执行步骤

• 找到 CV-AUC 最大的分组数best_idx（图中 n=6）
• 计算阈值threshold = best_score – best_std（图中橙色虚线）
• 在所有 CV-AUC ≥ threshold 的候选中，选分组数最少的（图中 n=4）

为什么有效

奥卡姆剃刀原则：如果两个模型的性能差异在统计噪声范围内（即差异小于一个标准差），那么它们在统计上是”等价”的。此时应选择更简单的模型——更少的分组数意味着更好的可解释性、更低的过拟合风险、更稳定的分箱边界。

与肘部法则的对比

Jenks 自然断点（类别变量）

Jenks 自然断点适用于高基数类别变量的分组，典型场景包括：

• 时间维度：小时（0-23）、星期、月份
• 地理维度：区域编码、邮政编码、地理网格
• 业务维度：线路编号、商品类目、渠道编号

核心前提：先按类别聚合计算每个类别的目标率（如转化率、点击率、流失率），再对”率值”进行一维聚类分组。

Jenks Natural Breaks 是一种一维聚类算法，通过最小化组内方差（SDCM）、最大化组间方差，寻找数据中的”自然断点”。它与 K-Means 的区别在于：K-Means 是多维聚类，而 Jenks 专门针对一维数据优化，算法效率更高。

两阶段方法：

• 阶段 1（聚合）：groupby(category).agg(cnt, rate)— 计算每个类别的样本量和目标率
• 阶段 2（分组）：对率值序列应用 Jenks 算法，将率值相近的类别归为同一组

关键参数

执行流程

• 按类别聚合：groupby(category).agg(cnt=列数, rate=目标均值)
• 低频过滤：样本量 <min_samples 的类别标记为 -1，不参与 Jenks 计算
• 遍历分组数：对有效类别的率值，遍历n_groups ∈ [min_groups, max_groups]
• 计算 GVF：每次用jenks_breaks 计算断点，然后计算 GVF
• 选择最优 n：按 elbow 或 max_gvf 策略选择最优分组数
• 分配组别：用最优断点将类别分配到各组，计算各组统计

优缺点分析

优点：

• 无监督但有效：不需目标变量参与建模，仅基于率值分布寻找自然断点
• 适应性强：适用于任意高基数类别变量
• 可解释：GVF 指标直观反映分组质量
• 低频保护：通过min_samples 过滤低频类别，避免小样本噪声干扰

缺点：

• 两阶段：先聚合再分组，聚合阶段丢失了个体层面信息
• GVF 单调递增：分组数越多 GVF 越高，需依赖肘部法则或人工限定上限
• 计算复杂度：Jenks 算法为 O(n²)，类别数 >10000 需采样
• 低频类别处理粗糙：统一标记为 -1，丢失了低频类别之间的差异

关键知识点：GVF 指标详解

GVF（Goodness of Variance Fit，方差拟合优度）是衡量 Jenks 分组质量的核心指标，取值范围 [0, 1]，越接近 1 表示分组越好。

计算公式

SDAM (Sum of Squared Deviations from Array Mean) = Σ(rate_i - mean_all)²
SDCM (Sum of Squared Deviations from Class Means) = Σ_g Σ(rate_i - mean_g)²
 
GVF = (SDAM - SDCM) / SDAM

图解 GVF

• 左图（分组前）：所有类别的目标率散布在全局均值周围，偏差（红色线段）较大，SDAM 较高。
• 右图（分组后）：类别被分为 3 组，每组内的偏差（彩色线段）明显减小，SDCM 较低。组内方差减小越多，GVF 越接近 1。

GVF 的单调递增问题

重要陷阱：GVF 随分组数单调递增。分组数越多，组内方差必然越小（极端情况下每个类别自成一组，GVF = 1）。

这意味着”选 GVF 最大的分组数”等价于”选最大分组数”，失去选择意义。因此代码中提供了两种应对策略：

• 肘部法则（推荐）：寻找 GVF 增量显著下降的拐点，增量低于均值 50% 时停止
• SE 规则：在 GVF 最优值的一个增量标准差范围内，选最简分组数

GVF 与 AUC 的本质区别

低频过滤机制：统计稳定性保障

为什么需要过滤低频类别

低频过滤是 Jenks 分组中一个看似简单但至关重要的环节：

• 小样本不稳定：某类别只有 3 条样本，目标率可能是 0% 或 100%，纯属噪声
• 干扰断点计算：低频类别的极端率值会拉偏 Jenks 断点位置
• 不满足大数定律：样本量不足时，统计量无法反映真实业务规律

图解低频过滤

左图展示了所有类别的样本量分布，红色柱表示低于 MIN_SAMPLES 阈值的低频类别。右图以散点图形式展示过滤效果：蓝色点为有效类别（参与 Jenks 计算），红色点为低频类别（标记为 -1）。

处理方式

有效类别 (cnt >= MIN_SAMPLES)
  → 参与正常分组，获得组号 0, 1, 2, ...
 
低频类别 (cnt < MIN_SAMPLES)
  → 统一标记为 -1（特殊组）
  → 不参与 Jenks 断点计算

设计考量：低频类别不直接丢弃，而是标记为 -1 单独管理，保证后续 transform 时不会遗漏。在实际应用中，-1 组可以后续单独处理或合并到最接近的组。

MIN_SAMPLES 参数建议

代码实现与使用示例

完整源码

binning.py

"""
变量分箱与自然分组工具模块
============================
提供两种核心分箱/分组方法：
  1. DecisionTreeBinner   — 决策树分箱（连续变量，AUC 优化）
  2. JenksBreaksGrouper    — Jenks 自然断点分组（类别变量，GVF 优化）
  3. BinningPipeline       — 多变量分箱流水线

设计原则：
  - 单一职责：每个类只负责一种分箱逻辑
  - 统一接口：所有分箱器均实现 fit / transform / fit_transform
  - 结果对象：BinningResult 封装边界、统计表、映射字典
  - 可复用可测试：纯函数式计算，无副作用
"""

from __future__ import annotations

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Optional

import numpy as np
import pandas as pd
from sklearn.tree import DecisionTreeClassifier
from sklearn.metrics import roc_auc_score
from sklearn.model_selection import StratifiedKFold
import jenkspy


# ============================================================
# 数据结构：分箱结果
# ============================================================
@dataclass
class BinningResult:
    """封装分箱/分组结果的不可变容器。"""

    col_name: str
    boundaries: list                # 分箱边界（含 ±inf）或 Jenks 断点
    stats: pd.DataFrame             # 各箱/组统计表
    label_map: dict                 # 原始值 -> 组标签映射（Jenks 用）
    meta: dict = field(default_factory=dict)   # 附加信息（最优参数、GVF/AUC 等）

    def __repr__(self) -> str:
        return (
            f"BinningResult(col='{self.col_name}', "
            f"n_bins={len(self.boundaries)-1}, "
            f"meta={self.meta})"
        )

    def export_mapping(self) -> pd.DataFrame:
        """
        导出组号-类别映射表。

        - 决策树分箱：返回边界区间表（组号, 区间, 样本量, 目标率）
        - Jenks 分组：返回逐类别映射表（组号, 类别, 样本量, 目标率(%)）

        Returns
        -------
        pd.DataFrame
        """
        method = self.meta.get("method", "")

        if method == "decision_tree":
            # 决策树：导出边界区间
            df = self.stats.copy()
            df.insert(0, "组号", range(len(df)))
            return df

        elif method == "jenks":
            # Jenks：导出逐类别映射
            cat_stats = self.meta.get("category_stats")
            if cat_stats is not None:
                return cat_stats[["组号", "类别", "样本量", "目标率(%)"]].copy()
            # 降级：从 label_map 构建
            rows = [{"组号": v, "类别": k} for k, v in self.label_map.items()]
            return pd.DataFrame(rows).sort_values("组号").reset_index(drop=True)

        return pd.DataFrame()


# ============================================================
# 抽象基类
# ============================================================
class BaseBinner(ABC):
    """所有分箱器的抽象基类，定义统一接口。"""

    @abstractmethod
    def fit(self, df: pd.DataFrame, col: str, target: str = "is_suc") -> "BaseBinner":
        """拟合分箱器，学习分箱边界。"""
        ...

    @abstractmethod
    def transform(self, df: pd.DataFrame, col: str) -> pd.Series:
        """将原始值映射为分箱标签。"""
        ...

    def fit_transform(
        self, df: pd.DataFrame, col: str, target: str = "is_suc"
    ) -> tuple[pd.Series, BinningResult]:
        """拟合 + 转换，返回标签列和结果对象。"""
        self.fit(df, col, target)
        labels = self.transform(df, col)
        return labels, self.result_


# ============================================================
# 1. 决策树分箱器（连续变量）
# ============================================================
class DecisionTreeBinner(BaseBinner):
    """
    基于 sklearn 决策树的连续变量分箱。

    原理：遍历不同 max_leaf_nodes，训练决策树，通过交叉验证评估泛化性能，
         选择最优分箱数，从树结构中提取分裂阈值作为分箱边界。

    分组数选择策略（strategy 参数）
    ------------------------------
    - ``cv``     : 交叉验证 AUC 最大。用 K 折验证集 AUC 替代训练集 AUC，
                   避免过拟合导致的"AUC 随分组数单调递增"问题。
                   **改进**：当选中 max_bins（CV-AUC 仍在单调递增）时，自动应用
                   1-SE 规则回退到更简洁的分组数，避免无意义的最大值选择。
    - ``elbow``  : 肘部法则。在 CV-AUC 增量序列上找拐点，增量低于均值一半时停止。
    - ``one_se`` : 1-SE 规则。在 CV-AUC 最优值的一个标准差范围内，选最简单的模型
                   （最少分组数）。统计学经典简洁模型选择法（推荐）。
    - ``train``  : 训练集 AUC 最大（原始行为，不推荐，容易过拟合）。

    Parameters
    ----------
    min_bins : int            最少分箱数（默认 3）
    max_bins : int            最多分箱数（默认 7）
    min_samples_leaf : float  每箱最小样本比例（默认 0.01 = 1%）
    cv_folds : int            交叉验证折数（默认 5），strategy='train' 时忽略
    strategy : str            分组数选择策略（默认 'one_se'）
    random_state : int        随机种子
    """

    def __init__(
        self,
        min_bins: int = 3,
        max_bins: int = 7,
        min_samples_leaf: float = 0.01,
        cv_folds: int = 5,
        strategy: str = "one_se",
        random_state: int = 42,
    ):
        self.min_bins = min_bins
        self.max_bins = max_bins
        self.min_samples_leaf = min_samples_leaf
        self.cv_folds = cv_folds
        self.strategy = strategy
        self.random_state = random_state
        self.result_: Optional[BinningResult] = None

    # ---- 核心逻辑 ----
    def fit(self, df: pd.DataFrame, col: str, target: str = "is_suc") -> "DecisionTreeBinner":
        mask = df[[col, target]].notna().all(axis=1)
        X = df.loc[mask, [col]]
        y = df.loc[mask, target]

        # 遍历分组数，收集指标
        records = []
        models = {}

        for n_leaves in range(self.min_bins, self.max_bins + 1):
            dt = DecisionTreeClassifier(
                max_leaf_nodes=n_leaves,
                min_samples_leaf=self.min_samples_leaf,
                random_state=self.random_state,
            )
            dt.fit(X, y)
            train_auc = roc_auc_score(y, dt.predict_proba(X)[:, 1])
            models[n_leaves] = dt

            if self.strategy == "train":
                print(f"  {col} | max_leaf_nodes={n_leaves} | train_AUC={train_auc:.6f}")
                records.append({
                    "n_leaves": n_leaves,
                    "train_auc": train_auc,
                    "cv_auc": None,
                    "cv_std": 0.0,
                })
            else:
                cv_auc, cv_std = self._cv_evaluate(X, y, n_leaves)
                print(f"  {col} | max_leaf_nodes={n_leaves} | "
                      f"train_AUC={train_auc:.6f} | cv_AUC={cv_auc:.6f} ± {cv_std:.4f}")
                records.append({
                    "n_leaves": n_leaves,
                    "train_auc": train_auc,
                    "cv_auc": cv_auc,
                    "cv_std": cv_std,
                })

        # 选择最优分组数
        best_n = self._select_best_n(records)
        best_model = models[best_n]
        best_record = next(r for r in records if r["n_leaves"] == best_n)
        best_cv_val = best_record["cv_auc"] if best_record["cv_auc"] is not None else best_record["train_auc"]
        print(f"  {col} | 最优 max_leaf_nodes={best_n} "
              f"(strategy={self.strategy}, "
              f"cv_AUC={best_cv_val:.6f})")

        # 从树结构提取分裂阈值
        tree = best_model.tree_
        thresholds = tree.threshold[tree.children_left != -1]
        boundaries = sorted([-np.inf] + thresholds.tolist() + [np.inf])

        # 构建统计表
        stats = self._build_stats(df, col, target, boundaries)

        self.result_ = BinningResult(
            col_name=col,
            boundaries=boundaries,
            stats=stats,
            label_map={},
            meta={
                "method": "decision_tree",
                "strategy": self.strategy,
                "best_n_leaves": best_n,
                "best_train_auc": best_record["train_auc"],
                "best_cv_auc": best_record["cv_auc"],
                "best_cv_std": best_record["cv_std"],
                "min_samples_leaf": self.min_samples_leaf,
                "cv_folds": self.cv_folds,
                "all_records": records,
            },
        )
        return self

    def transform(self, df: pd.DataFrame, col: str) -> pd.Series:
        if self.result_ is None:
            raise RuntimeError("请先调用 fit()")
        return pd.cut(df[col], bins=self.result_.boundaries, right=False)

    # ---- 私有方法 ----
    def _cv_evaluate(self, X: pd.DataFrame, y: pd.Series, n_leaves: int) -> tuple:
        """K 折交叉验证评估某分组数的验证集 AUC。"""
        skf = StratifiedKFold(
            n_splits=self.cv_folds,
            shuffle=True,
            random_state=self.random_state,
        )
        aucs = []
        for train_idx, val_idx in skf.split(X, y):
            X_tr, X_val = X.iloc[train_idx], X.iloc[val_idx]
            y_tr, y_val = y.iloc[train_idx], y.iloc[val_idx]
            dt = DecisionTreeClassifier(
                max_leaf_nodes=n_leaves,
                min_samples_leaf=self.min_samples_leaf,
                random_state=self.random_state,
            )
            dt.fit(X_tr, y_tr)
            proba = dt.predict_proba(X_val)[:, 1]
            # 单类别 fold 保护
            if len(np.unique(y_val)) < 2:
                continue
            aucs.append(roc_auc_score(y_val, proba))
        if not aucs:
            # 所有 fold 都是单类别的极端情况
            return roc_auc_score(y, DecisionTreeClassifier(
                max_leaf_nodes=n_leaves,
                min_samples_leaf=self.min_samples_leaf,
                random_state=self.random_state,
            ).fit(X, y).predict_proba(X)[:, 1]), 0.0
        return float(np.mean(aucs)), float(np.std(aucs))

    def _select_best_n(self, records: list) -> int:
        """
        按策略选择最优分组数。

        - train  : 训练集 AUC 最大
        - cv     : CV-AUC 最大；若选中 max_bins 则自动应用 1-SE 回退
        - elbow  : CV-AUC 增量肘部法则
        - one_se : 1-SE 规则
        """
        n_list = [r["n_leaves"] for r in records]

        if self.strategy == "train":
            scores = [r["train_auc"] for r in records]
            return n_list[int(np.argmax(scores))]

        # 以下策略均基于 cv_auc
        scores = [r["cv_auc"] for r in records]
        stds = [r["cv_std"] for r in records]

        if self.strategy == "cv":
            best_idx = int(np.argmax(scores))
            best_n = n_list[best_idx]
            # 当 cv 选中 max_bins（CV-AUC 仍在单调递增）时，
            # 自动应用 1-SE 规则回退到更简洁的分组数
            if best_n == self.max_bins and best_idx > 0:
                best_score = scores[best_idx]
                best_std = stds[best_idx]
                threshold = best_score - best_std
                for i in range(len(n_list)):
                    if scores[i] >= threshold and n_list[i] < best_n:
                        cv_diff = best_score - scores[i]
                        print(f"  ⚠ cv 策略选中最大分组数 {best_n}（CV-AUC 仍在递增），"
                              f"自动应用 1-SE 规则回退到 {n_list[i]} "
                              f"(CV-AUC 差异 {cv_diff:.6f} < std {best_std:.4f})")
                        return n_list[i]
            return best_n

        if self.strategy == "elbow":
            return self._elbow_detect(n_list, scores)

        if self.strategy == "one_se":
            return self._one_se_rule(n_list, scores, stds)

        raise ValueError(f"未知 strategy: {self.strategy}")

    @staticmethod
    def _elbow_detect(n_list: list, scores: list) -> int:
        """肘部法则：增量低于均值一半时停止。"""
        deltas = [scores[i + 1] - scores[i] for i in range(len(scores) - 1)]
        if not deltas:
            return n_list[-1]
        threshold = np.mean(deltas) * 0.5
        for idx, delta in enumerate(deltas):
            if delta < threshold:
                return n_list[idx]
        return n_list[-1]

    @staticmethod
    def _one_se_rule(n_list: list, scores: list, stds: list) -> int:
        """
        1-SE 规则：在 CV-AUC 最优值的一个标准差范围内，选最简单的模型。

        步骤：
        1. 找到 CV-AUC 最大的分组数 best_idx
        2. 计算阈值 = best_score - best_std
        3. 在所有 CV-AUC >= 阈值 的候选中，选分组数最少的
        """
        best_idx = int(np.argmax(scores))
        best_score = scores[best_idx]
        best_std = stds[best_idx]
        threshold = best_score - best_std
        for i in range(len(n_list)):
            if scores[i] >= threshold:
                return n_list[i]
        return n_list[best_idx]

    @staticmethod
    def _build_stats(
        df: pd.DataFrame, col: str, target: str, boundaries: list
    ) -> pd.DataFrame:
        """构建各分箱的统计表。"""
        tmp = df[[col, target]].dropna().copy()
        labels = [f"[{boundaries[i]:.2f}, {boundaries[i+1]:.2f})"
                  for i in range(len(boundaries) - 1)]
        tmp["_bin"] = pd.cut(tmp[col], bins=boundaries, labels=labels, right=False)
        stats = tmp.groupby("_bin", observed=False).agg(
            sample_count=(target, "count"),
            rate=(target, "mean"),
        ).reset_index()
        stats["rate_pct"] = stats["rate"] * 100
        stats.columns = ["分段区间", "样本量", "目标率", "目标率(%)"]
        return stats[["分段区间", "样本量", "目标率(%)"]]


# ============================================================
# 2. Jenks 自然断点分组器（类别变量）
# ============================================================
class JenksBreaksGrouper(BaseBinner):
    """
    基于 Jenks Natural Breaks 的类别变量自然分组。

    原理：先按类别聚合计算目标率，再用 Jenks 算法对率值做一维聚类，
         以 GVF 最大化或肘部法则选择最优分组数。

    分组数选择策略（strategy 参数）
    ------------------------------
    - ``elbow``   : 肘部法则（推荐）。GVF 增量低于均值一半时停止。
                    GVF 随分组数单调递增，elbow 能有效识别"边际收益递减"拐点。
    - ``one_se``  : 1-SE 规则。在 GVF 最优值的一个增量标准差范围内，选最简分组数。
                    类似决策树 one_se，用增量序列的标准差估计"噪声水平"。
    - ``max_gvf`` : GVF 最大值。**注意**：GVF 数学上单调递增，此策略会始终选中
                    max_groups，等同于不做选择。仅在需要"最细粒度分组"时使用。

    Parameters
    ----------
    min_groups : int         最少分组数（默认 3）
    max_groups : int         最多分组数（默认 7）
    min_samples : int        类别最小样本量阈值，低于此值的类别标记为 -1（默认 50）
    min_group_size : int     每组最少类别数，用于肘部法则过滤（默认 5）
    strategy : str           最优分组数选择策略：'elbow'（默认）、'one_se' 或 'max_gvf'
    """

    def __init__(
        self,
        min_groups: int = 3,
        max_groups: int = 7,
        min_samples: int = 50,
        min_group_size: int = 5,
        strategy: str = "elbow",
    ):
        self.min_groups = min_groups
        self.max_groups = max_groups
        self.min_samples = min_samples
        self.min_group_size = min_group_size
        self.strategy = strategy
        self.result_: Optional[BinningResult] = None

    # ---- 核心逻辑 ----
    def fit(self, df: pd.DataFrame, col: str, target: str = "is_suc") -> "JenksBreaksGrouper":
        # 1. 按类别聚合
        agg = df.groupby(col).agg(
            cnt=(target, "count"),
            rate=(target, "mean"),
        ).reset_index()

        # 2. 低频过滤
        valid = agg[agg["cnt"] >= self.min_samples].copy()
        low_freq = agg[agg["cnt"] < self.min_samples].copy()
        print(f"  {col} | 总类别数={len(agg)}, 有效={len(valid)}, "
              f"低频={len(low_freq)}(样本={low_freq['cnt'].sum()})")

        if len(valid) < self.min_groups:
            print(f"  {col} | 有效类别不足，跳过分组")
            self.result_ = BinningResult(
                col_name=col, boundaries=[], stats=pd.DataFrame(),
                label_map={}, meta={"method": "jenks", "skipped": True},
            )
            return self

        # 3. 寻找最优分组
        rates = valid["rate"].values
        best_n, best_breaks, gvf_list, n_list = self._find_optimal_groups(rates)

        # 4. 分配组别
        valid["group"] = self._assign_groups(rates, best_breaks)
        low_freq["group"] = -1

        # 5. 构建映射 & 统计
        label_map = dict(zip(
            pd.concat([valid[col], low_freq[col]]),
            pd.concat([valid["group"], low_freq["group"]]),
        ))
        stats = self._build_stats(valid, col)

        # 6. 保存逐类别统计（供导出用）
        category_stats = valid[[col, "group", "cnt", "rate"]].copy()
        category_stats.columns = ["类别", "组号", "样本量", "目标率"]
        category_stats["目标率(%)"] = category_stats["目标率"] * 100
        category_stats = category_stats.sort_values("组号").reset_index(drop=True)

        self.result_ = BinningResult(
            col_name=col,
            boundaries=best_breaks,
            stats=stats,
            label_map=label_map,
            meta={
                "method": "jenks",
                "strategy": self.strategy,
                "best_n_groups": best_n,
                "best_gvf": gvf_list[n_list.index(best_n)] if best_n in n_list else None,
                "n_valid": len(valid),
                "n_low_freq": len(low_freq),
                "gvf_series": dict(zip(n_list, gvf_list)),
                "category_stats": category_stats,
            },
        )
        return self

    def transform(self, df: pd.DataFrame, col: str) -> pd.Series:
        if self.result_ is None:
            raise RuntimeError("请先调用 fit()")
        return df[col].map(self.result_.label_map).fillna(-2).astype(int)

    # ---- 私有方法 ----
    def _find_optimal_groups(self, rates: np.ndarray) -> tuple:
        """
        遍历分组数，计算 GVF，按策略选择最优。

        Returns: (best_n_groups, best_breaks, gvf_list, n_groups_list)
        """
        gvf_list, breaks_dict, n_list = [], {}, []

        for n_groups in range(self.min_groups, self.max_groups + 1):
            if n_groups >= len(rates):
                continue
            try:
                breaks = jenkspy.jenks_breaks(rates, n_classes=n_groups)
            except Exception as e:
                print(f"  n_groups={n_groups} 计算失败: {e}")
                continue

            group_ids = self._assign_groups(rates, breaks)

            # 肘部法则和 one_se 需要检查每组类别数
            if self.strategy in ("elbow", "one_se"):
                counts = np.bincount(group_ids, minlength=n_groups)
                if np.any(counts < self.min_group_size):
                    continue

            gvf = self._calc_gvf(rates, group_ids, n_groups)
            gvf_list.append(gvf)
            breaks_dict[n_groups] = breaks
            n_list.append(n_groups)
            breaks_disp = [round(float(b), 4) for b in breaks]
            print(f"  n_groups={n_groups} | GVF={gvf:.6f} | breaks={breaks_disp}")

        if not gvf_list:
            print(f"  ⚠ 所有分组数均因 min_group_size={self.min_group_size} 被过滤，"
                  f"返回1组。建议降低 min_group_size 参数（如 min_group_size=2）。")
            return 1, [rates.min(), rates.max()], [], []

        if self.strategy == "elbow":
            best_n = self._elbow_detect(n_list, gvf_list)
        elif self.strategy == "one_se":
            best_n = self._one_se_rule(n_list, gvf_list)
        else:
            best_n = n_list[int(np.argmax(gvf_list))]
            # max_gvf 策略警告：GVF 单调递增，选中 max_groups 是预期行为
            if best_n == n_list[-1] and len(n_list) > 1:
                print(f"  ⚠ max_gvf 策略选中最大分组数 {best_n}："
                      f"GVF 随分组数单调递增，建议改用 'elbow' 或 'one_se' 策略")

        best_breaks = breaks_dict[best_n]
        print(f"  最优 n_groups={best_n} (strategy={self.strategy})")
        return best_n, best_breaks, gvf_list, n_list

    @staticmethod
    def _assign_groups(rates: np.ndarray, breaks: list) -> np.ndarray:
        """根据断点将率值分配到组（从 0 开始）。"""
        n_groups = len(breaks) - 1
        group_ids = np.zeros(len(rates), dtype=int)
        for i, r in enumerate(rates):
            for g in range(n_groups - 1):
                if r <= breaks[g + 1]:
                    group_ids[i] = g
                    break
            else:
                group_ids[i] = n_groups - 1
        return group_ids

    @staticmethod
    def _calc_gvf(rates: np.ndarray, group_ids: np.ndarray, n_groups: int) -> float:
        """计算 GVF (Goodness of Variance Fit)。"""
        sdam = np.sum((rates - rates.mean()) ** 2)
        sdcm = 0.0
        for g in range(n_groups):
            gv = rates[group_ids == g]
            if len(gv) > 0:
                sdcm += np.sum((gv - gv.mean()) ** 2)
        return (sdam - sdcm) / sdam if sdam > 0 else 0.0

    @staticmethod
    def _elbow_detect(n_list: list, gvf_list: list) -> int:
        """肘部法则：寻找 GVF 增量显著下降的点。"""
        deltas = [gvf_list[i + 1] - gvf_list[i] for i in range(len(gvf_list) - 1)]
        if not deltas:
            return n_list[-1]
        threshold = np.mean(deltas) * 0.5
        for idx, delta in enumerate(deltas):
            if delta < threshold:
                return n_list[idx]
        return n_list[-1]

    @staticmethod
    def _one_se_rule(n_list: list, gvf_list: list) -> int:
        """
        1-SE 规则（Jenks 版）：在 GVF 最优值的一个增量标准差范围内，选最简分组数。

        GVF 随分组数单调递增但增量递减。用增量序列的标准差估计"噪声水平"，
        在 max GVF 减去一个标准差的范围内，选最少的分组数。

        步骤：
        1. 找到 GVF 最大的分组数 best_idx
        2. 计算增量序列的标准差 noise = std(deltas)
        3. 阈值 = best_gvf - noise
        4. 在所有 GVF >= 阈值 的候选中，选分组数最少的
        """
        best_idx = int(np.argmax(gvf_list))
        best_gvf = gvf_list[best_idx]

        if len(gvf_list) < 3:
            return n_list[best_idx]

        deltas = [gvf_list[i + 1] - gvf_list[i] for i in range(len(gvf_list) - 1)]
        noise = float(np.std(deltas))
        threshold = best_gvf - noise

        for i in range(len(n_list)):
            if gvf_list[i] >= threshold:
                return n_list[i]
        return n_list[best_idx]

    @staticmethod
    def _build_stats(valid: pd.DataFrame, col: str, max_display: int = 20) -> pd.DataFrame:
        """
        构建各分组的统计表。

        Parameters
        ----------
        valid : pd.DataFrame  有效类别聚合表（含 group 列）
        col : str             类别列名
        max_display : int     每组显示的最大类别数，超出时截断（默认 20）
        """
        summary = valid.groupby("group").agg(
            items_in_group=(col, lambda x: sorted(x.tolist())),
            item_count=(col, "size"),
            sample_count=("cnt", "sum"),
            avg_rate=("rate", "mean"),
            min_rate=("rate", "min"),
            max_rate=("rate", "max"),
        ).reset_index()

        # 截断显示：超过 max_display 个类别时，显示前5 + ... + 后3 + 总数
        def _fmt_items(items, max_disp):
            if len(items) <= max_disp:
                return ",".join(str(v) for v in items)
            head = ", ".join(str(v) for v in items[:5])
            tail = ", ".join(str(v) for v in items[-3:])
            return f"{head}, ..., {tail}  (共 {len(items)} 个)"

        summary["items_str"] = summary["items_in_group"].apply(
            lambda x: _fmt_items(x, max_display)
        )
        summary["rate_range"] = summary.apply(
            lambda r: f"[{r['min_rate']*100:.2f}%, {r['max_rate']*100:.2f}%]", axis=1
        )
        summary["avg_rate_pct"] = summary["avg_rate"] * 100

        result = summary[["group", "items_str", "item_count",
                          "sample_count", "rate_range", "avg_rate_pct"]].copy()
        result.columns = ["组号", "包含类别", "类别数", "样本量", "率值范围", "平均率(%)"]
        return result.sort_values("平均率(%)").reset_index(drop=True)


# ============================================================
# 3. 分箱流水线（多变量批量处理）
# ============================================================
class BinningPipeline:
    """
    批量处理多个变量的分箱流水线。

    用法：
        pipe = BinningPipeline()
        pipe.add("estimate_kilo", DecisionTreeBinner(min_bins=3, max_bins=7, strategy="one_se"))
        pipe.add("dispatch_hour", JenksBreaksGrouper(min_samples=0, strategy="elbow"))
        pipe.run(df, target="is_suc")
        pipe.print_results()
    """

    def __init__(self):
        self._tasks: list[tuple[str, BaseBinner]] = []
        self.results: dict[str, BinningResult] = {}
        self.labeled_df: Optional[pd.DataFrame] = None

    def add(self, col: str, binner: BaseBinner) -> "BinningPipeline":
        """注册一个变量及其对应的分箱器。"""
        self._tasks.append((col, binner))
        return self

    def run(self, df: pd.DataFrame, target: str = "is_suc") -> pd.DataFrame:
        """执行所有分箱任务，返回带标签的 DataFrame。"""
        self.labeled_df = df.copy()
        for col, binner in self._tasks:
            print(f"\n{'='*60}")
            print(f"维度: {col}")
            print(f"{'='*60}")
            labels, result = binner.fit_transform(df, col, target)
            self.labeled_df[f"{col}_group"] = labels
            self.results[col] = result
        return self.labeled_df

    def print_results(self) -> None:
        """打印所有分箱结果的统计表。"""
        for col, result in self.results.items():
            print(f"\n{'='*60}")
            print(f"[{col}] 分箱结果")
            print(f"  方法: {result.meta.get('method', 'N/A')}")

            # 决策树分箱结果
            if result.meta.get("method") == "decision_tree":
                strategy = result.meta.get("strategy", "N/A")
                print(f"  策略: {strategy}")
                if result.meta.get("best_cv_auc") is not None:
                    print(f"  最优 CV-AUC: {result.meta['best_cv_auc']:.6f} "
                          f"± {result.meta.get('best_cv_std', 0):.4f}")
                print(f"  训练集 AUC: {result.meta.get('best_train_auc', 0):.6f}")
                # 打印完整对比表
                all_records = result.meta.get("all_records", [])
                if all_records:
                    print(f"\n  分组数对比:")
                    print(f"  {'n_leaves':>8} | {'train_AUC':>10} | "
                          f"{'cv_AUC':>10} | {'cv_std':>8}")
                    print(f"  {'-'*8}-+-{'-'*10}-+-{'-'*10}-+-{'-'*8}")
                    for r in all_records:
                        cv_str = f"{r['cv_auc']:.6f}" if r['cv_auc'] is not None else "N/A"
                        print(f"  {r['n_leaves']:>8} | {r['train_auc']:>10.6f} | "
                              f"{cv_str:>10} | {r['cv_std']:>8.4f}")
                    best_n = result.meta.get("best_n_leaves")
                    print(f"  >>> 选择 n_leaves={best_n} (strategy={strategy})")

            # Jenks 分组结果
            elif result.meta.get("method") == "jenks":
                if "best_gvf" in result.meta and result.meta["best_gvf"] is not None:
                    print(f"  最优 GVF: {result.meta['best_gvf']:.6f}")
                gvf_series = result.meta.get("gvf_series", {})
                if gvf_series:
                    print(f"\n  分组数对比:")
                    for ng, gv in sorted(gvf_series.items()):
                        marker = " <<<" if ng == result.meta.get("best_n_groups") else ""
                        print(f"  n_groups={ng}: GVF={gv:.6f}{marker}")

            boundaries_disp = [round(b, 4) if isinstance(b, float) else b for b in result.boundaries]
            print(f"  边界/断点: {boundaries_disp}")
            print(f"\n{result.stats.to_string(index=False)}")

    def get_result(self, col: str) -> Optional[BinningResult]:
        """获取某个变量的分箱结果。"""
        return self.results.get(col)

    # ---- 导出方法 ----
    def export_mappings(self, excel_path: Optional[str] = None) -> dict[str, pd.DataFrame]:
        """
        导出所有变量的组号-类别映射表。

        Parameters
        ----------
        excel_path : str, optional
            若提供路径，将所有映射写入多 sheet Excel 文件。

        Returns
        -------
        dict[str, pd.DataFrame]  变量名 -> 映射表
        """
        mappings = {}
        for col, result in self.results.items():
            mappings[col] = result.export_mapping()

        if excel_path:
            try:
                import openpyxl  # noqa: F401
            except ImportError:
                raise ImportError(
                    "导出 Excel 需要 openpyxl，请先安装: pip install openpyxl"
                )
            with pd.ExcelWriter(excel_path, engine="openpyxl") as writer:
                for col, df in mappings.items():
                    # Excel sheet 名最长 31 字符，且不能含特殊字符
                    sheet_name = col[:31].replace("/", "_").replace("\\", "_")
                    df.to_excel(writer, sheet_name=sheet_name, index=False)
            print(f"映射表已导出: {excel_path} ({len(mappings)} 个 sheet)")

        return mappings

    def export_sql_mapping(
        self, col: str, table_name: str = "dim_route_group",
        col_alias: Optional[str] = None, max_sql_rows: int = 500,
    ) -> str:
        """
        生成类别→组号的 SQL 映射表语句。

        适用于类别数极多（如 route）无法用 CASE WHEN 表达的场景，
        生成 CREATE TABLE + INSERT 语句，在数仓中 JOIN 使用。

        - 类别数 ≤ max_sql_rows：生成完整 INSERT ... VALUES 语句
        - 类别数 > max_sql_rows：生成建表语句 + 提示从 CSV 导入

        Parameters
        ----------
        col : str            变量名（须已 run 过）
        table_name : str     映射表名（默认 'dim_route_group'）
        col_alias : str      映射表中类别列的列名，默认用原变量名
        max_sql_rows : int   直接生成 SQL 的最大行数（默认 500），超出则建议 CSV 导入

        Returns
        -------
        str  SQL 语句
        """
        result = self.results.get(col)
        if result is None:
            raise KeyError(f"未找到变量 '{col}' 的分箱结果，请先 run()")

        col_name = col_alias or col
        mapping = result.export_mapping()
        method = result.meta.get("method", "")
        total = len(mapping)

        lines = []

        if method == "jenks":
            lines.append(f"-- {col} 分组映射表 ({total} 个类别, {result.meta.get('best_n_groups', '?')} 个组)")
            lines.append(f"-- 策略: {result.meta.get('strategy', 'N/A')}, "
                         f"GVF: {result.meta.get('best_gvf', 0):.6f}")
            lines.append(f"CREATE TABLE IF NOT EXISTS {table_name} (")
            lines.append(f"    {col_name}  STRING COMMENT '原始类别',")
            lines.append(f"    group_id    INT    COMMENT '组号 (-1=低频过滤)'")
            lines.append(f") COMMENT '{col} 自然分组映射 (Jenks)';")
            lines.append("")

            if total <= max_sql_rows:
                # 直接生成 INSERT VALUES
                lines.append(f"INSERT INTO TABLE {table_name}")
                lines.append(f"VALUES")
                for i, (_, row) in enumerate(mapping.iterrows()):
                    cat = str(row["类别"]).replace("'", "\\'")
                    gid = int(row["组号"])
                    sep = "," if i < total - 1 else ";"
                    lines.append(f"    ('{cat}', {gid}){sep}")
            else:
                # 类别过多，建议 CSV 导入
                lines.append(f"-- ⚠ 类别数 {total} 超过 {max_sql_rows}，建议从 CSV 导入：")
                lines.append(f"-- 1. 导出 CSV:  pipe.export_mappings('{col}_mapping.xlsx')")
                lines.append(f"-- 2. 上传至数仓后 LOAD DATA 或 Spark 读取")
                lines.append(f"-- 3. 或使用以下 INSERT 覆盖前 {max_sql_rows} 行（示例）:")
                lines.append(f"INSERT INTO TABLE {table_name}")
                lines.append(f"VALUES")
                for i in range(min(max_sql_rows, total)):
                    row = mapping.iloc[i]
                    cat = str(row["类别"]).replace("'", "\\'")
                    gid = int(row["组号"])
                    sep = "," if i < min(max_sql_rows, total) - 1 else ";"
                    lines.append(f"    ('{cat}', {gid}){sep}")
                lines.append(f"-- ... 剩余 {total - max_sql_rows} 行请从 CSV 导入")

        elif method == "decision_tree":
            boundaries = result.boundaries
            n_bins = len(boundaries) - 1
            lines.append(f"-- {col} 分箱边界 ({n_bins} 个分箱)")
            lines.append(f"CREATE TABLE IF NOT EXISTS {table_name} (")
            lines.append(f"    group_id    INT    COMMENT '组号',")
            lines.append(f"    lower_bound DOUBLE COMMENT '下界',")
            lines.append(f"    upper_bound DOUBLE COMMENT '上界'")
            lines.append(f") COMMENT '{col} 决策树分箱边界';")
            lines.append("")
            lines.append(f"INSERT INTO TABLE {table_name}")
            lines.append(f"VALUES")
            for i in range(n_bins):
                lo = boundaries[i]
                hi = boundaries[i + 1]
                lo_str = "-99999999.0" if lo == -np.inf else str(lo)
                hi_str = "99999999.0" if hi == np.inf else str(hi)
                sep = "," if i < n_bins - 1 else ";"
                lines.append(f"    ({i}, {lo_str}, {hi_str}){sep}")

        return "\n".join(lines)

    def export_sql_case_when(self, col: str, indent: str = "    ") -> str:
        """
        生成决策树分箱的 SQL CASE WHEN 语句。

        仅适用于决策树分箱（连续变量），Jenks 分组请用 export_sql_mapping。

        Parameters
        ----------
        col : str     变量名
        indent : str  缩进（默认 4 空格）

        Returns
        -------
        str  SQL CASE WHEN 语句
        """
        result = self.results.get(col)
        if result is None:
            raise KeyError(f"未找到变量 '{col}' 的分箱结果")
        if result.meta.get("method") != "decision_tree":
            raise ValueError(f"'{col}' 不是决策树分箱，无法生成 CASE WHEN")

        boundaries = result.boundaries
        lines = [f"CASE"]
        for i in range(len(boundaries) - 1):
            lo = boundaries[i]
            hi = boundaries[i + 1]
            if i == 0:
                lines.append(f"{indent}WHEN {col} < {hi} THEN {i}")
            elif i == len(boundaries) - 2:
                lines.append(f"{indent}WHEN {col} >= {lo} THEN {i}")
            else:
                lines.append(f"{indent}WHEN {col} >= {lo} AND {col} < {hi} THEN {i}")
        lines.append(f"END AS {col}_group")
        return "\n".join(lines)

使用示例

from binning import BinningPipeline
 
print("=" * 60)
print("批量分箱流水线")
print("=" * 60)
 
pipe = BinningPipeline()
 
# 注册连续变量 → 决策树分箱
pipe.add("distance", DecisionTreeBinner(
    min_bins=3, max_bins=7, strategy="one_se", cv_folds=5
))
 
# 注册类别变量 → Jenks 自然断点
pipe.add("hour", JenksBreaksGrouper(
    min_groups=3, max_groups=7, min_samples=0, strategy="elbow"
))
 
pipe.add("region", JenksBreaksGrouper(
    min_groups=3, max_groups=7, min_samples=50, strategy="elbow"
))
 
# 一键执行
labeled_df = pipe.run(df, target="conversion")
 
# 打印所有结果
pipe.print_results()
 
# labeled_df 新增了 *_group 列
print(f"\n带标签的 DataFrame 列: {list(labeled_df.columns)}")
print(f"\ndistance_group 分布:")
print(labeled_df["distance_group"].value_counts().sort_index())

# ---- 1. 导出 Excel 映射表 ----
pipe.export_mappings("binning_mappings.xlsx")
# 每个变量一个 sheet，包含组号、区间/类别、样本量、目标率
 
# ---- 2. 导出决策树分箱的 SQL CASE WHEN ----
sql_case = pipe.export_sql_case_when("distance")
print("SQL CASE WHEN (distance):")
print(sql_case)