Modeling_Tool.Sample

Sample management layer: splitting, stratified sampling, class balancing, reject inference, and distribution adaptation. User guides: Sample Management and Reject Inference and Distribution Adaptation.

Splitting and Sampling: 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.

Classes:

Name	Description
SampleSplitter : Split data into train/test samples.
StratifiedSampler : Stratified sampling with target balance control.
SampleBalancer : Balance samples using various techniques.

Examples:

>>> 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.

Parameters:

Name	Type	Description	Default
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

Attributes:

Name	Type	Description
train_index_	ndarray	Indices for training data.
test_index_	ndarray	Indices for testing data.

Methods:

Name	Description
split	Split data into train and test sets.
split_df	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)}")

Source code in 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.

        Parameters
        ----------
        index : pandas.Index, numpy.ndarray or list
            Row labels (or positions) to split.
        y : pandas.Series or numpy.ndarray
            Target values, one per label, used for stratification.
        test_size : float or None, default None
            Share of the labels that goes to the test side; None uses the ``test_size`` of the splitter.
        stratify : bool or None, default None
            Whether to stratify by ``y``; None uses the ``stratify`` setting of the splitter. Stratification is skipped
            when ``y`` has a single class.

        Returns
        -------
        tuple of numpy.ndarray
            ``(train_index, test_index)``. The same arrays are stored as ``train_index_`` and ``test_index_``.

        Raises
        ------
        ValueError
            If ``index`` and ``y`` have different lengths.
        """
        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 row POSITIONS and select with ``iloc``: selecting by label with ``df.loc`` returns every row that
        # carries a duplicated index label (a frame built with ``pd.concat`` has them), which duplicated rows and
        # leaked training rows into the test frame. For a unique index the partition is the same as before.
        train_pos, test_pos = self.split_indices(
            np.arange(len(df)),
            strat if strat is not None else y,
            test_size=test_size,
            stratify=self.stratify,
        )
        labels = df.index.to_numpy()
        self.train_index_ = labels[train_pos]
        self.test_index_ = labels[test_pos]

        train_df = df.iloc[train_pos].copy()
        test_df = df.iloc[test_pos].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.

Parameters:

Name	Type	Description	Default
X	DataFrame or ndarray	Features.	required
y	Series or ndarray	Target variable.	required
test_size	float	Override default test size.	None
stratify	bool	Override default stratify setting.	None

Returns:

Type	Description
tuple	(X_train, X_test, y_train, y_test)

Examples:

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

Source code in 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.

Parameters:

Name	Type	Description	Default
index	(Index, ndarray or list)	Row labels (or positions) to split.	required
y	Series or ndarray	Target values, one per label, used for stratification.	required
test_size	float or None	Share of the labels that goes to the test side; None uses the test_size of the splitter.	None
stratify	bool or None	Whether to stratify by y; None uses the stratify setting of the splitter. Stratification is skipped when y has a single class.	None

Returns:

Type	Description
tuple of numpy.ndarray	(train_index, test_index). The same arrays are stored as train_index_ and test_index_.

Raises:

Type	Description
ValueError	If index and y have different lengths.

Source code in 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.

    Parameters
    ----------
    index : pandas.Index, numpy.ndarray or list
        Row labels (or positions) to split.
    y : pandas.Series or numpy.ndarray
        Target values, one per label, used for stratification.
    test_size : float or None, default None
        Share of the labels that goes to the test side; None uses the ``test_size`` of the splitter.
    stratify : bool or None, default None
        Whether to stratify by ``y``; None uses the ``stratify`` setting of the splitter. Stratification is skipped
        when ``y`` has a single class.

    Returns
    -------
    tuple of numpy.ndarray
        ``(train_index, test_index)``. The same arrays are stored as ``train_index_`` and ``test_index_``.

    Raises
    ------
    ValueError
        If ``index`` and ``y`` have different lengths.
    """
    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.

Parameters:

Name	Type	Description	Default
df	DataFrame	Input DataFrame.	required
target	str	Target column name.	required
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

Returns:

Type	Description
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

Source code in 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 row POSITIONS and select with ``iloc``: selecting by label with ``df.loc`` returns every row that
    # carries a duplicated index label (a frame built with ``pd.concat`` has them), which duplicated rows and
    # leaked training rows into the test frame. For a unique index the partition is the same as before.
    train_pos, test_pos = self.split_indices(
        np.arange(len(df)),
        strat if strat is not None else y,
        test_size=test_size,
        stratify=self.stratify,
    )
    labels = df.index.to_numpy()
    self.train_index_ = labels[train_pos]
    self.test_index_ = labels[test_pos]

    train_df = df.iloc[train_pos].copy()
    test_df = df.iloc[test_pos].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.

Parameters:

Name	Type	Description	Default
target_rate	float	Target bad rate in the sample.	None
random_state	int	Random seed.	None

Methods:

Name	Description
sample	Perform stratified sampling.
balance	Balance sample by adjusting target distribution.

Examples:

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

Source code in 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.

Parameters:

Name	Type	Description	Default
df	DataFrame	Input data.	required
target	str	Target column name.	required
n_samples	int	Number of samples to draw.	None
sample_frac	float	Fraction of data to sample.	None

Returns:

Type	Description
DataFrame	Sampled DataFrame.

Examples:

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

Source code in 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.

Parameters:

Name	Type	Description	Default
df	DataFrame	Input data.	required
target	str	Target column name.	required
method	str	Balancing method: 'undersample', 'oversample', 'smote'.	'undersample'

Returns:

Type	Description
DataFrame	Balanced DataFrame.

Examples:

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

Source code in 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.

Parameters:

Name	Type	Description	Default
method	str	Balancing method.	'random'
target_ratio	float	Desired minority/majority ratio.	None
random_state	int	Random seed.	None

Methods:

Name	Description
fit_resample	Resample features and target.
get_balanced_indices	Get indices for balanced sampling.

Examples:

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

Source code in 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 = _build_imblearn_sampler(NearMiss, self.random_state, version=1)
        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 = _build_imblearn_sampler(TomekLinks, 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 = _build_imblearn_sampler(EditedNearestNeighbours, 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.

Parameters:

Name	Type	Description	Default
X	DataFrame or ndarray	Features.	required
y	Series or ndarray	Target.	required

Returns:

Type	Description
tuple	(X_resampled, y_resampled)

Examples:

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

Source code in 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)

Compare model performance across INS/OOS splits made with different seeds, to help choose a seed.

For every seed in range(seed_range[0], seed_range[1]) the rows with oot_split_col == 1 are split into INS and OOS (stratified by tgt_name), the rows with oot_split_col == 2 are the OOT sample, and a PerformanceEvaluator summarizes the model on the three samples.

Parameters:

Name	Type	Description	Default
master_df	DataFrame	Full modeling population with the target, the split column and the features (or the score column).	required
oot_split_col	str	Column that marks the modeling sample (value 1) and the OOT sample (value 2).	required
model	object or str	A fitted model whose feature names can be read, or the name of a score column that already exists in master_df.	required
tgt_name	str	Name of the binary target column.	required
seed_range	tuple of int	(start, stop) of the seeds to try; stop is excluded.	(3000, 3050)
ins_prop	float	Share of the modeling sample that goes to INS; the rest is OOS.	0.7

Returns:

Type	Description
DataFrame	The performance summaries of all seeds stacked on top of each other, with a seed column. The function does not pick a seed; choose one from this table.

Source code in 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):
    """Compare model performance across INS/OOS splits made with different seeds, to help choose a seed.

    For every seed in ``range(seed_range[0], seed_range[1])`` the rows with ``oot_split_col == 1`` are split into INS
    and OOS (stratified by ``tgt_name``), the rows with ``oot_split_col == 2`` are the OOT sample, and a
    ``PerformanceEvaluator`` summarizes the model on the three samples.

    Parameters
    ----------
    master_df : pandas.DataFrame
        Full modeling population with the target, the split column and the features (or the score column).
    oot_split_col : str
        Column that marks the modeling sample (value 1) and the OOT sample (value 2).
    model : object or str
        A fitted model whose feature names can be read, or the name of a score column that already exists in
        ``master_df``.
    tgt_name : str
        Name of the binary target column.
    seed_range : tuple of int, default (3000, 3050)
        ``(start, stop)`` of the seeds to try; ``stop`` is excluded.
    ins_prop : float, default 0.7
        Share of the modeling sample that goes to INS; the rest is OOS.

    Returns
    -------
    pandas.DataFrame
        The performance summaries of all seeds stacked on top of each other, with a ``seed`` column. The function does
        not pick a seed; choose one from this table.
    """

    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])].copy()

        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 Inference: 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.

Classes:

Name	Description
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.

Examples:

>>> from Modeling_Tool import RejectInferenceFactory
>>> 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.

Parameters:

Name	Type	Description	Default
target_col	str	Name of the target column.	'target'
score_col	str	Name of the score/probability column.	'score'
score_direction	str	Meaning of the score, 'high_good' or 'high_bad'. With 'high_good' a higher score means a lower risk, so the bad probability is 1 - score; with 'high_bad' the score is itself the bad probability. Any other value raises ValueError at construction.	'high_good'
random_state	int or None	Seed of the numpy random generator used by the methods that draw random labels (SimpleAugmentInferrer and ParcelingInferrer). None gives non-reproducible draws; with an integer, every call to infer repeats the same draws.	None

Attributes:

Name	Type	Description
target_col	str	Stored constructor argument.
score_col	str	Stored constructor argument.
score_direction	str	Stored constructor argument.
random_state	int or None	Stored constructor argument.

Methods:

Name	Description
infer	Apply reject inference.

Source code in 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.
    score_direction : str, default 'high_good'
        Meaning of the score, ``'high_good'`` or ``'high_bad'``. With ``'high_good'``
        a higher score means a lower risk, so the bad probability is ``1 - score``;
        with ``'high_bad'`` the score is itself the bad probability. Any other value
        raises ``ValueError`` at construction.
    random_state : int or None, default None
        Seed of the numpy random generator used by the methods that draw random
        labels (``SimpleAugmentInferrer`` and ``ParcelingInferrer``). ``None`` gives
        non-reproducible draws; with an integer, every call to `infer` repeats the
        same draws.

    Attributes
    ----------
    target_col : str
        Stored constructor argument.
    score_col : str
        Stored constructor argument.
    score_direction : str
        Stored constructor argument.
    random_state : int or None
        Stored constructor argument.

    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.
        score_direction : str, default 'high_good'
            ``'high_good'`` (a higher score means a lower risk, bad probability
            ``1 - score``) or ``'high_bad'`` (the score is the bad probability).
        random_state : int or None, default None
            Seed of the random generator used to draw random labels.

        Raises
        ------
        ValueError
            If ``score_direction`` is not ``'high_good'`` or ``'high_bad'``.
        """
        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. If None, the ``score_col`` given to the constructor
            is used.

        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.

Parameters:

Name	Type	Description	Default
df_approved	DataFrame	DataFrame with approved applications (has target).	required
df_rejected	DataFrame	DataFrame with rejected applications (no target).	required
score_col	str	Score column name. If None, the score_col given to the constructor is used.	None

Returns:

Type	Description
DataFrame	Combined DataFrame with inferred targets for rejected applications.

Source code in 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. If None, the ``score_col`` given to the constructor
        is used.

    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.

Every rejected application receives an independent 0/1 label drawn with probability equal to that bad rate; scores are not used.

Parameters:

Name	Type	Description	Default
target_col	str	Name of the target column. Its mean in df_approved is the bad rate (unless bad_rate is given) and the inferred labels of the rejected applications are written to it.	'target'
score_col	str	Default score column name of infer; not used by this method.	'score'
bad_rate	float or None	Override bad rate to use. If None, the mean of target_col in df_approved is used.	None
score_direction	str	'high_good' or 'high_bad'. Validated at construction (any other value raises ValueError) but not used by this method.	'high_good'
random_state	int or None	Seed of the random generator that draws the labels. None gives non-reproducible labels.	None

Notes

Every constructor parameter is also stored as a public attribute of the same name.

Examples:

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

Source code in 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.

    Every rejected application receives an independent 0/1 label drawn with
    probability equal to that bad rate; scores are not used.

    Parameters
    ----------
    target_col : str, default 'target'
        Name of the target column. Its mean in ``df_approved`` is the bad rate
        (unless ``bad_rate`` is given) and the inferred labels of the rejected
        applications are written to it.
    score_col : str, default 'score'
        Default score column name of `infer`; not used by this method.
    bad_rate : float or None, default None
        Override bad rate to use. If None, the mean of ``target_col`` in
        ``df_approved`` is used.
    score_direction : str, default 'high_good'
        ``'high_good'`` or ``'high_bad'``. Validated at construction (any other value
        raises ``ValueError``) but not used by this method.
    random_state : int or None, default None
        Seed of the random generator that draws the labels. ``None`` gives
        non-reproducible labels.

    Notes
    -----
    Every constructor parameter is also stored as a public attribute of the same name.

    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. Must contain the target column unless
            ``bad_rate`` was given.
        df_rejected : pandas.DataFrame
            Rejected applications.
        score_col : str, optional
            Score column. Not used by this method; accepted for interface
            compatibility.

        Returns
        -------
        pandas.DataFrame
            Combined data with inferred targets: the approved rows followed by the
            rejected rows (index reset), with the target column of the rejected rows
            overwritten by random 0/1 labels. The inputs are not modified.
        """
        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.

Parameters:

Name	Type	Description	Default
df_approved	DataFrame	Approved applications. Must contain the target column unless bad_rate was given.	required
df_rejected	DataFrame	Rejected applications.	required
score_col	str	Score column. Not used by this method; accepted for interface compatibility.	None

Returns:

Type	Description
DataFrame	Combined data with inferred targets: the approved rows followed by the rejected rows (index reset), with the target column of the rejected rows overwritten by random 0/1 labels. The inputs are not modified.

Source code in 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. Must contain the target column unless
        ``bad_rate`` was given.
    df_rejected : pandas.DataFrame
        Rejected applications.
    score_col : str, optional
        Score column. Not used by this method; accepted for interface
        compatibility.

    Returns
    -------
    pandas.DataFrame
        Combined data with inferred targets: the approved rows followed by the
        rejected rows (index reset), with the target column of the rejected rows
        overwritten by random 0/1 labels. The inputs are not modified.
    """
    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).

This is the rule for score_direction='high_good' (the default): a score equal to the cutoff is labelled bad. With score_direction='high_bad' the rule is reversed: scores at or above the cutoff are bad (target=1) and scores below it are good (target=0). Only the scores of the rejected applications are used.

Parameters:

Name	Type	Description	Default
target_col	str	Name of the target column that receives the inferred labels of the rejected applications.	'target'
score_col	str	Name of the score column of the rejected applications, used when infer is called without score_col.	'score'
cutoff	float	Score cutoff threshold, compared with the raw score (no rescaling).	0.5
score_direction	str	'high_good' (scores less than or equal to cutoff are bad) or 'high_bad' (scores greater than or equal to cutoff are bad). Any other value raises ValueError at construction.	'high_good'
random_state	int or None	Accepted for consistency with the other methods; this method draws no random numbers, so it has no effect.	None

Notes

Every constructor parameter is also stored as a public attribute of the same name.

Examples:

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

Source code in 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).

    This is the rule for ``score_direction='high_good'`` (the default): a score
    equal to the cutoff is labelled bad. With ``score_direction='high_bad'`` the
    rule is reversed: scores at or above the cutoff are bad (target=1) and scores
    below it are good (target=0). Only the scores of the rejected applications
    are used.

    Parameters
    ----------
    target_col : str, default 'target'
        Name of the target column that receives the inferred labels of the
        rejected applications.
    score_col : str, default 'score'
        Name of the score column of the rejected applications, used when `infer`
        is called without ``score_col``.
    cutoff : float, default 0.5
        Score cutoff threshold, compared with the raw score (no rescaling).
    score_direction : str, default 'high_good'
        ``'high_good'`` (scores less than or equal to ``cutoff`` are bad) or
        ``'high_bad'`` (scores greater than or equal to ``cutoff`` are bad). Any
        other value raises ``ValueError`` at construction.
    random_state : int or None, default None
        Accepted for consistency with the other methods; this method draws no
        random numbers, so it has no effect.

    Notes
    -----
    Every constructor parameter is also stored as a public attribute of the same name.

    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 (concatenated unchanged; their scores and
            target are not read).
        df_rejected : pandas.DataFrame
            Rejected applications.
        score_col : str, optional
            Score column of ``df_rejected``. If None, the ``score_col`` given to the
            constructor is used.

        Returns
        -------
        pandas.DataFrame
            Combined data with inferred targets: the approved rows followed by the
            rejected rows (index reset). The target of the rejected rows is a float
            column holding 1.0 (bad), 0.0 (good) or NaN. Rejects whose score was NaN
            (see above) will have ``target = NaN`` rather than ``0``.

        Raises
        ------
        ValueError
            If ``df_rejected`` is not empty and every rejected row has a NaN
            (or non-numeric) score.
        """
        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.

Parameters:

Name	Type	Description	Default
df_approved	DataFrame	Approved applications (concatenated unchanged; their scores and target are not read).	required
df_rejected	DataFrame	Rejected applications.	required
score_col	str	Score column of df_rejected. If None, the score_col given to the constructor is used.	None

Returns:

Type	Description
DataFrame	Combined data with inferred targets: the approved rows followed by the rejected rows (index reset). The target of the rejected rows is a float column holding 1.0 (bad), 0.0 (good) or NaN. Rejects whose score was NaN (see above) will have target = NaN rather than 0.

Raises:

Type	Description
ValueError	If df_rejected is not empty and every rejected row has a NaN (or non-numeric) score.

Source code in 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 (concatenated unchanged; their scores and
        target are not read).
    df_rejected : pandas.DataFrame
        Rejected applications.
    score_col : str, optional
        Score column of ``df_rejected``. If None, the ``score_col`` given to the
        constructor is used.

    Returns
    -------
    pandas.DataFrame
        Combined data with inferred targets: the approved rows followed by the
        rejected rows (index reset). The target of the rejected rows is a float
        column holding 1.0 (bad), 0.0 (good) or NaN. Rejects whose score was NaN
        (see above) will have ``target = NaN`` rather than ``0``.

    Raises
    ------
    ValueError
        If ``df_rejected`` is not empty and every rejected row has a NaN
        (or non-numeric) score.
    """
    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.

Concretely, every rejected application is duplicated into a bad copy (target 1, _weight = p_bad * weight_factor) and a good copy (target 0, _weight = (1 - p_bad) * weight_factor), where p_bad is the bad probability derived from the score (see score_direction). The approved applications are kept with _weight = 1.0. The output therefore has an extra _weight column, to be used as sample weights when training.

Parameters:

Name	Type	Description	Default
target_col	str	Name of the target column that receives the inferred labels (1 and 0) of the rejected copies.	'target'
score_col	str	Name of the score column of the rejected applications, used when infer is called without score_col.	'score'
weight_factor	float	Factor to adjust weights: it multiplies the _weight of every rejected copy (the weights of the approved rows stay 1.0).	1.0
score_direction	str	'high_good' (the bad probability is 1 - score) or 'high_bad' (the bad probability is the score). The score is read as a probability and the bad probability is clipped to [0, 1]. Any other value raises ValueError at construction.	'high_good'
random_state	int or None	Accepted for consistency with the other methods; this method draws no random numbers, so it has no effect.	None
nan_score_policy	str	How infer treats rejected rows whose bad probability is NaN (missing or non-numeric score): 'drop' removes them (with a RuntimeWarning), 'raise' raises ValueError, 'fill_0.5' sets their bad probability to 0.5 (the legacy behaviour). The value is validated when infer is called, not at construction.	'drop'

Notes

Every constructor parameter is also stored as a public attribute of the same name.

Examples:

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

Source code in 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.

    Concretely, every rejected application is duplicated into a bad copy
    (target 1, ``_weight = p_bad * weight_factor``) and a good copy (target 0,
    ``_weight = (1 - p_bad) * weight_factor``), where ``p_bad`` is the bad
    probability derived from the score (see ``score_direction``). The approved
    applications are kept with ``_weight = 1.0``. The output therefore has an
    extra ``_weight`` column, to be used as sample weights when training.

    Parameters
    ----------
    target_col : str, default 'target'
        Name of the target column that receives the inferred labels (1 and 0) of
        the rejected copies.
    score_col : str, default 'score'
        Name of the score column of the rejected applications, used when `infer`
        is called without ``score_col``.
    weight_factor : float, default 1.0
        Factor to adjust weights: it multiplies the ``_weight`` of every rejected
        copy (the weights of the approved rows stay 1.0).
    score_direction : str, default 'high_good'
        ``'high_good'`` (the bad probability is ``1 - score``) or ``'high_bad'`` (the
        bad probability is the score). The score is read as a probability and the
        bad probability is clipped to [0, 1]. Any other value raises ``ValueError``
        at construction.
    random_state : int or None, default None
        Accepted for consistency with the other methods; this method draws no
        random numbers, so it has no effect.
    nan_score_policy : str, default 'drop'
        How `infer` treats rejected rows whose bad probability is NaN (missing or
        non-numeric score): ``'drop'`` removes them (with a ``RuntimeWarning``),
        ``'raise'`` raises ``ValueError``, ``'fill_0.5'`` sets their bad probability
        to 0.5 (the legacy behaviour). The value is validated when `infer` is called,
        not at construction.

    Notes
    -----
    Every constructor parameter is also stored as a public attribute of the same name.

    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 (kept with ``_weight = 1.0``; an existing
            ``_weight`` column is overwritten in the returned copy).
        df_rejected : pandas.DataFrame
            Rejected applications.
        score_col : str, optional
            Score column of ``df_rejected``. If None, the ``score_col`` given to the
            constructor is used.

        Returns
        -------
        pandas.DataFrame
            Combined data with inferred targets and a ``_weight`` column (index
            reset): first the approved rows, then the bad copies of the rejected rows,
            then their good copies. Rejected rows dropped by ``nan_score_policy='drop'``
            appear in neither copy.

        Raises
        ------
        ValueError
            If ``nan_score_policy`` is not ``'drop'``, ``'raise'`` or ``'fill_0.5'``,
            or if it is ``'raise'`` and some bad probabilities are NaN.
        """
        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.

Parameters:

Name	Type	Description	Default
df_approved	DataFrame	Approved applications (kept with _weight = 1.0; an existing _weight column is overwritten in the returned copy).	required
df_rejected	DataFrame	Rejected applications.	required
score_col	str	Score column of df_rejected. If None, the score_col given to the constructor is used.	None

Returns:

Type	Description
DataFrame	Combined data with inferred targets and a _weight column (index reset): first the approved rows, then the bad copies of the rejected rows, then their good copies. Rejected rows dropped by nan_score_policy='drop' appear in neither copy.

Raises:

Type	Description
ValueError	If nan_score_policy is not 'drop', 'raise' or 'fill_0.5', or if it is 'raise' and some bad probabilities are NaN.

Source code in 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 (kept with ``_weight = 1.0``; an existing
        ``_weight`` column is overwritten in the returned copy).
    df_rejected : pandas.DataFrame
        Rejected applications.
    score_col : str, optional
        Score column of ``df_rejected``. If None, the ``score_col`` given to the
        constructor is used.

    Returns
    -------
    pandas.DataFrame
        Combined data with inferred targets and a ``_weight`` column (index
        reset): first the approved rows, then the bad copies of the rejected rows,
        then their good copies. Rejected rows dropped by ``nan_score_policy='drop'``
        appear in neither copy.

    Raises
    ------
    ValueError
        If ``nan_score_policy`` is not ``'drop'``, ``'raise'`` or ``'fill_0.5'``,
        or if it is ``'raise'`` and some bad probabilities are NaN.
    """
    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.

The parcels are quantile bins of the bad probability of the approved applications (see score_direction). Every rejected application falls into the parcel matching its own bad probability and receives a random 0/1 label drawn with that parcel's approved bad rate; rejected rows that cannot be assigned to a parcel (NaN score) use the overall approved bad rate.

Parameters:

Name	Type	Description	Default
target_col	str	Name of the target column. Its mean per parcel in df_approved gives the parcel bad rates, and the inferred labels of the rejected applications are written to it.	'target'
score_col	str	Name of the score column, which must exist in both df_approved and df_rejected; used when infer is called without score_col.	'score'
n_parcels	int	Number of score parcels (quantile bins of the approved scores). Fewer parcels are created when the quantile edges are not unique, and a single parcel when the approved scores cannot be binned.	10
score_direction	str	'high_good' (the bad probability is 1 - score) or 'high_bad' (the bad probability is the score). The score is read as a probability and the bad probability is clipped to [0, 1]. Any other value raises ValueError at construction.	'high_good'
random_state	int or None	Seed of the random generator that draws the labels. None gives non-reproducible labels.	None

Attributes:

Name	Type	Description
parcel_rates_	Series or None	Bad rate of the approved applications in each parcel, indexed by parcel number. None until infer has been called, then overwritten by every call.
parcel_edges_	ndarray or None	Edges of the parcels in bad-probability space, with the first edge set to -inf and the last to inf. None until infer has been called, then overwritten by every call.

Notes

Every constructor parameter is also stored as a public attribute of the same name.

Examples:

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

Source code in 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.

    The parcels are quantile bins of the bad probability of the approved
    applications (see ``score_direction``). Every rejected application falls into
    the parcel matching its own bad probability and receives a random 0/1 label
    drawn with that parcel's approved bad rate; rejected rows that cannot be
    assigned to a parcel (NaN score) use the overall approved bad rate.

    Parameters
    ----------
    target_col : str, default 'target'
        Name of the target column. Its mean per parcel in ``df_approved`` gives the
        parcel bad rates, and the inferred labels of the rejected applications are
        written to it.
    score_col : str, default 'score'
        Name of the score column, which must exist in both ``df_approved`` and
        ``df_rejected``; used when `infer` is called without ``score_col``.
    n_parcels : int, default 10
        Number of score parcels (quantile bins of the approved scores). Fewer parcels
        are created when the quantile edges are not unique, and a single parcel
        when the approved scores cannot be binned.
    score_direction : str, default 'high_good'
        ``'high_good'`` (the bad probability is ``1 - score``) or ``'high_bad'`` (the
        bad probability is the score). The score is read as a probability and the
        bad probability is clipped to [0, 1]. Any other value raises ``ValueError``
        at construction.
    random_state : int or None, default None
        Seed of the random generator that draws the labels. ``None`` gives
        non-reproducible labels.

    Attributes
    ----------
    parcel_rates_ : pandas.Series or None
        Bad rate of the approved applications in each parcel, indexed by parcel
        number. ``None`` until `infer` has been called, then overwritten by every call.
    parcel_edges_ : numpy.ndarray or None
        Edges of the parcels in bad-probability space, with the first edge set to
        ``-inf`` and the last to ``inf``. ``None`` until `infer` has been called, then
        overwritten by every call.

    Notes
    -----
    Every constructor parameter is also stored as a public attribute of the same name.

    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; must contain the target column and the score column.
        df_rejected : pandas.DataFrame
            Rejected applications; must contain the score column.
        score_col : str, optional
            Score column of both frames. If None, the ``score_col`` given to the
            constructor is used.

        Returns
        -------
        pandas.DataFrame
            Combined data with inferred targets: the approved rows followed by the
            rejected rows (index reset), with the target column of the rejected rows
            overwritten by random 0/1 labels. The inputs are not modified; the parcel
            statistics are stored in ``parcel_rates_`` and ``parcel_edges_``.
        """
        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.

Parameters:

Name	Type	Description	Default
df_approved	DataFrame	Approved applications; must contain the target column and the score column.	required
df_rejected	DataFrame	Rejected applications; must contain the score column.	required
score_col	str	Score column of both frames. If None, the score_col given to the constructor is used.	None

Returns:

Type	Description
DataFrame	Combined data with inferred targets: the approved rows followed by the rejected rows (index reset), with the target column of the rejected rows overwritten by random 0/1 labels. The inputs are not modified; the parcel statistics are stored in parcel_rates_ and parcel_edges_.

Source code in 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; must contain the target column and the score column.
    df_rejected : pandas.DataFrame
        Rejected applications; must contain the score column.
    score_col : str, optional
        Score column of both frames. If None, the ``score_col`` given to the
        constructor is used.

    Returns
    -------
    pandas.DataFrame
        Combined data with inferred targets: the approved rows followed by the
        rejected rows (index reset), with the target column of the rejected rows
        overwritten by random 0/1 labels. The inputs are not modified; the parcel
        statistics are stored in ``parcel_rates_`` and ``parcel_edges_``.
    """
    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.

Examples:

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

Source code in 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, case-insensitive: ``'simple'`` or ``'augment'``
            (``SimpleAugmentInferrer``), ``'hard'`` or ``'hardcutoff'``
            (``HardCutoffInferrer``), ``'fuzzy'`` (``FuzzyAugmentInferrer``),
            ``'parceling'`` or ``'parcel'`` (``ParcelingInferrer``).
        **kwargs
            Additional parameters for the method: keyword arguments passed to the
            constructor of the selected class (for example ``target_col``,
            ``score_col``, ``score_direction``, ``random_state`` and the
            method-specific parameters such as ``bad_rate``, ``cutoff``,
            ``weight_factor``, ``nan_score_policy`` or ``n_parcels``).

        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: every name accepted by `create` (all seven
            aliases), in no guaranteed order.
        """
        return list(set(cls._methods.keys()))

create classmethod

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

Create a reject inference method.

Parameters:

Name	Type	Description	Default
method	str	Method name, case-insensitive: 'simple' or 'augment' (SimpleAugmentInferrer), 'hard' or 'hardcutoff' (HardCutoffInferrer), 'fuzzy' (FuzzyAugmentInferrer), 'parceling' or 'parcel' (ParcelingInferrer).	'parceling'
**kwargs		Additional parameters for the method: keyword arguments passed to the constructor of the selected class (for example target_col, score_col, score_direction, random_state and the method-specific parameters such as bad_rate, cutoff, weight_factor, nan_score_policy or n_parcels).	{}

Returns:

Type	Description
RejectInferrer	Instantiated reject inferrer.

Raises:

Type	Description
ValueError	If method name is not recognized.

Source code in 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, case-insensitive: ``'simple'`` or ``'augment'``
        (``SimpleAugmentInferrer``), ``'hard'`` or ``'hardcutoff'``
        (``HardCutoffInferrer``), ``'fuzzy'`` (``FuzzyAugmentInferrer``),
        ``'parceling'`` or ``'parcel'`` (``ParcelingInferrer``).
    **kwargs
        Additional parameters for the method: keyword arguments passed to the
        constructor of the selected class (for example ``target_col``,
        ``score_col``, ``score_direction``, ``random_state`` and the
        method-specific parameters such as ``bad_rate``, ``cutoff``,
        ``weight_factor``, ``nan_score_policy`` or ``n_parcels``).

    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.

Returns:

Type	Description
list of str	Available method names: every name accepted by create (all seven aliases), in no guaranteed order.

Source code in 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: every name accepted by `create` (all seven
        aliases), in no guaranteed order.
    """
    return list(set(cls._methods.keys()))

Distribution Adaptation: Distribution_Adaptation

Distribution_Adaptation

DistributionAdaptation

Sample weights that adapt a training set to an out-of-time (OOT) distribution.

The class computes one importance weight per training row, so that the weighted training distribution moves towards the OOT distribution. The weights are produced either by fit (stored in sample_weights, read with get_weights) or by calling estimate_density_ratio / covariate_shift_weighting directly.

Parameters:

Name	Type	Description	Default
method	str	Weighting method used by fit: 'density_ratio': density-ratio estimation (estimate_density_ratio). 'covariate_shift': covariate-shift correction (covariate_shift_weighting). any other value, for example 'kl_divergence': hybrid weighting, the average of the two weight vectors above. No separate KL-divergence weighting is implemented, and the value is not validated.	'density_ratio'

Attributes:

Name	Type	Description
method	str	Weighting method passed to the constructor; only fit reads it.
sample_weights	ndarray or None	Weights computed by the last call to fit (one per training row); None before fit is called.
feature_importances	None	Reserved attribute that is always None; nothing in the class sets it.

Source code in Modeling_Tool/Sample/Distribution_Adaptation.py

class DistributionAdaptation:
    """
    Sample weights that adapt a training set to an out-of-time (OOT) distribution.

    The class computes one importance weight per training row, so that the weighted
    training distribution moves towards the OOT distribution. The weights are
    produced either by `fit` (stored in `sample_weights`, read with `get_weights`) or
    by calling `estimate_density_ratio` / `covariate_shift_weighting` directly.

    Parameters
    ----------
    method : str, default 'density_ratio'
        Weighting method used by `fit`:

        - ``'density_ratio'``: density-ratio estimation
          (`estimate_density_ratio`).
        - ``'covariate_shift'``: covariate-shift correction
          (`covariate_shift_weighting`).
        - any other value, for example ``'kl_divergence'``: hybrid weighting,
          the average of the two weight vectors above. No separate KL-divergence
          weighting is implemented, and the value is not validated.

    Attributes
    ----------
    method : str
        Weighting method passed to the constructor; only `fit` reads it.
    sample_weights : numpy.ndarray or None
        Weights computed by the last call to `fit` (one per training row);
        ``None`` before `fit` is called.
    feature_importances : None
        Reserved attribute that is always ``None``; nothing in the class sets it.
    """
    def __init__(self, method='density_ratio'):
        """
        Initialize the distribution adapter.

        Parameters
        ----------
        method : str, default 'density_ratio'
            Weighting method used by `fit`:

            - ``'density_ratio'``: density-ratio estimation
              (`estimate_density_ratio`).
            - ``'covariate_shift'``: covariate-shift correction
              (`covariate_shift_weighting`).
            - any other value, for example ``'kl_divergence'``: hybrid weighting,
              the average of the two weight vectors above. No separate KL-divergence
              weighting is implemented, and the value is not validated.
        """
        self.method = method
        self.sample_weights = None
        self.feature_importances = None

    def estimate_density_ratio(self, X_train, X_oot):
        """
        Compute sample weights using density-ratio estimation.
        A simplified implementation of methods such as KLIEP/KMM.

        Two Gaussian kernel density estimates (bandwidth 0.5) are fitted, one on
        ``X_train`` and one on ``X_oot``. The weight of each training row is the ratio
        ``p_oot(x) / p_train(x)`` evaluated at that row. The ratios are clipped to
        ``[0.1, 10]`` and then divided by their mean.

        Parameters
        ----------
        X_train : array-like of shape (n_train, n_features)
            Training features (numpy array or DataFrame), numeric without missing values.
        X_oot : array-like of shape (n_oot, n_features)
            OOT features with the same columns, in the same order, as ``X_train``.

        Returns
        -------
        numpy.ndarray of shape (n_train,)
            Importance weight of each training row, in the row order of ``X_train``.
            The weights average exactly 1.

        Raises
        ------
        ValueError
            If the inputs contain NaN or have a different number of columns
            (raised by scikit-learn).

        Notes
        -----
        The kernel bandwidth is fixed at 0.5 and no feature scaling is applied, so the
        features should be on comparable scales. The method does not read ``method``
        and does not store its result in ``sample_weights`` (`fit` does).
        """
        from sklearn.neighbors import KernelDensity

        # Estimate the densities with 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)

        # Compute the density ratio: p_oot(x) / p_train(x)
        log_density_train = kde_train.score_samples(X_train)
        log_density_oot = kde_oot.score_samples(X_train)

        # Guard against division by zero and numerical instability
        density_ratio = np.exp(log_density_oot - log_density_train)

        # Clip outliers
        density_ratio = np.clip(density_ratio, 0.1, 10)

        # Normalize
        density_ratio = density_ratio / density_ratio.mean()

        return density_ratio

    def covariate_shift_weighting(self, X_train, X_oot):
        """
        Estimate sample importance weights using a domain classifier.

        A logistic regression (``C=1.0``, ``max_iter=1000``, ``random_state=42``) is
        trained on the stacked rows of ``X_train`` (domain 0) and ``X_oot`` (domain 1).
        The weight of each training row is the odds ``p(oot|x) / (1 - p(oot|x) + 1e-10)``
        predicted for that row. The odds are clipped to ``[0.1, 10]`` and then divided
        by their mean.

        Parameters
        ----------
        X_train : array-like of shape (n_train, n_features)
            Training features (numpy array or DataFrame), numeric without missing values.
        X_oot : array-like of shape (n_oot, n_features)
            OOT features with the same columns, in the same order, as ``X_train``.

        Returns
        -------
        numpy.ndarray of shape (n_train,)
            Importance weight of each training row, in the row order of ``X_train``.
            The weights average exactly 1.

        Raises
        ------
        ValueError
            If the inputs contain NaN or have a different number of columns
            (raised by numpy or scikit-learn).

        Notes
        -----
        No feature scaling is applied and the odds are not corrected for the different
        sizes of ``X_train`` and ``X_oot`` before clipping. The method does not read
        ``method`` and does not store its result in ``sample_weights`` (`fit` does).
        """
        from sklearn.linear_model import LogisticRegression

        # Create domain labels: 0 for the training set, 1 for OOT
        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)])

        # Train the domain classifier
        domain_classifier = LogisticRegression(
            C=1.0, max_iter=1000, random_state=42
        )
        domain_classifier.fit(X_combined, y_domain)

        # Predict the probability that each training sample comes from OOT
        p_oot = domain_classifier.predict_proba(X_train)[:, 1]

        # Compute the weights: p(oot|x) / p(train|x)
        # Guard against numerical instability
        epsilon = 1e-10
        weights = p_oot / (1 - p_oot + epsilon)

        # Smooth the weights using a beta distribution
        weights = np.clip(weights, 0.1, 10)
        weights = weights / weights.mean()

        return weights

    def fit(self, X_train, X_oot, y_train=None):
        """
        Compute sample weights that adapt the training distribution to the OOT distribution.

        The weights are computed according to ``method`` and stored in
        ``sample_weights``: ``'density_ratio'`` uses `estimate_density_ratio`,
        ``'covariate_shift'`` uses `covariate_shift_weighting`, and any other value
        uses the average of the two weight vectors.

        Parameters
        ----------
        X_train : array-like of shape (n_train, n_features)
            Training features (numpy array or DataFrame), numeric without missing values.
        X_oot : array-like of shape (n_oot, n_features)
            OOT features with the same columns, in the same order, as ``X_train``.
        y_train : array-like or None, default None
            Not used; accepted only for interface compatibility.

        Returns
        -------
        DistributionAdaptation
            The fitted instance itself (``self``), with ``sample_weights`` set to a
            numpy array of shape (n_train,).
        """
        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:
            # By default, use the hybrid method (average of both weights)
            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 the sample weights.

        Returns
        -------
        numpy.ndarray or None
            The weights stored by the last call to `fit` (one per training row), or
            ``None`` if `fit` has not been called.
        """
        return self.sample_weights

    def visualize_distribution_comparison(self, X_train, X_oot, features=None, n_features=5):
        """
        Visualize the distribution differences between the training set and the OOT set.

        Draws one filled kernel-density plot per feature (training set labelled
        ``Train``, OOT set labelled ``OOT``) on a single row of ``n_features``
        subplots, then calls ``plt.tight_layout()`` and ``plt.show()``.

        Parameters
        ----------
        X_train : numpy.ndarray of shape (n_train, n_total_features)
            Training features. Must be a 2-D numpy array because columns are
            selected by position; a DataFrame raises an error.
        X_oot : numpy.ndarray of shape (n_oot, n_total_features)
            OOT features with the same column order as ``X_train``.
        features : sequence of int or None, default None
            Column positions (not names) to plot; only the first ``n_features``
            entries are used. If ``None``, the ``n_features`` columns of ``X_train``
            with the largest variance are plotted, in ascending order of variance.
        n_features : int, default 5
            Number of subplots in the figure and maximum number of features plotted.

        Returns
        -------
        None
            The figure is only displayed.

        Notes
        -----
        The figure always has ``n_features`` subplots, so subplots are left empty when
        fewer features are available. Each subplot is titled ``Feature <column position>``.
        """
        if features is None:
            # Select the features with the largest variance
            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

            # Plot the distributions
            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)

Compute sample weights using density-ratio estimation. A simplified implementation of methods such as KLIEP/KMM.

Two Gaussian kernel density estimates (bandwidth 0.5) are fitted, one on X_train and one on X_oot. The weight of each training row is the ratio p_oot(x) / p_train(x) evaluated at that row. The ratios are clipped to [0.1, 10] and then divided by their mean.

Parameters:

Name	Type	Description	Default
X_train	array-like of shape (n_train, n_features)	Training features (numpy array or DataFrame), numeric without missing values.	required
X_oot	array-like of shape (n_oot, n_features)	OOT features with the same columns, in the same order, as X_train.	required

Returns:

Type	Description
numpy.ndarray of shape (n_train,)	Importance weight of each training row, in the row order of X_train. The weights average exactly 1.

Raises:

Type	Description
ValueError	If the inputs contain NaN or have a different number of columns (raised by scikit-learn).

Notes

The kernel bandwidth is fixed at 0.5 and no feature scaling is applied, so the features should be on comparable scales. The method does not read method and does not store its result in sample_weights (fit does).

Source code in Modeling_Tool/Sample/Distribution_Adaptation.py

def estimate_density_ratio(self, X_train, X_oot):
    """
    Compute sample weights using density-ratio estimation.
    A simplified implementation of methods such as KLIEP/KMM.

    Two Gaussian kernel density estimates (bandwidth 0.5) are fitted, one on
    ``X_train`` and one on ``X_oot``. The weight of each training row is the ratio
    ``p_oot(x) / p_train(x)`` evaluated at that row. The ratios are clipped to
    ``[0.1, 10]`` and then divided by their mean.

    Parameters
    ----------
    X_train : array-like of shape (n_train, n_features)
        Training features (numpy array or DataFrame), numeric without missing values.
    X_oot : array-like of shape (n_oot, n_features)
        OOT features with the same columns, in the same order, as ``X_train``.

    Returns
    -------
    numpy.ndarray of shape (n_train,)
        Importance weight of each training row, in the row order of ``X_train``.
        The weights average exactly 1.

    Raises
    ------
    ValueError
        If the inputs contain NaN or have a different number of columns
        (raised by scikit-learn).

    Notes
    -----
    The kernel bandwidth is fixed at 0.5 and no feature scaling is applied, so the
    features should be on comparable scales. The method does not read ``method``
    and does not store its result in ``sample_weights`` (`fit` does).
    """
    from sklearn.neighbors import KernelDensity

    # Estimate the densities with 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)

    # Compute the density ratio: p_oot(x) / p_train(x)
    log_density_train = kde_train.score_samples(X_train)
    log_density_oot = kde_oot.score_samples(X_train)

    # Guard against division by zero and numerical instability
    density_ratio = np.exp(log_density_oot - log_density_train)

    # Clip outliers
    density_ratio = np.clip(density_ratio, 0.1, 10)

    # Normalize
    density_ratio = density_ratio / density_ratio.mean()

    return density_ratio

covariate_shift_weighting

covariate_shift_weighting(X_train, X_oot)

Estimate sample importance weights using a domain classifier.

A logistic regression (C=1.0, max_iter=1000, random_state=42) is trained on the stacked rows of X_train (domain 0) and X_oot (domain 1). The weight of each training row is the odds p(oot|x) / (1 - p(oot|x) + 1e-10) predicted for that row. The odds are clipped to [0.1, 10] and then divided by their mean.

Parameters:

Name	Type	Description	Default
X_train	array-like of shape (n_train, n_features)	Training features (numpy array or DataFrame), numeric without missing values.	required
X_oot	array-like of shape (n_oot, n_features)	OOT features with the same columns, in the same order, as X_train.	required

Returns:

Type	Description
numpy.ndarray of shape (n_train,)	Importance weight of each training row, in the row order of X_train. The weights average exactly 1.

Raises:

Type	Description
ValueError	If the inputs contain NaN or have a different number of columns (raised by numpy or scikit-learn).

Notes

No feature scaling is applied and the odds are not corrected for the different sizes of X_train and X_oot before clipping. The method does not read method and does not store its result in sample_weights (fit does).

Source code in Modeling_Tool/Sample/Distribution_Adaptation.py

def covariate_shift_weighting(self, X_train, X_oot):
    """
    Estimate sample importance weights using a domain classifier.

    A logistic regression (``C=1.0``, ``max_iter=1000``, ``random_state=42``) is
    trained on the stacked rows of ``X_train`` (domain 0) and ``X_oot`` (domain 1).
    The weight of each training row is the odds ``p(oot|x) / (1 - p(oot|x) + 1e-10)``
    predicted for that row. The odds are clipped to ``[0.1, 10]`` and then divided
    by their mean.

    Parameters
    ----------
    X_train : array-like of shape (n_train, n_features)
        Training features (numpy array or DataFrame), numeric without missing values.
    X_oot : array-like of shape (n_oot, n_features)
        OOT features with the same columns, in the same order, as ``X_train``.

    Returns
    -------
    numpy.ndarray of shape (n_train,)
        Importance weight of each training row, in the row order of ``X_train``.
        The weights average exactly 1.

    Raises
    ------
    ValueError
        If the inputs contain NaN or have a different number of columns
        (raised by numpy or scikit-learn).

    Notes
    -----
    No feature scaling is applied and the odds are not corrected for the different
    sizes of ``X_train`` and ``X_oot`` before clipping. The method does not read
    ``method`` and does not store its result in ``sample_weights`` (`fit` does).
    """
    from sklearn.linear_model import LogisticRegression

    # Create domain labels: 0 for the training set, 1 for OOT
    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)])

    # Train the domain classifier
    domain_classifier = LogisticRegression(
        C=1.0, max_iter=1000, random_state=42
    )
    domain_classifier.fit(X_combined, y_domain)

    # Predict the probability that each training sample comes from OOT
    p_oot = domain_classifier.predict_proba(X_train)[:, 1]

    # Compute the weights: p(oot|x) / p(train|x)
    # Guard against numerical instability
    epsilon = 1e-10
    weights = p_oot / (1 - p_oot + epsilon)

    # Smooth the weights using a beta distribution
    weights = np.clip(weights, 0.1, 10)
    weights = weights / weights.mean()

    return weights

fit

fit(X_train, X_oot, y_train=None)

Compute sample weights that adapt the training distribution to the OOT distribution.

The weights are computed according to method and stored in sample_weights: 'density_ratio' uses estimate_density_ratio, 'covariate_shift' uses covariate_shift_weighting, and any other value uses the average of the two weight vectors.

Parameters:

Name	Type	Description	Default
X_train	array-like of shape (n_train, n_features)	Training features (numpy array or DataFrame), numeric without missing values.	required
X_oot	array-like of shape (n_oot, n_features)	OOT features with the same columns, in the same order, as X_train.	required
y_train	array - like or None	Not used; accepted only for interface compatibility.	None

Returns:

Type	Description
DistributionAdaptation	The fitted instance itself (self), with sample_weights set to a numpy array of shape (n_train,).

Source code in Modeling_Tool/Sample/Distribution_Adaptation.py

def fit(self, X_train, X_oot, y_train=None):
    """
    Compute sample weights that adapt the training distribution to the OOT distribution.

    The weights are computed according to ``method`` and stored in
    ``sample_weights``: ``'density_ratio'`` uses `estimate_density_ratio`,
    ``'covariate_shift'`` uses `covariate_shift_weighting`, and any other value
    uses the average of the two weight vectors.

    Parameters
    ----------
    X_train : array-like of shape (n_train, n_features)
        Training features (numpy array or DataFrame), numeric without missing values.
    X_oot : array-like of shape (n_oot, n_features)
        OOT features with the same columns, in the same order, as ``X_train``.
    y_train : array-like or None, default None
        Not used; accepted only for interface compatibility.

    Returns
    -------
    DistributionAdaptation
        The fitted instance itself (``self``), with ``sample_weights`` set to a
        numpy array of shape (n_train,).
    """
    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:
        # By default, use the hybrid method (average of both weights)
        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()

Return the sample weights.

Returns:

Type	Description
ndarray or None	The weights stored by the last call to fit (one per training row), or None if fit has not been called.

Source code in Modeling_Tool/Sample/Distribution_Adaptation.py

def get_weights(self):
    """
    Return the sample weights.

    Returns
    -------
    numpy.ndarray or None
        The weights stored by the last call to `fit` (one per training row), or
        ``None`` if `fit` has not been called.
    """
    return self.sample_weights

visualize_distribution_comparison

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

Visualize the distribution differences between the training set and the OOT set.

Draws one filled kernel-density plot per feature (training set labelled Train, OOT set labelled OOT) on a single row of n_features subplots, then calls plt.tight_layout() and plt.show().

Parameters:

Name	Type	Description	Default
X_train	numpy.ndarray of shape (n_train, n_total_features)	Training features. Must be a 2-D numpy array because columns are selected by position; a DataFrame raises an error.	required
X_oot	numpy.ndarray of shape (n_oot, n_total_features)	OOT features with the same column order as X_train.	required
features	sequence of int or None	Column positions (not names) to plot; only the first n_features entries are used. If None, the n_features columns of X_train with the largest variance are plotted, in ascending order of variance.	None
n_features	int	Number of subplots in the figure and maximum number of features plotted.	5

Returns:

Type	Description
None	The figure is only displayed.

Notes

The figure always has n_features subplots, so subplots are left empty when fewer features are available. Each subplot is titled Feature <column position>.

Source code in Modeling_Tool/Sample/Distribution_Adaptation.py

def visualize_distribution_comparison(self, X_train, X_oot, features=None, n_features=5):
    """
    Visualize the distribution differences between the training set and the OOT set.

    Draws one filled kernel-density plot per feature (training set labelled
    ``Train``, OOT set labelled ``OOT``) on a single row of ``n_features``
    subplots, then calls ``plt.tight_layout()`` and ``plt.show()``.

    Parameters
    ----------
    X_train : numpy.ndarray of shape (n_train, n_total_features)
        Training features. Must be a 2-D numpy array because columns are
        selected by position; a DataFrame raises an error.
    X_oot : numpy.ndarray of shape (n_oot, n_total_features)
        OOT features with the same column order as ``X_train``.
    features : sequence of int or None, default None
        Column positions (not names) to plot; only the first ``n_features``
        entries are used. If ``None``, the ``n_features`` columns of ``X_train``
        with the largest variance are plotted, in ascending order of variance.
    n_features : int, default 5
        Number of subplots in the figure and maximum number of features plotted.

    Returns
    -------
    None
        The figure is only displayed.

    Notes
    -----
    The figure always has ``n_features`` subplots, so subplots are left empty when
    fewer features are available. Each subplot is titled ``Feature <column position>``.
    """
    if features is None:
        # Select the features with the largest variance
        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

        # Plot the distributions
        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()