import dill
import sklearn
import numpy as np
from packaging import version
from sklearn.base import BaseEstimator
from sklearn.base import is_regressor, is_classifier
from sklearn.isotonic import IsotonicRegression
from sklearn.linear_model import LogisticRegression
from sklearn.preprocessing import LabelBinarizer
from sklearn.utils import column_or_1d
class ModelBase(BaseEstimator):
"""Common Methods for Modeva models
"""
@property
def name(self):
# Required for low-code panel
if hasattr(self, '_name') and self._name is not None:
return self._name
else:
return self.__class__.__name__
@name.setter
def name(self, val):
self._name = val
@property
def version(self):
# Required for low-code panel
return 1
def _validate_fit_inputs(self, X, y, sample_weight):
"""
Validate the inputs.
"""
if is_classifier(self):
if version.parse(sklearn.__version__) < version.parse("1.6"):
X, y = self._validate_data(X, y)
else:
X, y = sklearn.utils.validation.validate_data(self, X, y)
if y.ndim == 2 and y.shape[1] == 1:
y = column_or_1d(y, warn=False)
self.label_binarizer_ = LabelBinarizer()
self.label_binarizer_.fit(y)
self.classes_ = self.label_binarizer_.classes_
y = self.label_binarizer_.transform(y) * 1.0
elif is_regressor(self):
if version.parse(sklearn.__version__) < version.parse("1.6"):
X, y = self._validate_data(X, y, y_numeric=True)
else:
X, y = sklearn.utils.validation.validate_data(self, X, y, y_numeric=True)
if y.ndim == 2 and y.shape[1] == 1:
y = column_or_1d(y, warn=True)
if sample_weight is None:
sample_weight = np.ones(X.shape[0])
else:
sample_weight = np.asarray(sample_weight).ravel()
sample_weight = X.shape[0] * sample_weight / sample_weight.sum()
return X, y.ravel(), sample_weight
def save(self, file_name: str):
"""
Save the model into file system.
Parameters
----------
file_name: str
The path and name of the file.
"""
dill.dump(self, open(file_name, "wb"))
def load(self, file_name: str):
"""
Load the model into memory from file system.
Parameters
----------
file_name: str
The path and name of the file.
Returns
-------
estimator object
"""
return dill.load(open(file_name, "rb"))
[docs]
class ModelBaseRegressor(ModelBase):
"""
Base Class for Modeva Regressors.
"""
def reset_calibrate_interval(self):
if hasattr(self, "calibration_qval_"):
self.calibration_qval_ = None
def reset_calibrate_proba(self):
if hasattr(self, "calibrator_"):
self.calibrator_ = None
[docs]
def calibrate_interval(self, X, y, alpha=0.1, max_depth: int = 5):
"""
Fit a conformal prediction model to the given data.
This method computes the model's prediction interval calibrated to the given data.
If the model is a regressor, splits the data with 50% for fitting lower (alpha / 5) and upper (1 - alpha / 2) gradient boosting trees-based quantile regression to the model's residual; and 50% for calibration.
If the model is a binary classifiers, it computes the calibration quantile based on predicted probabilities for the positive class.
Parameters
----------
X : X : np.ndarray of shape (n_samples, n_features)
Feature matrix for prediction.
y : array-like of shape (n_samples, )
Target values.
alpha : float, default=0.1
Expected miscoverage for the conformal prediction.
max_depth : int, default=5
Maximum depth of the gradient boosting trees for regression tasks.
Only used when task_type is REGRESSION.
Raises
------
ValueError: If the model is neither a regressor nor a classifier.
"""
from ..testsuite.diagnose.reliability import ReliabilityTest
prediction = self.predict(X)
self.model_qr_, self.calibration_qval_, = ReliabilityTest._run_regressor(X,
y,
prediction,
alpha=alpha,
max_depth=max_depth,
random_state=0)
[docs]
def predict_interval(self, X):
"""
Predict the prediction interval for the given data based on the conformal prediction model.
It splits the data with 50% for fitting lower (alpha / 5) and upper (1 - alpha / 2) gradient boosting trees-based quantile regression to the model's residual; and 50% for calibration.
Parameters
----------
X : np.ndarray of shape (n_samples, n_features)
Feature matrix for prediction.
Returns
-------
np.ndarray: The lower and upper bounds of the prediction intervals for each sample
in the format [n_samples, 2] for regressors or a flattened array for classifiers.
Raises
------
ValueError: If `fit_conformal` has not been called to fit the conformal prediction model
before calling this method.
"""
from ..testsuite.diagnose.reliability import ReliabilityTest
if hasattr(self, "calibration_qval_") and self.calibration_qval_ is not None:
prediction = self.predict(X)
quantile = ReliabilityTest.predict_interval_regressor(self.model_qr_, X)
pred_low = prediction.ravel() + quantile[0].ravel() - self.calibration_qval_
pred_high = prediction.ravel() + quantile[1].ravel() + self.calibration_qval_
return np.vstack([pred_low, pred_high]).T
else:
raise ValueError("Run fit_conformal first to fit the conformal prediction model.")
[docs]
def predict(self, X):
"""
Model predictions, calling the child class's '_predict' method.
Parameters
----------
X : np.ndarray of shape (n_samples, n_features)
Feature matrix for prediction.
Returns
-------
np.ndarray: The (calibrated) final prediction
"""
return self._predict(X)
[docs]
class ModelBaseClassifier(ModelBase):
"""
Base Class for Modeva Classifiers.
"""
def reset_calibrate_interval(self):
if hasattr(self, "quantile_"):
self.quantile_ = None
def reset_calibrate_proba(self):
if hasattr(self, "calibrator_"):
self.calibrator_ = None
[docs]
def calibrate_interval(self, X, y, alpha=0.1):
"""
Fit a conformal prediction model to the given data.
This method computes the model's prediction interval calibrated to the given data.
It computes the calibration quantile based on predicted probabilities for the positive class.
Parameters
----------
X : X : np.ndarray of shape (n_samples, n_features)
Feature matrix for prediction.
y : array-like of shape (n_samples, )
Target values.
alpha : float, default=0.1
Expected miscoverage for the conformal prediction.
Raises
------
ValueError: If the model is neither a regressor nor a classifier.
"""
from ..testsuite.diagnose.reliability import ReliabilityTest
prediction = self.predict_proba(X)[:, 1]
self.quantile_ = ReliabilityTest._run_classifier(y=y,
prediction=prediction,
alpha=alpha)
[docs]
def predict_interval(self, X):
"""
Predict the prediction set for the given data based on the conformal prediction model.
This method computes the model prediction interval (regression) or prediction sets (classification) using conformal prediction.
Parameters
----------
X : np.ndarray of shape (n_samples, n_features)
Feature matrix for prediction.
Returns
-------
np.ndarray: The lower and upper bounds of the prediction intervals for each sample
in the format [n_samples, 2] for regressors or a flattened array for classifiers.
Raises
------
ValueError: If `fit_conformal` has not been called to fit the conformal prediction model
before calling this method.
"""
from ..testsuite.diagnose.reliability import ReliabilityTest
if hasattr(self, "quantile_") and self.quantile_ is not None:
prediction = self.predict_proba(X)[:, 1]
eval_prediction_sets = ReliabilityTest.predict_interval_classifier(self.quantile_, prediction)
return np.hstack(eval_prediction_sets).ravel()
else:
raise ValueError("Run fit_conformal first to fit the conformal prediction model.")
[docs]
def calibrate_proba(self, X, y, sample_weight=None, method='sigmoid'):
"""
Fit the calibration method on the model's predictions.
Parameters
----------
X : np.ndarray of shape (n_samples, n_features)
Feature matrix for prediction.
y : np.ndarray of shape (n_samples, )
Ground truth labels.
sample_weight : array-like, shape (n_samples,), default=None
Sample weights.
method : {'sigmoid', 'isotonic'}, default='sigmoid'
The calibration method.
- 'sigmoid': Platt’s method, i.e., fit a logistic regression on predicted probabilities and y
- 'isotonic': Fit an isotonic regression on predicted probabilities and y.
Returns
--------
self: Calibrated estimator
"""
# Obtain the predicted probabilities from the model
probs = self.predict_proba(X)[:, 1] # Probabilities for class 1
# Apply the selected calibration method
if method == 'sigmoid':
# Platt Scaling (logistic regression on predicted probabilities)
self.calibrator_ = LogisticRegression()
self.calibrator_.fit(probs.reshape(-1, 1), y.reshape(-1, 1), sample_weight=sample_weight)
elif method == 'isotonic':
# Isotonic Regression
self.calibrator_ = IsotonicRegression(out_of_bounds='clip')
self.calibrator_.fit(probs.ravel(), y.ravel(), sample_weight=sample_weight)
else:
raise ValueError("Invalid method. Choose either 'sigmoid' or 'isotonic'.")
return self
[docs]
def predict_proba(self, X, calibration: bool = True):
"""
Predict (calibrated) probabilities for X.
Parameters
----------
X : np.ndarray of shape (n_samples, n_features)
Feature matrix for prediction.
calibration : bool, default=True
If True, will return calibrated probability if calibration is done.
Otherwise, will return raw probability.
Returns
-------
np.ndarray: The (calibrated) predicted probabilities
"""
# Get raw predicted probabilities from the base model
raw_probs = self._predict_proba(X)
# Calibrate using the selected method
if calibration and hasattr(self, "calibrator_") and self.calibrator_ is not None:
if isinstance(self.calibrator_, LogisticRegression):
# Apply Platt scaling (logistic regression transformation)
calibrated_probs = self.calibrator_.predict_proba(raw_probs[:, 1].reshape(-1, 1))[:, 1]
elif isinstance(self.calibrator_, IsotonicRegression):
# Apply Isotonic regression
calibrated_probs = self.calibrator_.transform(raw_probs[:, 1])
else:
raise ValueError("Calibration method not set properly.")
# Return calibrated probabilities (second column for class 1)
return np.column_stack((1 - calibrated_probs, calibrated_probs))
else:
return raw_probs
[docs]
def predict(self, X, calibration: bool = True):
"""
Model predictions, calling the child class's '_predict' method.
Parameters
----------
X : np.ndarray of shape (n_samples, n_features)
Feature matrix for prediction.
calibration : bool, default=True
If True, will use calibrated probability if calibration is done.
Otherwise, will use raw probability.
Returns
-------
np.ndarray: The (calibrated) final prediction
"""
return np.array(self.classes_)[np.argmax(self.predict_proba(X, calibration=calibration), axis=1)]
[docs]
def decision_function(self, X, calibration: bool = True):
"""
Computes the decision function for the given input data.
Parameters
----------
X : np.ndarray of shape (n_samples, n_features)
Feature matrix for prediction.
calibration : bool, default=True
If True, will use calibrated probability if calibration is done.
Otherwise, will use raw probability.
Returns
-------
logit_prediction : array, shape (n_samples,) or (n_samples, n_classes)
Array of (calibrated) logit predictions.
"""
if len(self.classes_) == 2:
proba = self.predict_proba(X, calibration=calibration)[:, 1]
clip_proba = np.clip(proba, 1e-7, 1 - 1e-7)
logit_prediction = np.log(clip_proba / (1 - clip_proba))
elif len(self.classes_) > 2:
proba = self.predict_proba(X, calibration=calibration)
clip_proba = np.clip(proba, 1e-7, 1 - 1e-7)
log_proba = np.log(clip_proba)
logit_prediction = log_proba - np.mean(log_proba, axis=1, keepdims=True)
return logit_prediction