from typing import Union, Dict
import numpy as np
import pandas as pd
from ...utils.constants import NUMERICAL, CATEGORICAL, DATE
def fit_xgb1(X, y, sample_weight, feature_names, feature_types, **kwargs):
"""
Fit an XGBoost model.
Parameters
----------
X : np.ndarray
Feature matrix.
y : np.ndarray
Target variable.
sample_weight : np.ndarray
Sample weights.
feature_names : list of str
Names of the features.
feature_types : list of str
Types of the features (NUMERICAL or CATEGORICAL).
**kwargs : keyword arguments
Additional parameters for the XGBoost model.
Returns
-------
MoXGBRegressor
Fitted XGBoost model.
"""
from ...models import MoXGBRegressor
xgb_model = MoXGBRegressor(max_depth=1, **kwargs)
xgb_model.fit(X, y, sample_weight)
xgb_model.extract_model_info(X=X, feature_names=feature_names, feature_types=feature_types)
return xgb_model
def get_slicing_bins(X,
residuals=None,
sample_weight=None,
feature_names=None,
feature_types=None,
method: str = "uniform",
bins: Union[int, Dict] = 10,
n_estimators: int = 1000):
"""
Get slicing bins for features based on specified method.
Parameters
----------
X : np.ndarray
Feature matrix.
residuals : np.ndarray, default=None
Residuals for the model.
sample_weight : np.ndarray, default=None
Sample weights.
feature_names : Tuple[str], default=None
Names of the features.
feature_types : Tuple[str], default=None
Types of the features.
method : str, default="uniform"
Method to use for binning ('uniform', 'quantile', 'auto-xgb1', 'precompute').
bins : Union[int, Dict], default=10
Number of bins or specific bin edges.
n_estimators : int, default=1000
Number of estimators for the XGBoost model.
Returns
-------
Tuple[Dict, Dict]
A tuple containing the bins dictionary and density dictionary.
"""
if feature_names is None:
feature_names = ["X" + str(fidx) for fidx in range(X.shape[1])]
if feature_types is None:
feature_types = [NUMERICAL for _ in range(X.shape[1])]
bins_dict = dict()
density_dict = dict()
features_idx = np.arange(len(feature_names))
if method == "uniform":
for fn, fidx in zip(feature_names, features_idx):
if feature_types[fidx] == CATEGORICAL:
bins_dict[fn], density_dict[fn] = np.unique(X[:, fidx], return_counts=True)
elif feature_types[fidx] in (NUMERICAL, DATE):
if feature_types[fidx] == DATE:
dates = pd.to_datetime(X[:, fidx])
bins_dict[fn] = np.linspace(dates.min().value,
dates.max().value,
bins + 1).astype(dates.dtype)
idx = np.digitize(dates.view(np.int64),
bins=bins_dict[fn][1:-1].view(np.int64), right=False)
elif feature_types[fidx] == NUMERICAL:
bins_dict[fn] = np.linspace(X[:, fidx].min(), X[:, fidx].max(), bins + 1).astype(float)
idx = np.digitize(X[:, fidx], bins=bins_dict[fn][1:-1], right=False)
density_dict[fn] = np.bincount(idx)
elif method == "quantile":
for fn, fidx in zip(feature_names, features_idx):
if feature_types[fidx] == CATEGORICAL:
bins_dict[fn], density_dict[fn] = np.unique(X[:, fidx], return_counts=True)
elif feature_types[fidx] in (NUMERICAL, DATE):
if feature_types[fidx] == DATE:
dates = pd.to_datetime(X[:, fidx])
quantiles = np.linspace(0.0, 1.0, bins + 1)
bins_dict[fn] = np.quantile(dates, quantiles)
idx = np.digitize(dates.view(np.int64),
bins=bins_dict[fn][1:-1].view(np.int64), right=False)
elif feature_types[fidx] == NUMERICAL:
bins_dict[fn] = np.unique(np.quantile(X[:, fidx],
np.linspace(0.0, 1.0, bins + 1)).astype(float))
idx = np.digitize(X[:, fidx], bins=bins_dict[fn][1:-1], right=False)
density_dict[fn] = np.bincount(idx)
elif method == "auto-xgb1":
abs_residual = np.abs(residuals.ravel())
xgb_model = fit_xgb1(X=X,
y=abs_residual,
sample_weight=sample_weight,
feature_names=feature_names,
feature_types=feature_types,
n_estimators=n_estimators,
max_bin=bins,
tree_method="hist")
if "main_effect" not in xgb_model.modeva_effects_:
raise ValueError("Got a runtime error for 'auto-xgb1', please try other two methods.")
for fn, fidx in zip(feature_names, features_idx):
if fn in xgb_model.modeva_effects_["main_effect"]:
item = xgb_model.modeva_effects_["main_effect"][fn]
if feature_types[fidx] == CATEGORICAL:
bins_dict[fn], density_dict[fn] = np.unique(X[:, fidx], return_counts=True)
elif feature_types[fidx] == NUMERICAL:
bins_dict[fn] = np.unique(np.hstack([np.array(item["splits"][0][1:-1]),
X[:, fidx].min(), X[:, fidx].max(),
])).astype(float)
idx = np.digitize(X[:, fidx], bins=bins_dict[fn][1:-1], right=False)
density_dict[fn] = np.bincount(idx)
else:
bins_dict[fn], density_dict[fn] = np.unique(np.hstack([X[:, fidx].min(),
X[:, fidx].max()]),
return_counts=True)
elif method == "precompute":
for fn, fidx in zip(feature_names, features_idx):
if feature_types[fidx] == CATEGORICAL:
if fn in bins:
bins_dict[fn] = np.unique(np.array(bins[fn]))
unique, counts = np.unique(X[:, fidx], return_counts=True)
data_counts = dict(zip(unique, counts))
density_dict[fn] = {category: data_counts.get(category, 0) for category in bins[fn]}
else:
bins_dict[fn], density_dict[fn] = np.unique(X[:, fidx], return_counts=True)
elif feature_types[fidx] in (NUMERICAL, DATE):
if feature_types[fidx] == DATE:
dates = pd.to_datetime(X[:, fidx])
if fn in bins:
bins_dict[fn] = pd.to_datetime(bins[fn]).values
else:
bins_dict[fn] = np.hstack([dates.min(), dates.max()])
idx = np.digitize(dates.view(np.int64),
bins=bins_dict[fn][1:-1].view(np.int64), right=False)
elif feature_types[fidx] == NUMERICAL:
if fn in bins:
bins_dict[fn] = np.unique(np.array(bins[fn])).astype(float)
else:
bins_dict[fn] = np.hstack([X[:, fidx].min(), X[:, fidx].max()]).astype(float)
idx = np.digitize(X[:, fidx], bins=bins_dict[fn][1:-1], right=False)
density_dict[fn] = np.bincount(idx)
else:
raise ValueError("'method' should be one of 'uniform', 'quantile', 'auto-xgb1' or 'precompute'.")
return bins_dict, density_dict
[docs]
def get_data_info(res_value):
"""
Extract data information from the result values.
It is designed for extracting the "good" / "bad" samples of slicing-based tests, and the results can be further used for testing data distribution drift.
Parameters
----------
res_value : list of dict
List containing result values with feature and sample information.
Returns
-------
dict
A dictionary containing data information for each feature. The structure is as follows:
.. code-block:: python
{
'feature_name': {
'dataset1': str, # Name of the dataset
'dataset2': str, # Name of the dataset
'sample_idx1': list, # List of sample IDs for "good" samples
'sample_idx2': list, # List of sample IDs for "bad" samples
'name1': str, # Label for "good" samples
'name2': str # Label for "bad" samples
}
}
Examples
--------
.. code-block:: python
results = ts.diagnose_slicing_robustness(features="PAY_1",
perturb_features=("PAY_1", "EDUCATION",),
noise_levels=0.1,
metric="AUC",
method="auto-xgb1",
threshold=0.7)
data_info = get_data_info(res_value=results.value)["PAY_1"]
"""
segment_data = []
for item in res_value:
if "Feature" in item:
features = item["Feature"]
else:
features = (item["Feature1"], item["Feature2"])
segment_data.append({"ID": item["Sample_ID"],
"Dataset": item["Sample_Dataset"],
"Features": features,
"Weak": item["Weak"]})
data_info = {}
segment_df = pd.DataFrame(segment_data)
## there should be only one unique dataset in the slicing
dataset = np.unique(segment_df["Dataset"])[0]
for fn in np.unique(segment_df["Features"]):
subdf = segment_df[segment_df["Features"] == fn][segment_df["Dataset"] == dataset]
subdf_weak = subdf[subdf["Weak"]]
subdf_non_weak = subdf[~subdf["Weak"]]
if subdf_weak.shape[0] > 0:
eval_weak_idx = np.hstack(subdf_weak["ID"])
else:
eval_weak_idx = np.empty((0, 0))
if subdf_non_weak.shape[0] > 0:
eval_non_weak_idx = np.hstack(subdf_non_weak["ID"])
else:
eval_non_weak_idx = np.empty((0, 0))
data_info[fn] = {"dataset1": dataset,
"dataset2": dataset,
"sample_idx1": eval_weak_idx.tolist(),
"sample_idx2": eval_non_weak_idx.tolist(),
"name1": "Weak Segments",
"name2": "Remaining"}
return data_info