Modeling_Tool.Sample

样本管理层 —— 切分、分层、均衡、拒绝推断、分布适配。

样本切分与采样 — Sample_Split

Sample_Split

Sampling classes for credit modeling.

This module provides classes for splitting samples, stratified sampling, and sample balancing for credit model development.

类：

名称	描述
SampleSplitter : Split data into train/test samples.
StratifiedSampler : Stratified sampling with target balance control.
SampleBalancer : Balance samples using various techniques.

示例：

>>> from Modeling_Tool_refactored.sample import SampleSplitter
>>> splitter = SampleSplitter()
>>> train, test = splitter.split(df, 'target', test_size=0.3)

SampleSplitter

Split data into training and testing samples.

This class provides flexible sample splitting with support for stratification, random sampling, and custom split ratios.

参数：

名称	类型	描述	默认
test_size	float	Proportion of data for testing (0 to 1).	0.3
random_state	int	Random seed for reproducibility.	None
stratify	bool	Whether to stratify by target variable.	True

属性：

名称	类型	描述
train_index_	ndarray	Indices for training data.
test_index_	ndarray	Indices for testing data.

方法：

名称	描述
split	Split data into train and test sets.
split_df	Split DataFrame while excluding certain columns.

示例：

>>> splitter = SampleSplitter(test_size=0.2, random_state=42)
>>> train, test = splitter.split(X, y)
>>> print(f"Train size: {len(train)}, Test size: {len(test)}")

源代码位于： Modeling_Tool/Sample/Sample_Split.py

class SampleSplitter:
    """
    Split data into training and testing samples.

    This class provides flexible sample splitting with support for
    stratification, random sampling, and custom split ratios.

    Parameters
    ----------
    test_size : float, default 0.3
        Proportion of data for testing (0 to 1).
    random_state : int, optional
        Random seed for reproducibility.
    stratify : bool, default True
        Whether to stratify by target variable.

    Attributes
    ----------
    train_index_ : numpy.ndarray
        Indices for training data.
    test_index_ : numpy.ndarray
        Indices for testing data.

    Methods
    -------
    split(X, y, test_size=None, stratify=None)
        Split data into train and test sets.
    split_df(df, target, exclude_cols=None)
        Split DataFrame while excluding certain columns.

    Examples
    --------
    >>> splitter = SampleSplitter(test_size=0.2, random_state=42)
    >>> train, test = splitter.split(X, y)
    >>> print(f"Train size: {len(train)}, Test size: {len(test)}")
    """

    def __init__(self, test_size: float = 0.3, 
                 random_state: Optional[int] = None,
                 stratify: bool = True):
        """
        Initialize SampleSplitter.

        Parameters
        ----------
        test_size : float, default 0.3
            Proportion for testing.
        random_state : int, optional
            Random seed.
        stratify : bool, default True
            Whether to stratify.
        """
        self.test_size = test_size
        self.random_state = random_state
        self.stratify = stratify
        self.train_index_ = None
        self.test_index_ = None

    def split(self, X: Union[pd.DataFrame, np.ndarray],
              y: Union[pd.Series, np.ndarray],
              test_size: Optional[float] = None,
              stratify: Optional[bool] = None) -> Tuple:
        """
        Split data into train and test sets.

        Parameters
        ----------
        X : pandas.DataFrame or numpy.ndarray
            Features.
        y : pandas.Series or numpy.ndarray
            Target variable.
        test_size : float, optional
            Override default test size.
        stratify : bool, optional
            Override default stratify setting.

        Returns
        -------
        tuple
            (X_train, X_test, y_train, y_test)

        Examples
        --------
        >>> X_train, X_test, y_train, y_test = splitter.split(X, y)
        """
        test_size = test_size if test_size is not None else self.test_size
        stratify = stratify if stratify is not None else self.stratify

        y_arr = np.array(y)
        stratify_param = y_arr if stratify and len(np.unique(y_arr)) > 1 else None

        return train_test_split(
            X, y,
            test_size=test_size,
            random_state=self.random_state,
            stratify=stratify_param
        )

    def split_indices(
        self,
        index: Union[pd.Index, np.ndarray, List[Any]],
        y: Union[pd.Series, np.ndarray],
        test_size: Optional[float] = None,
        stratify: Optional[bool] = None,
    ) -> Tuple[np.ndarray, np.ndarray]:
        """Split index labels without copying the underlying feature frame."""
        resolved_test_size = test_size if test_size is not None else self.test_size
        resolved_stratify = self.stratify if stratify is None else bool(stratify)
        index_values = np.asarray(index)
        y_values = np.asarray(y)
        if len(index_values) != len(y_values):
            raise ValueError("index and y must have the same length")
        stratify_values = (
            y_values
            if resolved_stratify and len(pd.unique(y_values)) > 1
            else None
        )
        train_idx, test_idx = train_test_split(
            index_values,
            test_size=resolved_test_size,
            random_state=self.random_state,
            stratify=stratify_values,
        )
        self.train_index_ = np.asarray(train_idx)
        self.test_index_ = np.asarray(test_idx)
        return self.train_index_, self.test_index_

    def split_df(self, df: pd.DataFrame, target: str,
                exclude_cols: Optional[List[str]] = None,
                test_size: Optional[float] = None) -> Tuple[pd.DataFrame, pd.DataFrame]:
        """
        Split DataFrame into train/test partitions while keeping every column.

        ``exclude_cols`` means "these columns should not participate in the
        split strategy" (i.e. they must not be used as stratify keys or as
        features driving the split), not "these columns should be dropped
        from the output frames." Before 0.4.2 the columns listed here were
        silently dropped from ``train_df`` / ``test_df``, which broke every
        downstream step that relied on carry-through keys such as ``user_id``
        or ``apply_date``. Fixed in 0.4.2: both output frames now contain the
        full column set of the input frame.

        Parameters
        ----------
        df : pandas.DataFrame
            Input DataFrame.
        target : str
            Target column name.
        exclude_cols : list of str, optional
            Columns to carry through the split without letting them influence
            the split strategy. Their values are preserved verbatim on both
            output frames.
        test_size : float, optional
            Override default test size.

        Returns
        -------
        tuple
            (train_df, test_df) — both include every column of ``df`` (target
            column, features, and any ``exclude_cols``).

        Examples
        --------
        >>> train_df, test_df = splitter.split_df(df, 'target', exclude_cols=['id', 'date'])
        >>> assert 'id' in train_df.columns and 'date' in test_df.columns
        """
        test_size = test_size if test_size is not None else self.test_size

        exclude_cols = list(exclude_cols or [])

        y = df[target]

        if self.stratify:
            strat = y
        else:
            strat = None

        # Split on the index so that we can reindex the full frame afterwards.
        # This preserves every column (features, target, and exclude_cols)
        # on both output frames.
        train_idx, test_idx = self.split_indices(
            df.index,
            strat if strat is not None else y,
            test_size=test_size,
            stratify=self.stratify,
        )

        train_df = df.loc[train_idx].copy()
        test_df = df.loc[test_idx].copy()

        return train_df, test_df

split

split(X: Union[DataFrame, ndarray], y: Union[Series, ndarray], test_size: Optional[float] = None, stratify: Optional[bool] = None) -> Tuple

Split data into train and test sets.

参数：

名称	类型	描述	默认
X	DataFrame or ndarray	Features.	必需
y	Series or ndarray	Target variable.	必需
test_size	float	Override default test size.	None
stratify	bool	Override default stratify setting.	None

返回：

类型	描述
tuple	(X_train, X_test, y_train, y_test)

示例：

>>> X_train, X_test, y_train, y_test = splitter.split(X, y)

源代码位于： Modeling_Tool/Sample/Sample_Split.py

def split(self, X: Union[pd.DataFrame, np.ndarray],
          y: Union[pd.Series, np.ndarray],
          test_size: Optional[float] = None,
          stratify: Optional[bool] = None) -> Tuple:
    """
    Split data into train and test sets.

    Parameters
    ----------
    X : pandas.DataFrame or numpy.ndarray
        Features.
    y : pandas.Series or numpy.ndarray
        Target variable.
    test_size : float, optional
        Override default test size.
    stratify : bool, optional
        Override default stratify setting.

    Returns
    -------
    tuple
        (X_train, X_test, y_train, y_test)

    Examples
    --------
    >>> X_train, X_test, y_train, y_test = splitter.split(X, y)
    """
    test_size = test_size if test_size is not None else self.test_size
    stratify = stratify if stratify is not None else self.stratify

    y_arr = np.array(y)
    stratify_param = y_arr if stratify and len(np.unique(y_arr)) > 1 else None

    return train_test_split(
        X, y,
        test_size=test_size,
        random_state=self.random_state,
        stratify=stratify_param
    )

split_indices

split_indices(index: Union[Index, ndarray, List[Any]], y: Union[Series, ndarray], test_size: Optional[float] = None, stratify: Optional[bool] = None) -> Tuple[ndarray, ndarray]

Split index labels without copying the underlying feature frame.

源代码位于： Modeling_Tool/Sample/Sample_Split.py

def split_indices(
    self,
    index: Union[pd.Index, np.ndarray, List[Any]],
    y: Union[pd.Series, np.ndarray],
    test_size: Optional[float] = None,
    stratify: Optional[bool] = None,
) -> Tuple[np.ndarray, np.ndarray]:
    """Split index labels without copying the underlying feature frame."""
    resolved_test_size = test_size if test_size is not None else self.test_size
    resolved_stratify = self.stratify if stratify is None else bool(stratify)
    index_values = np.asarray(index)
    y_values = np.asarray(y)
    if len(index_values) != len(y_values):
        raise ValueError("index and y must have the same length")
    stratify_values = (
        y_values
        if resolved_stratify and len(pd.unique(y_values)) > 1
        else None
    )
    train_idx, test_idx = train_test_split(
        index_values,
        test_size=resolved_test_size,
        random_state=self.random_state,
        stratify=stratify_values,
    )
    self.train_index_ = np.asarray(train_idx)
    self.test_index_ = np.asarray(test_idx)
    return self.train_index_, self.test_index_

split_df

split_df(df: DataFrame, target: str, exclude_cols: Optional[List[str]] = None, test_size: Optional[float] = None) -> Tuple[DataFrame, DataFrame]

Split DataFrame into train/test partitions while keeping every column.

exclude_cols means "these columns should not participate in the split strategy" (i.e. they must not be used as stratify keys or as features driving the split), not "these columns should be dropped from the output frames." Before 0.4.2 the columns listed here were silently dropped from train_df / test_df, which broke every downstream step that relied on carry-through keys such as user_id or apply_date. Fixed in 0.4.2: both output frames now contain the full column set of the input frame.

参数：

名称	类型	描述	默认
df	DataFrame	Input DataFrame.	必需
target	str	Target column name.	必需
exclude_cols	list of str	Columns to carry through the split without letting them influence the split strategy. Their values are preserved verbatim on both output frames.	None
test_size	float	Override default test size.	None

返回：

类型	描述
tuple	(train_df, test_df) — both include every column of df (target column, features, and any exclude_cols).

示例：

>>> train_df, test_df = splitter.split_df(df, 'target', exclude_cols=['id', 'date'])
>>> assert 'id' in train_df.columns and 'date' in test_df.columns

源代码位于： Modeling_Tool/Sample/Sample_Split.py

def split_df(self, df: pd.DataFrame, target: str,
            exclude_cols: Optional[List[str]] = None,
            test_size: Optional[float] = None) -> Tuple[pd.DataFrame, pd.DataFrame]:
    """
    Split DataFrame into train/test partitions while keeping every column.

    ``exclude_cols`` means "these columns should not participate in the
    split strategy" (i.e. they must not be used as stratify keys or as
    features driving the split), not "these columns should be dropped
    from the output frames." Before 0.4.2 the columns listed here were
    silently dropped from ``train_df`` / ``test_df``, which broke every
    downstream step that relied on carry-through keys such as ``user_id``
    or ``apply_date``. Fixed in 0.4.2: both output frames now contain the
    full column set of the input frame.

    Parameters
    ----------
    df : pandas.DataFrame
        Input DataFrame.
    target : str
        Target column name.
    exclude_cols : list of str, optional
        Columns to carry through the split without letting them influence
        the split strategy. Their values are preserved verbatim on both
        output frames.
    test_size : float, optional
        Override default test size.

    Returns
    -------
    tuple
        (train_df, test_df) — both include every column of ``df`` (target
        column, features, and any ``exclude_cols``).

    Examples
    --------
    >>> train_df, test_df = splitter.split_df(df, 'target', exclude_cols=['id', 'date'])
    >>> assert 'id' in train_df.columns and 'date' in test_df.columns
    """
    test_size = test_size if test_size is not None else self.test_size

    exclude_cols = list(exclude_cols or [])

    y = df[target]

    if self.stratify:
        strat = y
    else:
        strat = None

    # Split on the index so that we can reindex the full frame afterwards.
    # This preserves every column (features, target, and exclude_cols)
    # on both output frames.
    train_idx, test_idx = self.split_indices(
        df.index,
        strat if strat is not None else y,
        test_size=test_size,
        stratify=self.stratify,
    )

    train_df = df.loc[train_idx].copy()
    test_df = df.loc[test_idx].copy()

    return train_df, test_df

StratifiedSampler

Stratified sampling with target balance control.

This class provides stratified sampling that maintains the target distribution while allowing controlled sampling.

参数：

名称	类型	描述	默认
target_rate	float	Target bad rate in the sample.	None
random_state	int	Random seed.	None

方法：

名称	描述
sample	Perform stratified sampling.
balance	Balance sample by adjusting target distribution.

示例：

>>> sampler = StratifiedSampler(target_rate=0.15)
>>> balanced = sampler.balance(df, 'target', method='undersample')

源代码位于： Modeling_Tool/Sample/Sample_Split.py

class StratifiedSampler:
    """
    Stratified sampling with target balance control.

    This class provides stratified sampling that maintains
    the target distribution while allowing controlled sampling.

    Parameters
    ----------
    target_rate : float, optional
        Target bad rate in the sample.
    random_state : int, optional
        Random seed.

    Methods
    -------
    sample(df, target, n_samples=None, sample_frac=None)
        Perform stratified sampling.
    balance(df, target, method='undersample')
        Balance sample by adjusting target distribution.

    Examples
    --------
    >>> sampler = StratifiedSampler(target_rate=0.15)
    >>> balanced = sampler.balance(df, 'target', method='undersample')
    """

    def __init__(self, target_rate: Optional[float] = None,
                 random_state: Optional[int] = None):
        """
        Initialize StratifiedSampler.
        """
        self.target_rate = target_rate
        self.random_state = random_state
        self.original_rate_ = None

    def sample(self, df: pd.DataFrame, target: str,
              n_samples: Optional[int] = None,
              sample_frac: Optional[float] = None) -> pd.DataFrame:
        """
        Perform stratified sampling.

        Parameters
        ----------
        df : pandas.DataFrame
            Input data.
        target : str
            Target column name.
        n_samples : int, optional
            Number of samples to draw.
        sample_frac : float, optional
            Fraction of data to sample.

        Returns
        -------
        pandas.DataFrame
            Sampled DataFrame.

        Examples
        --------
        >>> sampled = sampler.sample(df, 'target', sample_frac=0.5)
        """
        if sample_frac is not None:
            return df.sample(frac=sample_frac, random_state=self.random_state)

        if n_samples is not None:
            return df.sample(n=n_samples, random_state=self.random_state)

        return df.copy()

    def balance(self, df: pd.DataFrame, target: str,
               method: str = 'undersample') -> pd.DataFrame:
        """
        Balance sample by adjusting target distribution.

        Parameters
        ----------
        df : pandas.DataFrame
            Input data.
        target : str
            Target column name.
        method : str, default 'undersample'
            Balancing method: 'undersample', 'oversample', 'smote'.

        Returns
        -------
        pandas.DataFrame
            Balanced DataFrame.

        Examples
        --------
        >>> balanced = sampler.balance(df, 'target', method='undersample')
        """
        self.original_rate_ = df[target].mean()

        if method == 'undersample':
            return self._undersample(df, target)
        elif method == 'oversample':
            return self._oversample(df, target)
        elif method == 'smote':
            return self._smote(df, target)
        else:
            raise ValueError(f"Unknown method: {method}")

    def _undersample(self, df: pd.DataFrame, target: str) -> pd.DataFrame:
        """
        Undersample majority class.
        """
        goods = df[df[target] == 0]
        bads = df[df[target] == 1]

        if self.target_rate is not None:
            n_bads = len(bads)
            n_goods = int(n_bads * (1 - self.target_rate) / self.target_rate)
            goods = goods.sample(n=min(n_goods, len(goods)), random_state=self.random_state)
        else:
            goods = goods.sample(n=len(bads), random_state=self.random_state)

        return pd.concat([goods, bads]).sample(frac=1, random_state=self.random_state)

    def _oversample(self, df: pd.DataFrame, target: str) -> pd.DataFrame:
        """
        Oversample minority class.
        """
        goods = df[df[target] == 0]
        bads = df[df[target] == 1]

        if len(bads) < len(goods):
            if self.target_rate is not None:
                n_goods = len(goods)
                n_bads = int(n_goods * self.target_rate / (1 - self.target_rate))
                bads = bads.sample(n=n_bads, replace=True, random_state=self.random_state)
            else:
                bads = bads.sample(n=len(goods), replace=True, random_state=self.random_state)

        return pd.concat([goods, bads]).sample(frac=1, random_state=self.random_state)

    def _smote(self, df: pd.DataFrame, target: str) -> pd.DataFrame:
        """
        SMOTE oversampling (requires imbalanced-learn).
        """
        try:
            from imblearn.over_sampling import SMOTE
        except ImportError:
            raise ImportError("imbalanced-learn required for SMOTE. Install with: pip install imbalanced-learn")

        goods = df[df[target] == 0]
        bads = df[df[target] == 1]

        if len(bads) < 6:
            return self._oversample(df, target)

        X_cols = [c for c in df.columns if c != target]
        X = df[X_cols].values
        y = df[target].values

        smote = SMOTE(random_state=self.random_state)
        X_resampled, y_resampled = smote.fit_resample(X, y)

        return pd.DataFrame(X_resampled, columns=X_cols).assign(**{target: y_resampled})

sample

sample(df: DataFrame, target: str, n_samples: Optional[int] = None, sample_frac: Optional[float] = None) -> DataFrame

Perform stratified sampling.

参数：

名称	类型	描述	默认
df	DataFrame	Input data.	必需
target	str	Target column name.	必需
n_samples	int	Number of samples to draw.	None
sample_frac	float	Fraction of data to sample.	None

返回：

类型	描述
DataFrame	Sampled DataFrame.

示例：

>>> sampled = sampler.sample(df, 'target', sample_frac=0.5)

源代码位于： Modeling_Tool/Sample/Sample_Split.py

def sample(self, df: pd.DataFrame, target: str,
          n_samples: Optional[int] = None,
          sample_frac: Optional[float] = None) -> pd.DataFrame:
    """
    Perform stratified sampling.

    Parameters
    ----------
    df : pandas.DataFrame
        Input data.
    target : str
        Target column name.
    n_samples : int, optional
        Number of samples to draw.
    sample_frac : float, optional
        Fraction of data to sample.

    Returns
    -------
    pandas.DataFrame
        Sampled DataFrame.

    Examples
    --------
    >>> sampled = sampler.sample(df, 'target', sample_frac=0.5)
    """
    if sample_frac is not None:
        return df.sample(frac=sample_frac, random_state=self.random_state)

    if n_samples is not None:
        return df.sample(n=n_samples, random_state=self.random_state)

    return df.copy()

balance

balance(df: DataFrame, target: str, method: str = 'undersample') -> DataFrame

Balance sample by adjusting target distribution.

参数：

名称	类型	描述	默认
df	DataFrame	Input data.	必需
target	str	Target column name.	必需
method	str	Balancing method: 'undersample', 'oversample', 'smote'.	'undersample'

返回：

类型	描述
DataFrame	Balanced DataFrame.

示例：

>>> balanced = sampler.balance(df, 'target', method='undersample')

源代码位于： Modeling_Tool/Sample/Sample_Split.py

def balance(self, df: pd.DataFrame, target: str,
           method: str = 'undersample') -> pd.DataFrame:
    """
    Balance sample by adjusting target distribution.

    Parameters
    ----------
    df : pandas.DataFrame
        Input data.
    target : str
        Target column name.
    method : str, default 'undersample'
        Balancing method: 'undersample', 'oversample', 'smote'.

    Returns
    -------
    pandas.DataFrame
        Balanced DataFrame.

    Examples
    --------
    >>> balanced = sampler.balance(df, 'target', method='undersample')
    """
    self.original_rate_ = df[target].mean()

    if method == 'undersample':
        return self._undersample(df, target)
    elif method == 'oversample':
        return self._oversample(df, target)
    elif method == 'smote':
        return self._smote(df, target)
    else:
        raise ValueError(f"Unknown method: {method}")

SampleBalancer

Advanced sample balancing with multiple methods.

This class provides various sampling techniques to handle class imbalance in credit modeling.

参数：

名称	类型	描述	默认
method	str	Balancing method.	'random'
target_ratio	float	Desired minority/majority ratio.	None
random_state	int	Random seed.	None

方法：

名称	描述
fit_resample	Resample features and target.
get_balanced_indices	Get indices for balanced sampling.

示例：

>>> balancer = SampleBalancer(method='nearmiss')
>>> X_bal, y_bal = balancer.fit_resample(X, y)

源代码位于： Modeling_Tool/Sample/Sample_Split.py

class SampleBalancer:
    """
    Advanced sample balancing with multiple methods.

    This class provides various sampling techniques to handle
    class imbalance in credit modeling.

    Parameters
    ----------
    method : str, default 'random'
        Balancing method.
    target_ratio : float, optional
        Desired minority/majority ratio.
    random_state : int, optional
        Random seed.

    Methods
    -------
    fit_resample(X, y)
        Resample features and target.
    get_balanced_indices(y)
        Get indices for balanced sampling.

    Examples
    --------
    >>> balancer = SampleBalancer(method='nearmiss')
    >>> X_bal, y_bal = balancer.fit_resample(X, y)
    """

    def __init__(self, method: str = 'random',
                 target_ratio: Optional[float] = None,
                 random_state: Optional[int] = None):
        """
        Initialize SampleBalancer.
        """
        self.method = method
        self.target_ratio = target_ratio
        self.random_state = random_state

    def fit_resample(self, X: Union[pd.DataFrame, np.ndarray],
                    y: Union[pd.Series, np.ndarray]) -> Tuple:
        """
        Resample data to balance classes.

        Parameters
        ----------
        X : pandas.DataFrame or numpy.ndarray
            Features.
        y : pandas.Series or numpy.ndarray
            Target.

        Returns
        -------
        tuple
            (X_resampled, y_resampled)

        Examples
        --------
        >>> X_bal, y_bal = balancer.fit_resample(X, y)
        """
        if self.method == 'random':
            return self._random_undersample(X, y)
        elif self.method == 'nearmiss':
            return self._nearmiss(X, y)
        elif self.method == 'tomek':
            return self._tomek_links(X, y)
        elif self.method == 'enn':
            return self._edited_nn(X, y)
        else:
            raise ValueError(f"Unknown method: {self.method}")

    def _random_undersample(self, X, y):
        """
        Random undersampling.
        """
        goods = y == 0
        bads = y == 1

        n_bads = goods.sum() if self.target_ratio else bads.sum()

        if self.target_ratio:
            n_goods = int(n_bads * (1 - self.target_ratio) / self.target_ratio)
        else:
            n_goods = n_bads

        goods_indices = np.where(goods)[0]
        bads_indices = np.where(bads)[0]

        np.random.seed(self.random_state)
        goods_sample = np.random.choice(goods_indices, size=n_goods, replace=False)

        selected = np.concatenate([goods_sample, bads_indices])
        np.random.shuffle(selected)

        if isinstance(X, pd.DataFrame):
            return X.iloc[selected].copy(), y.iloc[selected].copy()
        return X[selected], y[selected]

    def _nearmiss(self, X, y):
        """
        NearMiss undersampling.
        """
        try:
            from imblearn.under_sampling import NearMiss
        except ImportError:
            raise ImportError("imbalanced-learn required. Install with: pip install imbalanced-learn")

        nm = NearMiss(version=1, random_state=self.random_state)
        return nm.fit_resample(X, y)

    def _tomek_links(self, X, y):
        """
        Tomek links cleaning.
        """
        try:
            from imblearn.under_sampling import TomekLinks
        except ImportError:
            raise ImportError("imbalanced-learn required. Install with: pip install imbalanced-learn")

        tl = TomekLinks(random_state=self.random_state)
        return tl.fit_resample(X, y)

    def _edited_nn(self, X, y):
        """
        Edited Nearest Neighbors cleaning.
        """
        try:
            from imblearn.under_sampling import EditedNearestNeighbours
        except ImportError:
            raise ImportError("imbalanced-learn required. Install with: pip install imbalanced-learn")

        enn = EditedNearestNeighbours(random_state=self.random_state)
        return enn.fit_resample(X, y)

fit_resample

fit_resample(X: Union[DataFrame, ndarray], y: Union[Series, ndarray]) -> Tuple

Resample data to balance classes.

参数：

名称	类型	描述	默认
X	DataFrame or ndarray	Features.	必需
y	Series or ndarray	Target.	必需

返回：

类型	描述
tuple	(X_resampled, y_resampled)

示例：

>>> X_bal, y_bal = balancer.fit_resample(X, y)

源代码位于： Modeling_Tool/Sample/Sample_Split.py

def fit_resample(self, X: Union[pd.DataFrame, np.ndarray],
                y: Union[pd.Series, np.ndarray]) -> Tuple:
    """
    Resample data to balance classes.

    Parameters
    ----------
    X : pandas.DataFrame or numpy.ndarray
        Features.
    y : pandas.Series or numpy.ndarray
        Target.

    Returns
    -------
    tuple
        (X_resampled, y_resampled)

    Examples
    --------
    >>> X_bal, y_bal = balancer.fit_resample(X, y)
    """
    if self.method == 'random':
        return self._random_undersample(X, y)
    elif self.method == 'nearmiss':
        return self._nearmiss(X, y)
    elif self.method == 'tomek':
        return self._tomek_links(X, y)
    elif self.method == 'enn':
        return self._edited_nn(X, y)
    else:
        raise ValueError(f"Unknown method: {self.method}")

select_sample_seed

select_sample_seed(master_df, oot_split_col, model, tgt_name, seed_range=(3000, 3050), ins_prop=0.7)

Select Best Seed for Sample Splitting.

源代码位于： Modeling_Tool/Sample/Sample_Split.py

def select_sample_seed(master_df, oot_split_col, model, tgt_name, seed_range = (3000, 3050), ins_prop = 0.7):
    """ Select Best Seed for Sample Splitting. """

    from tqdm import tqdm

    if isinstance(model, str):
        score = model

    perf_res = pd.DataFrame()
    for seed in tqdm(range(seed_range[0], seed_range[1])):

        train_df = master_df.loc[master_df[oot_split_col].isin([1])]
        oot_df = master_df.loc[master_df[oot_split_col].isin([2])]

        sampler = SampleSplitter(test_size = (1 - ins_prop), random_state=seed, stratify=True)
        mdl_df, val_df = sampler.split_df(train_df, tgt_name)

        mdl_df['sample_ind_fnl'] = "ins"
        val_df['sample_ind_fnl'] = "oos"
        oot_df['sample_ind_fnl'] = "oot"

        drv_w_sample_ind = pd.concat([mdl_df, val_df, oot_df])

        ins_df = drv_w_sample_ind.loc[drv_w_sample_ind['sample_ind_fnl'].isin(['ins'])]
        oos_df = drv_w_sample_ind.loc[drv_w_sample_ind['sample_ind_fnl'].isin(['oos'])]
        oot_df = drv_w_sample_ind.loc[drv_w_sample_ind['sample_ind_fnl'].isin(['oot'])]

        evaluator = PerformanceEvaluator(tgt_name=tgt_name, model=model, feature_cols=get_feature_names(model)) if not isinstance(model, str) else PerformanceEvaluator(tgt_name=tgt_name, scr_name = score)

        if ins_df.shape[0] > 0:
            evaluator = evaluator.add_dataset('train', ins_df)

        if oos_df.shape[0] > 0:
            evaluator = evaluator.add_dataset('validation', oos_df)

        if oot_df.shape[0] > 0:
            evaluator = evaluator.add_dataset('oot', oot_df)

        result = evaluator.evaluate(to_show=False, display = False)
        result['seed'] = seed

        perf_res = pd.concat([perf_res, result], axis = 0)

    return perf_res

拒绝推断 — Reject_Infer

Reject_Infer

Reject inference classes for credit modeling.

This module provides classes for applying reject inference techniques to handle the selection bias in credit modeling when using approved loan data only.

类：

名称	描述
RejectInferrer : Base class for reject inference.
RejectInferenceFactory : Factory for creating reject inference methods.
ParcelingInferrer : Parceling method for reject inference.
FuzzyAugmentInferrer : Fuzzy augmentation method.
HardCutoffInferrer : Hard cutoff method.
SimpleAugmentInferrer : Simple augmentation method.

示例：

>>> from Modeling_Tool_refactored.sample import RejectInferrer
>>> inferrer = RejectInferenceFactory.create('parceling')
>>> df_inferred = inferrer.infer(df_approved, df_rejected, 'score')

RejectInferrer

Bases: ABC

Abstract base class for reject inference methods.

Reject inference is used to address selection bias when building credit models on approved loans only.

参数：

名称	类型	描述	默认
target_col	str	Name of the target column.	'target'
score_col	str	Name of the score/probability column.	'score'

方法：

名称	描述
infer	Apply reject inference.

源代码位于： Modeling_Tool/Sample/Reject_Infer.py

class RejectInferrer(ABC):
    """
    Abstract base class for reject inference methods.

    Reject inference is used to address selection bias when building
    credit models on approved loans only.

    Parameters
    ----------
    target_col : str, default 'target'
        Name of the target column.
    score_col : str, default 'score'
        Name of the score/probability column.

    Methods
    -------
    infer(df_approved, df_rejected, score_col)
        Apply reject inference.
    """

    def __init__(
        self,
        target_col: str = 'target',
        score_col: str = 'score',
        score_direction: str = 'high_good',
        random_state: Optional[int] = None,
    ):
        """
        Initialize RejectInferrer.

        Parameters
        ----------
        target_col : str, default 'target'
            Target column name.
        score_col : str, default 'score'
            Score column name.
        """
        if score_direction not in {"high_good", "high_bad"}:
            raise ValueError("score_direction must be 'high_good' or 'high_bad'")
        self.target_col = target_col
        self.score_col = score_col
        self.score_direction = score_direction
        self.random_state = random_state

    def _bad_probability(self, score: pd.Series) -> pd.Series:
        prob = pd.to_numeric(score, errors="coerce").astype(float)
        if self.score_direction == "high_good":
            prob = 1.0 - prob
        return prob.clip(0.0, 1.0)

    def _rng(self) -> np.random.Generator:
        return np.random.default_rng(self.random_state)

    def _filter_nan_bad_probability(
        self,
        df_rejected: pd.DataFrame,
        p_bad: pd.Series,
        score_col: str,
        nan_score_policy: str,
    ) -> Tuple[pd.DataFrame, pd.Series]:
        if nan_score_policy not in {"drop", "raise", "fill_0.5"}:
            raise ValueError("nan_score_policy must be one of 'drop', 'raise', or 'fill_0.5'")

        stats = warn_if_nan_ratio_exceeds(
            p_bad,
            threshold=0.01,
            context="reject inference bad probability",
        )
        n_nan = int(stats["n_nan"])
        if n_nan == 0:
            return df_rejected, p_bad

        n_total = int(stats["n_total"])
        if nan_score_policy == "raise":
            raise ValueError(
                f"Reject inference bad probability contains {n_nan}/{n_total} NaN values "
                f"from score_col={score_col!r}."
            )
        if nan_score_policy == "fill_0.5":
            return df_rejected, p_bad.fillna(0.5)

        keep = p_bad.notna()
        dropped = int((~keep).sum())
        warnings.warn(
            f"Reject inference dropped {dropped}/{n_total} rejected rows with NaN bad probability "
            f"from score_col={score_col!r}. Set nan_score_policy='fill_0.5' to reproduce "
            f"the legacy 50/50 behaviour.",
            RuntimeWarning,
            stacklevel=2,
        )
        return df_rejected.loc[keep].copy(), p_bad.loc[keep]

    @abstractmethod
    def infer(self, df_approved: pd.DataFrame,
              df_rejected: pd.DataFrame,
              score_col: Optional[str] = None) -> pd.DataFrame:
        """
        Apply reject inference.

        Parameters
        ----------
        df_approved : pandas.DataFrame
            DataFrame with approved applications (has target).
        df_rejected : pandas.DataFrame
            DataFrame with rejected applications (no target).
        score_col : str, optional
            Score column name.

        Returns
        -------
        pandas.DataFrame
            Combined DataFrame with inferred targets for rejected applications.
        """
        pass

infer abstractmethod

infer(df_approved: DataFrame, df_rejected: DataFrame, score_col: Optional[str] = None) -> DataFrame

Apply reject inference.

参数：

名称	类型	描述	默认
df_approved	DataFrame	DataFrame with approved applications (has target).	必需
df_rejected	DataFrame	DataFrame with rejected applications (no target).	必需
score_col	str	Score column name.	None

返回：

类型	描述
DataFrame	Combined DataFrame with inferred targets for rejected applications.

源代码位于： Modeling_Tool/Sample/Reject_Infer.py

@abstractmethod
def infer(self, df_approved: pd.DataFrame,
          df_rejected: pd.DataFrame,
          score_col: Optional[str] = None) -> pd.DataFrame:
    """
    Apply reject inference.

    Parameters
    ----------
    df_approved : pandas.DataFrame
        DataFrame with approved applications (has target).
    df_rejected : pandas.DataFrame
        DataFrame with rejected applications (no target).
    score_col : str, optional
        Score column name.

    Returns
    -------
    pandas.DataFrame
        Combined DataFrame with inferred targets for rejected applications.
    """
    pass

SimpleAugmentInferrer

Bases: RejectInferrer

Simple augmentation reject inference method.

Assigns the average bad rate from approved applications to all rejected applications.

参数：

名称	类型	描述	默认
bad_rate	float	Override bad rate to use.	None

示例：

>>> inferrer = SimpleAugmentInferrer()
>>> df_combined = inferrer.infer(df_approved, df_rejected)

源代码位于： Modeling_Tool/Sample/Reject_Infer.py

class SimpleAugmentInferrer(RejectInferrer):
    """
    Simple augmentation reject inference method.

    Assigns the average bad rate from approved applications
    to all rejected applications.

    Parameters
    ----------
    bad_rate : float, optional
        Override bad rate to use.

    Examples
    --------
    >>> inferrer = SimpleAugmentInferrer()
    >>> df_combined = inferrer.infer(df_approved, df_rejected)
    """

    def __init__(
        self,
        target_col: str = 'target',
        score_col: str = 'score',
        bad_rate: Optional[float] = None,
        score_direction: str = 'high_good',
        random_state: Optional[int] = None,
    ):
        """
        Initialize SimpleAugmentInferrer.
        """
        super().__init__(target_col, score_col, score_direction=score_direction, random_state=random_state)
        self.bad_rate = bad_rate

    def infer(self, df_approved: pd.DataFrame,
              df_rejected: pd.DataFrame,
              score_col: Optional[str] = None) -> pd.DataFrame:
        """
        Apply simple augmentation.

        Parameters
        ----------
        df_approved : pandas.DataFrame
            Approved applications.
        df_rejected : pandas.DataFrame
            Rejected applications.
        score_col : str, optional
            Score column.

        Returns
        -------
        pandas.DataFrame
            Combined data with inferred targets.
        """
        score_col = score_col or self.score_col

        if self.bad_rate is None:
            bad_rate = df_approved[self.target_col].mean()
        else:
            bad_rate = self.bad_rate

        inferred_target = self._rng().binomial(1, bad_rate, len(df_rejected))

        df_rejected_copy = df_rejected.copy()
        df_rejected_copy[self.target_col] = inferred_target

        return pd.concat([df_approved, df_rejected_copy], ignore_index=True)

infer

infer(df_approved: DataFrame, df_rejected: DataFrame, score_col: Optional[str] = None) -> DataFrame

Apply simple augmentation.

参数：

名称	类型	描述	默认
df_approved	DataFrame	Approved applications.	必需
df_rejected	DataFrame	Rejected applications.	必需
score_col	str	Score column.	None

返回：

类型	描述
DataFrame	Combined data with inferred targets.

源代码位于： Modeling_Tool/Sample/Reject_Infer.py

def infer(self, df_approved: pd.DataFrame,
          df_rejected: pd.DataFrame,
          score_col: Optional[str] = None) -> pd.DataFrame:
    """
    Apply simple augmentation.

    Parameters
    ----------
    df_approved : pandas.DataFrame
        Approved applications.
    df_rejected : pandas.DataFrame
        Rejected applications.
    score_col : str, optional
        Score column.

    Returns
    -------
    pandas.DataFrame
        Combined data with inferred targets.
    """
    score_col = score_col or self.score_col

    if self.bad_rate is None:
        bad_rate = df_approved[self.target_col].mean()
    else:
        bad_rate = self.bad_rate

    inferred_target = self._rng().binomial(1, bad_rate, len(df_rejected))

    df_rejected_copy = df_rejected.copy()
    df_rejected_copy[self.target_col] = inferred_target

    return pd.concat([df_approved, df_rejected_copy], ignore_index=True)

HardCutoffInferrer

Bases: RejectInferrer

Hard cutoff reject inference method.

Assigns all rejected applications below a score threshold as bad (target=1), and all above as good (target=0).

参数：

名称	类型	描述	默认
cutoff	float	Score cutoff threshold.	0.5

示例：

>>> inferrer = HardCutoffInferrer(cutoff=0.3)
>>> df_combined = inferrer.infer(df_approved, df_rejected, 'probability')

源代码位于： Modeling_Tool/Sample/Reject_Infer.py

class HardCutoffInferrer(RejectInferrer):
    """
    Hard cutoff reject inference method.

    Assigns all rejected applications below a score threshold
    as bad (target=1), and all above as good (target=0).

    Parameters
    ----------
    cutoff : float, default 0.5
        Score cutoff threshold.

    Examples
    --------
    >>> inferrer = HardCutoffInferrer(cutoff=0.3)
    >>> df_combined = inferrer.infer(df_approved, df_rejected, 'probability')
    """

    def __init__(
        self,
        target_col: str = 'target',
        score_col: str = 'score',
        cutoff: float = 0.5,
        score_direction: str = 'high_good',
        random_state: Optional[int] = None,
    ):
        """
        Initialize HardCutoffInferrer.
        """
        super().__init__(target_col, score_col, score_direction=score_direction, random_state=random_state)
        self.cutoff = cutoff

    def infer(self, df_approved: pd.DataFrame,
              df_rejected: pd.DataFrame,
              score_col: Optional[str] = None) -> pd.DataFrame:
        """
        Apply hard cutoff inference.

        NaN-scored rejects (0.4.2 fix, N17)
        ------------------------------------
        Before 0.4.2, any rejected row whose ``score_col`` value was NaN was
        silently labelled ``0`` ("good") because ``NaN >= cutoff`` and
        ``NaN <= cutoff`` both evaluate to ``False``, which the old code then
        cast through ``.astype(int)``. That silently seeded the training set
        with as many synthetic "good" rejects as there were NaN scores, biasing
        every downstream model that consumed the combined frame.

        From 0.4.2:

        * If **every** reject has a NaN score, ``ValueError`` is raised with the
          missing count — the caller has no signal to infer from and must
          re-check upstream scoring.
        * If **some** rejects have NaN scores, a ``RuntimeWarning`` is emitted
          naming the NaN count and share, and those rows carry ``target = NaN``
          in the returned frame (not ``0``). Callers that intersect on the
          target column will drop them explicitly instead of training on
          silently-fabricated labels.
        * Rejects with finite scores are labelled exactly as before.

        Parameters
        ----------
        df_approved : pandas.DataFrame
            Approved applications.
        df_rejected : pandas.DataFrame
            Rejected applications.
        score_col : str, optional
            Score column.

        Returns
        -------
        pandas.DataFrame
            Combined data with inferred targets. Rejects whose score was NaN
            (see above) will have ``target = NaN`` rather than ``0``.
        """
        score_col = score_col or self.score_col

        df_rejected_copy = df_rejected.copy()

        scores = pd.to_numeric(df_rejected_copy[score_col], errors="coerce")
        nan_mask = scores.isna()
        n_nan = int(nan_mask.sum())
        n_total = int(len(scores))

        if n_total > 0 and n_nan == n_total:
            raise ValueError(
                f"HardCutoffInferrer.infer: every rejected row (n={n_total}) has a "
                f"NaN value in score_col={score_col!r}. Cannot infer targets. "
                f"Check the upstream prescore step or pass a different score column."
            )

        if n_nan > 0:
            share = n_nan / n_total if n_total else 0.0
            warnings.warn(
                f"HardCutoffInferrer.infer: {n_nan}/{n_total} ({share:.1%}) rejected "
                f"rows have NaN in score_col={score_col!r}. These rows now carry "
                f"target=NaN in the returned frame (pre-0.4.2 they were silently "
                f"labelled 0). Drop them explicitly downstream if the training step "
                f"cannot handle NaN targets.",
                RuntimeWarning,
                stacklevel=2,
            )

        if self.score_direction == "high_bad":
            labels = (scores >= self.cutoff)
        else:
            labels = (scores <= self.cutoff)

        # Preserve NaN-ness explicitly: labels for NaN scores become NaN, not
        # False -> 0. Cast the finite side back to int without losing the NaN
        # rows.
        target_series = labels.astype("float")
        target_series[nan_mask] = np.nan
        df_rejected_copy[self.target_col] = target_series

        return pd.concat([df_approved, df_rejected_copy], ignore_index=True)

infer

infer(df_approved: DataFrame, df_rejected: DataFrame, score_col: Optional[str] = None) -> DataFrame

Apply hard cutoff inference.

NaN-scored rejects (0.4.2 fix, N17)

Before 0.4.2, any rejected row whose score_col value was NaN was silently labelled 0 ("good") because NaN >= cutoff and NaN <= cutoff both evaluate to False, which the old code then cast through .astype(int). That silently seeded the training set with as many synthetic "good" rejects as there were NaN scores, biasing every downstream model that consumed the combined frame.

From 0.4.2:

• If every reject has a NaN score, ValueError is raised with the missing count — the caller has no signal to infer from and must re-check upstream scoring.
• If some rejects have NaN scores, a RuntimeWarning is emitted naming the NaN count and share, and those rows carry target = NaN in the returned frame (not 0). Callers that intersect on the target column will drop them explicitly instead of training on silently-fabricated labels.
• Rejects with finite scores are labelled exactly as before.

参数：

名称	类型	描述	默认
df_approved	DataFrame	Approved applications.	必需
df_rejected	DataFrame	Rejected applications.	必需
score_col	str	Score column.	None

返回：

类型	描述
DataFrame	Combined data with inferred targets. Rejects whose score was NaN (see above) will have target = NaN rather than 0.

源代码位于： Modeling_Tool/Sample/Reject_Infer.py

def infer(self, df_approved: pd.DataFrame,
          df_rejected: pd.DataFrame,
          score_col: Optional[str] = None) -> pd.DataFrame:
    """
    Apply hard cutoff inference.

    NaN-scored rejects (0.4.2 fix, N17)
    ------------------------------------
    Before 0.4.2, any rejected row whose ``score_col`` value was NaN was
    silently labelled ``0`` ("good") because ``NaN >= cutoff`` and
    ``NaN <= cutoff`` both evaluate to ``False``, which the old code then
    cast through ``.astype(int)``. That silently seeded the training set
    with as many synthetic "good" rejects as there were NaN scores, biasing
    every downstream model that consumed the combined frame.

    From 0.4.2:

    * If **every** reject has a NaN score, ``ValueError`` is raised with the
      missing count — the caller has no signal to infer from and must
      re-check upstream scoring.
    * If **some** rejects have NaN scores, a ``RuntimeWarning`` is emitted
      naming the NaN count and share, and those rows carry ``target = NaN``
      in the returned frame (not ``0``). Callers that intersect on the
      target column will drop them explicitly instead of training on
      silently-fabricated labels.
    * Rejects with finite scores are labelled exactly as before.

    Parameters
    ----------
    df_approved : pandas.DataFrame
        Approved applications.
    df_rejected : pandas.DataFrame
        Rejected applications.
    score_col : str, optional
        Score column.

    Returns
    -------
    pandas.DataFrame
        Combined data with inferred targets. Rejects whose score was NaN
        (see above) will have ``target = NaN`` rather than ``0``.
    """
    score_col = score_col or self.score_col

    df_rejected_copy = df_rejected.copy()

    scores = pd.to_numeric(df_rejected_copy[score_col], errors="coerce")
    nan_mask = scores.isna()
    n_nan = int(nan_mask.sum())
    n_total = int(len(scores))

    if n_total > 0 and n_nan == n_total:
        raise ValueError(
            f"HardCutoffInferrer.infer: every rejected row (n={n_total}) has a "
            f"NaN value in score_col={score_col!r}. Cannot infer targets. "
            f"Check the upstream prescore step or pass a different score column."
        )

    if n_nan > 0:
        share = n_nan / n_total if n_total else 0.0
        warnings.warn(
            f"HardCutoffInferrer.infer: {n_nan}/{n_total} ({share:.1%}) rejected "
            f"rows have NaN in score_col={score_col!r}. These rows now carry "
            f"target=NaN in the returned frame (pre-0.4.2 they were silently "
            f"labelled 0). Drop them explicitly downstream if the training step "
            f"cannot handle NaN targets.",
            RuntimeWarning,
            stacklevel=2,
        )

    if self.score_direction == "high_bad":
        labels = (scores >= self.cutoff)
    else:
        labels = (scores <= self.cutoff)

    # Preserve NaN-ness explicitly: labels for NaN scores become NaN, not
    # False -> 0. Cast the finite side back to int without losing the NaN
    # rows.
    target_series = labels.astype("float")
    target_series[nan_mask] = np.nan
    df_rejected_copy[self.target_col] = target_series

    return pd.concat([df_approved, df_rejected_copy], ignore_index=True)

FuzzyAugmentInferrer

Bases: RejectInferrer

Fuzzy augmentation reject inference method.

Weights approved applications based on their predicted probability and creates pseudo-target values for rejected applications.

参数：

名称	类型	描述	默认
weight_factor	float	Factor to adjust weights.	1.0

示例：

>>> inferrer = FuzzyAugmentInferrer(weight_factor=0.9)
>>> df_combined = inferrer.infer(df_approved, df_rejected, 'probability')

源代码位于： Modeling_Tool/Sample/Reject_Infer.py

class FuzzyAugmentInferrer(RejectInferrer):
    """
    Fuzzy augmentation reject inference method.

    Weights approved applications based on their predicted probability
    and creates pseudo-target values for rejected applications.

    Parameters
    ----------
    weight_factor : float, default 1.0
        Factor to adjust weights.

    Examples
    --------
    >>> inferrer = FuzzyAugmentInferrer(weight_factor=0.9)
    >>> df_combined = inferrer.infer(df_approved, df_rejected, 'probability')
    """

    def __init__(
        self,
        target_col: str = 'target',
        score_col: str = 'score',
        weight_factor: float = 1.0,
        score_direction: str = 'high_good',
        random_state: Optional[int] = None,
        nan_score_policy: str = "drop",
    ):
        """
        Initialize FuzzyAugmentInferrer.
        """
        super().__init__(target_col, score_col, score_direction=score_direction, random_state=random_state)
        self.weight_factor = weight_factor
        self.nan_score_policy = nan_score_policy

    def infer(self, df_approved: pd.DataFrame,
              df_rejected: pd.DataFrame,
              score_col: Optional[str] = None) -> pd.DataFrame:
        """
        Apply fuzzy augmentation.

        Parameters
        ----------
        df_approved : pandas.DataFrame
            Approved applications.
        df_rejected : pandas.DataFrame
            Rejected applications.
        score_col : str, optional
            Score column.

        Returns
        -------
        pandas.DataFrame
            Combined data with inferred targets.
        """
        score_col = score_col or self.score_col

        df_approved_copy = df_approved.copy()
        df_approved_copy['_weight'] = 1.0

        p_bad = self._bad_probability(df_rejected[score_col])
        df_rejected, p_bad = self._filter_nan_bad_probability(
            df_rejected,
            p_bad,
            score_col=score_col,
            nan_score_policy=self.nan_score_policy,
        )
        bad_copy = df_rejected.copy()
        bad_copy[self.target_col] = 1
        bad_copy['_weight'] = p_bad.to_numpy(dtype=float) * float(self.weight_factor)

        good_copy = df_rejected.copy()
        good_copy[self.target_col] = 0
        good_copy['_weight'] = (1.0 - p_bad.to_numpy(dtype=float)) * float(self.weight_factor)

        return pd.concat([df_approved_copy, bad_copy, good_copy], ignore_index=True)

infer

infer(df_approved: DataFrame, df_rejected: DataFrame, score_col: Optional[str] = None) -> DataFrame

Apply fuzzy augmentation.

参数：

名称	类型	描述	默认
df_approved	DataFrame	Approved applications.	必需
df_rejected	DataFrame	Rejected applications.	必需
score_col	str	Score column.	None

返回：

类型	描述
DataFrame	Combined data with inferred targets.

源代码位于： Modeling_Tool/Sample/Reject_Infer.py

def infer(self, df_approved: pd.DataFrame,
          df_rejected: pd.DataFrame,
          score_col: Optional[str] = None) -> pd.DataFrame:
    """
    Apply fuzzy augmentation.

    Parameters
    ----------
    df_approved : pandas.DataFrame
        Approved applications.
    df_rejected : pandas.DataFrame
        Rejected applications.
    score_col : str, optional
        Score column.

    Returns
    -------
    pandas.DataFrame
        Combined data with inferred targets.
    """
    score_col = score_col or self.score_col

    df_approved_copy = df_approved.copy()
    df_approved_copy['_weight'] = 1.0

    p_bad = self._bad_probability(df_rejected[score_col])
    df_rejected, p_bad = self._filter_nan_bad_probability(
        df_rejected,
        p_bad,
        score_col=score_col,
        nan_score_policy=self.nan_score_policy,
    )
    bad_copy = df_rejected.copy()
    bad_copy[self.target_col] = 1
    bad_copy['_weight'] = p_bad.to_numpy(dtype=float) * float(self.weight_factor)

    good_copy = df_rejected.copy()
    good_copy[self.target_col] = 0
    good_copy['_weight'] = (1.0 - p_bad.to_numpy(dtype=float)) * float(self.weight_factor)

    return pd.concat([df_approved_copy, bad_copy, good_copy], ignore_index=True)

ParcelingInferrer

Bases: RejectInferrer

Parceling reject inference method.

Splits rejected applications into parcels based on score bands and assigns average bad rate from approved applications in each parcel.

参数：

名称	类型	描述	默认
n_parcels	int	Number of score parcels.	10

示例：

>>> inferrer = ParcelingInferrer(n_parcels=5)
>>> df_combined = inferrer.infer(df_approved, df_rejected, 'score')

源代码位于： Modeling_Tool/Sample/Reject_Infer.py

class ParcelingInferrer(RejectInferrer):
    """
    Parceling reject inference method.

    Splits rejected applications into parcels based on score bands
    and assigns average bad rate from approved applications in
    each parcel.

    Parameters
    ----------
    n_parcels : int, default 10
        Number of score parcels.

    Examples
    --------
    >>> inferrer = ParcelingInferrer(n_parcels=5)
    >>> df_combined = inferrer.infer(df_approved, df_rejected, 'score')
    """

    def __init__(
        self,
        target_col: str = 'target',
        score_col: str = 'score',
        n_parcels: int = 10,
        score_direction: str = 'high_good',
        random_state: Optional[int] = None,
    ):
        """
        Initialize ParcelingInferrer.
        """
        super().__init__(target_col, score_col, score_direction=score_direction, random_state=random_state)
        self.n_parcels = n_parcels
        self.parcel_rates_ = None
        self.parcel_edges_ = None

    def infer(self, df_approved: pd.DataFrame,
              df_rejected: pd.DataFrame,
              score_col: Optional[str] = None) -> pd.DataFrame:
        """
        Apply parceling inference.

        Parameters
        ----------
        df_approved : pandas.DataFrame
            Approved applications.
        df_rejected : pandas.DataFrame
            Rejected applications.
        score_col : str, optional
            Score column.

        Returns
        -------
        pandas.DataFrame
            Combined data with inferred targets.
        """
        score_col = score_col or self.score_col

        df_approved_copy = df_approved.copy()
        df_rejected_copy = df_rejected.copy()

        score_for_bins = self._bad_probability(df_approved_copy[score_col])
        try:
            approved_parcel, edges = pd.qcut(
                score_for_bins,
                q=self.n_parcels,
                labels=False,
                retbins=True,
                duplicates='drop',
            )
        except ValueError:
            approved_parcel = pd.Series(0, index=df_approved_copy.index)
            edges = np.array([score_for_bins.min(), score_for_bins.max()], dtype=float)

        if len(edges) < 2 or not np.isfinite(edges).all() or edges[0] == edges[-1]:
            approved_parcel = pd.Series(0, index=df_approved_copy.index)
            edges = np.array([-np.inf, np.inf], dtype=float)
        else:
            edges = np.asarray(edges, dtype=float)
            edges[0] = -np.inf
            edges[-1] = np.inf

        df_approved_copy['_parcel'] = approved_parcel
        parcel_rates = df_approved_copy.groupby('_parcel')[self.target_col].mean()
        self.parcel_rates_ = parcel_rates
        self.parcel_edges_ = edges

        rejected_score_for_bins = self._bad_probability(df_rejected_copy[score_col])
        df_rejected_copy['_parcel'] = pd.cut(
            rejected_score_for_bins,
            bins=edges,
            labels=False,
            include_lowest=True,
        )

        p_bad = df_rejected_copy['_parcel'].map(parcel_rates).fillna(df_approved_copy[self.target_col].mean())
        df_rejected_copy[self.target_col] = self._rng().binomial(1, p_bad.clip(0.0, 1.0).to_numpy(dtype=float))

        df_approved_copy = df_approved_copy.drop('_parcel', axis=1)
        df_rejected_copy = df_rejected_copy.drop('_parcel', axis=1)

        return pd.concat([df_approved_copy, df_rejected_copy], ignore_index=True)

infer

infer(df_approved: DataFrame, df_rejected: DataFrame, score_col: Optional[str] = None) -> DataFrame

Apply parceling inference.

参数：

名称	类型	描述	默认
df_approved	DataFrame	Approved applications.	必需
df_rejected	DataFrame	Rejected applications.	必需
score_col	str	Score column.	None

返回：

类型	描述
DataFrame	Combined data with inferred targets.

源代码位于： Modeling_Tool/Sample/Reject_Infer.py

def infer(self, df_approved: pd.DataFrame,
          df_rejected: pd.DataFrame,
          score_col: Optional[str] = None) -> pd.DataFrame:
    """
    Apply parceling inference.

    Parameters
    ----------
    df_approved : pandas.DataFrame
        Approved applications.
    df_rejected : pandas.DataFrame
        Rejected applications.
    score_col : str, optional
        Score column.

    Returns
    -------
    pandas.DataFrame
        Combined data with inferred targets.
    """
    score_col = score_col or self.score_col

    df_approved_copy = df_approved.copy()
    df_rejected_copy = df_rejected.copy()

    score_for_bins = self._bad_probability(df_approved_copy[score_col])
    try:
        approved_parcel, edges = pd.qcut(
            score_for_bins,
            q=self.n_parcels,
            labels=False,
            retbins=True,
            duplicates='drop',
        )
    except ValueError:
        approved_parcel = pd.Series(0, index=df_approved_copy.index)
        edges = np.array([score_for_bins.min(), score_for_bins.max()], dtype=float)

    if len(edges) < 2 or not np.isfinite(edges).all() or edges[0] == edges[-1]:
        approved_parcel = pd.Series(0, index=df_approved_copy.index)
        edges = np.array([-np.inf, np.inf], dtype=float)
    else:
        edges = np.asarray(edges, dtype=float)
        edges[0] = -np.inf
        edges[-1] = np.inf

    df_approved_copy['_parcel'] = approved_parcel
    parcel_rates = df_approved_copy.groupby('_parcel')[self.target_col].mean()
    self.parcel_rates_ = parcel_rates
    self.parcel_edges_ = edges

    rejected_score_for_bins = self._bad_probability(df_rejected_copy[score_col])
    df_rejected_copy['_parcel'] = pd.cut(
        rejected_score_for_bins,
        bins=edges,
        labels=False,
        include_lowest=True,
    )

    p_bad = df_rejected_copy['_parcel'].map(parcel_rates).fillna(df_approved_copy[self.target_col].mean())
    df_rejected_copy[self.target_col] = self._rng().binomial(1, p_bad.clip(0.0, 1.0).to_numpy(dtype=float))

    df_approved_copy = df_approved_copy.drop('_parcel', axis=1)
    df_rejected_copy = df_rejected_copy.drop('_parcel', axis=1)

    return pd.concat([df_approved_copy, df_rejected_copy], ignore_index=True)

RejectInferenceFactory

Factory class for creating reject inference methods.

示例：

>>> inferrer = RejectInferenceFactory.create('parceling', n_parcels=5)
>>> inferrer = RejectInferenceFactory.create('fuzzy', weight_factor=0.9)

源代码位于： Modeling_Tool/Sample/Reject_Infer.py

class RejectInferenceFactory:
    """
    Factory class for creating reject inference methods.

    Examples
    --------
    >>> inferrer = RejectInferenceFactory.create('parceling', n_parcels=5)
    >>> inferrer = RejectInferenceFactory.create('fuzzy', weight_factor=0.9)
    """

    _methods = {
        'simple': SimpleAugmentInferrer,
        'augment': SimpleAugmentInferrer,
        'hard': HardCutoffInferrer,
        'hardcutoff': HardCutoffInferrer,
        'fuzzy': FuzzyAugmentInferrer,
        'parceling': ParcelingInferrer,
        'parcel': ParcelingInferrer
    }

    @classmethod
    def create(cls, method: str = 'parceling', **kwargs) -> RejectInferrer:
        """
        Create a reject inference method.

        Parameters
        ----------
        method : str, default 'parceling'
            Method name.
        **kwargs
            Additional parameters for the method.

        Returns
        -------
        RejectInferrer
            Instantiated reject inferrer.

        Raises
        ------
        ValueError
            If method name is not recognized.
        """
        method_lower = method.lower()
        if method_lower not in cls._methods:
            raise ValueError(
                f"Unknown method '{method}'. "
                f"Available: {list(set(cls._methods.keys()))}"
            )
        return cls._methods[method_lower](**kwargs)

    @classmethod
    def available_methods(cls) -> List[str]:
        """
        Get list of available methods.

        Returns
        -------
        list of str
            Available method names.
        """
        return list(set(cls._methods.keys()))

create classmethod

create(method: str = 'parceling', **kwargs) -> RejectInferrer

Create a reject inference method.

参数：

名称	类型	描述	默认
method	str	Method name.	'parceling'
**kwargs		Additional parameters for the method.	{}

返回：

类型	描述
RejectInferrer	Instantiated reject inferrer.

引发：

类型	描述
ValueError	If method name is not recognized.

源代码位于： Modeling_Tool/Sample/Reject_Infer.py

@classmethod
def create(cls, method: str = 'parceling', **kwargs) -> RejectInferrer:
    """
    Create a reject inference method.

    Parameters
    ----------
    method : str, default 'parceling'
        Method name.
    **kwargs
        Additional parameters for the method.

    Returns
    -------
    RejectInferrer
        Instantiated reject inferrer.

    Raises
    ------
    ValueError
        If method name is not recognized.
    """
    method_lower = method.lower()
    if method_lower not in cls._methods:
        raise ValueError(
            f"Unknown method '{method}'. "
            f"Available: {list(set(cls._methods.keys()))}"
        )
    return cls._methods[method_lower](**kwargs)

available_methods classmethod

available_methods() -> List[str]

Get list of available methods.

返回：

类型	描述
list of str	Available method names.

源代码位于： Modeling_Tool/Sample/Reject_Infer.py

@classmethod
def available_methods(cls) -> List[str]:
    """
    Get list of available methods.

    Returns
    -------
    list of str
        Available method names.
    """
    return list(set(cls._methods.keys()))

分布适配 — Distribution_Adaptation

Distribution_Adaptation

DistributionAdaptation

源代码位于： Modeling_Tool/Sample/Distribution_Adaptation.py

class DistributionAdaptation:
    def __init__(self, method='density_ratio'):
        """
        初始化分布适配器

        Parameters:
        -----------
        method: str
            'density_ratio': 密度比估计
            'kl_divergence': KL散度加权
            'covariate_shift': 协变量偏移修正
        """
        self.method = method
        self.sample_weights = None
        self.feature_importances = None

    def estimate_density_ratio(self, X_train, X_oot):
        """
        使用密度比估计方法计算样本权重
        KLIEP/KMM等方法的简化实现
        """
        from sklearn.neighbors import KernelDensity

        # 使用KDE估计密度
        kde_train = KernelDensity(kernel='gaussian', bandwidth=0.5)
        kde_oot = KernelDensity(kernel='gaussian', bandwidth=0.5)

        kde_train.fit(X_train)
        kde_oot.fit(X_oot)

        # 计算密度比: p_oot(x) / p_train(x)
        log_density_train = kde_train.score_samples(X_train)
        log_density_oot = kde_oot.score_samples(X_train)

        # 防止除零和数值不稳定
        density_ratio = np.exp(log_density_oot - log_density_train)

        # 截断处理异常值
        density_ratio = np.clip(density_ratio, 0.1, 10)

        # 归一化
        density_ratio = density_ratio / density_ratio.mean()

        return density_ratio

    def covariate_shift_weighting(self, X_train, X_oot):
        """
        使用领域分类器估计样本重要性权重
        """
        from sklearn.linear_model import LogisticRegression

        # 创建领域标签: 训练集为0, OOT为1
        n_train = len(X_train)
        n_oot = len(X_oot)

        X_combined = np.vstack([X_train, X_oot])
        y_domain = np.hstack([np.zeros(n_train), np.ones(n_oot)])

        # 训练领域分类器
        domain_classifier = LogisticRegression(
            C=1.0, max_iter=1000, random_state=42
        )
        domain_classifier.fit(X_combined, y_domain)

        # 预测训练样本来自OOT的概率
        p_oot = domain_classifier.predict_proba(X_train)[:, 1]

        # 计算权重: p(oot|x) / p(train|x)
        # 防止数值不稳定
        epsilon = 1e-10
        weights = p_oot / (1 - p_oot + epsilon)

        # 使用beta分布平滑权重
        weights = np.clip(weights, 0.1, 10)
        weights = weights / weights.mean()

        return weights

    def fit(self, X_train, X_oot, y_train=None):
        """
        计算适应OOT分布的样本权重
        """
        if self.method == 'density_ratio':
            self.sample_weights = self.estimate_density_ratio(X_train, X_oot)
        elif self.method == 'covariate_shift':
            self.sample_weights = self.covariate_shift_weighting(X_train, X_oot)
        else:
            # 默认使用混合方法
            w1 = self.estimate_density_ratio(X_train, X_oot)
            w2 = self.covariate_shift_weighting(X_train, X_oot)
            self.sample_weights = (w1 + w2) / 2

        return self

    def get_weights(self):
        """返回样本权重"""
        return self.sample_weights

    def visualize_distribution_comparison(self, X_train, X_oot, features=None, n_features=5):
        """
        可视化训练集和OOT集的分布差异
        """
        if features is None:
            # 选择方差最大的特征
            variances = np.var(X_train, axis=0)
            features = np.argsort(variances)[-n_features:]

        fig, axes = plt.subplots(1, n_features, figsize=(5*n_features, 4))

        for idx, feature_idx in enumerate(features[:n_features]):
            ax = axes[idx] if n_features > 1 else axes

            # 绘制分布
            sns.kdeplot(X_train[:, feature_idx], ax=ax, label='Train', fill=True, alpha=0.5)
            sns.kdeplot(X_oot[:, feature_idx], ax=ax, label='OOT', fill=True, alpha=0.5)

            ax.set_title(f'Feature {feature_idx}')
            ax.legend()
            ax.set_xlabel('Value')
            ax.set_ylabel('Density')

        plt.tight_layout()
        plt.show()

estimate_density_ratio

estimate_density_ratio(X_train, X_oot)

使用密度比估计方法计算样本权重 KLIEP/KMM等方法的简化实现

源代码位于： Modeling_Tool/Sample/Distribution_Adaptation.py

def estimate_density_ratio(self, X_train, X_oot):
    """
    使用密度比估计方法计算样本权重
    KLIEP/KMM等方法的简化实现
    """
    from sklearn.neighbors import KernelDensity

    # 使用KDE估计密度
    kde_train = KernelDensity(kernel='gaussian', bandwidth=0.5)
    kde_oot = KernelDensity(kernel='gaussian', bandwidth=0.5)

    kde_train.fit(X_train)
    kde_oot.fit(X_oot)

    # 计算密度比: p_oot(x) / p_train(x)
    log_density_train = kde_train.score_samples(X_train)
    log_density_oot = kde_oot.score_samples(X_train)

    # 防止除零和数值不稳定
    density_ratio = np.exp(log_density_oot - log_density_train)

    # 截断处理异常值
    density_ratio = np.clip(density_ratio, 0.1, 10)

    # 归一化
    density_ratio = density_ratio / density_ratio.mean()

    return density_ratio

covariate_shift_weighting

covariate_shift_weighting(X_train, X_oot)

使用领域分类器估计样本重要性权重

源代码位于： Modeling_Tool/Sample/Distribution_Adaptation.py

def covariate_shift_weighting(self, X_train, X_oot):
    """
    使用领域分类器估计样本重要性权重
    """
    from sklearn.linear_model import LogisticRegression

    # 创建领域标签: 训练集为0, OOT为1
    n_train = len(X_train)
    n_oot = len(X_oot)

    X_combined = np.vstack([X_train, X_oot])
    y_domain = np.hstack([np.zeros(n_train), np.ones(n_oot)])

    # 训练领域分类器
    domain_classifier = LogisticRegression(
        C=1.0, max_iter=1000, random_state=42
    )
    domain_classifier.fit(X_combined, y_domain)

    # 预测训练样本来自OOT的概率
    p_oot = domain_classifier.predict_proba(X_train)[:, 1]

    # 计算权重: p(oot|x) / p(train|x)
    # 防止数值不稳定
    epsilon = 1e-10
    weights = p_oot / (1 - p_oot + epsilon)

    # 使用beta分布平滑权重
    weights = np.clip(weights, 0.1, 10)
    weights = weights / weights.mean()

    return weights

fit

fit(X_train, X_oot, y_train=None)

计算适应OOT分布的样本权重

源代码位于： Modeling_Tool/Sample/Distribution_Adaptation.py

def fit(self, X_train, X_oot, y_train=None):
    """
    计算适应OOT分布的样本权重
    """
    if self.method == 'density_ratio':
        self.sample_weights = self.estimate_density_ratio(X_train, X_oot)
    elif self.method == 'covariate_shift':
        self.sample_weights = self.covariate_shift_weighting(X_train, X_oot)
    else:
        # 默认使用混合方法
        w1 = self.estimate_density_ratio(X_train, X_oot)
        w2 = self.covariate_shift_weighting(X_train, X_oot)
        self.sample_weights = (w1 + w2) / 2

    return self

get_weights

get_weights()

返回样本权重

源代码位于： Modeling_Tool/Sample/Distribution_Adaptation.py

def get_weights(self):
    """返回样本权重"""
    return self.sample_weights

visualize_distribution_comparison

visualize_distribution_comparison(X_train, X_oot, features=None, n_features=5)

可视化训练集和OOT集的分布差异

源代码位于： Modeling_Tool/Sample/Distribution_Adaptation.py

def visualize_distribution_comparison(self, X_train, X_oot, features=None, n_features=5):
    """
    可视化训练集和OOT集的分布差异
    """
    if features is None:
        # 选择方差最大的特征
        variances = np.var(X_train, axis=0)
        features = np.argsort(variances)[-n_features:]

    fig, axes = plt.subplots(1, n_features, figsize=(5*n_features, 4))

    for idx, feature_idx in enumerate(features[:n_features]):
        ax = axes[idx] if n_features > 1 else axes

        # 绘制分布
        sns.kdeplot(X_train[:, feature_idx], ax=ax, label='Train', fill=True, alpha=0.5)
        sns.kdeplot(X_oot[:, feature_idx], ax=ax, label='OOT', fill=True, alpha=0.5)

        ax.set_title(f'Feature {feature_idx}')
        ax.legend()
        ax.set_xlabel('Value')
        ax.set_ylabel('Density')

    plt.tight_layout()
    plt.show()