Prediction๏ƒ

This module provides standard wrappings for ML models.

class prediction.BasePredictor(model, is_trained=False, **compile_kwargs)๏ƒ

Bases: object

Wrapper of a point prediction model \(\hat{f}\). Enables to standardize the interface of predictors and to expose generic fit(), predict() and copy() methods.

Parameters:

  • model (Any) โ€“ prediction model \(\hat{f}\)

  • is_trained (bool) โ€“ boolean flag that informs if the model is pre-trained. If True, the call to fit() will be skipped

  • compile_kwargs โ€“ keyword arguments to be used if needed during the call model.compile() on the underlying model

Sklearn regression examples:

from deel.puncc.api.prediction import BasePredictor
from sklearn.ensemble import RandomForestRegressor
import numpy as np

# Generate data
X_train = np.random.uniform(0, 10, 1000)
X_new = np.random.uniform(0, 10, 100)
y_train = np.sin(X)

# Instantiate two random forest models composed of 100 trees.
rf_model1 = RandomForestRegressor(n_estimators=100)
rf_model2 = RandomForestRegressor(n_estimators=100)

# Consider that `rf_model2` is previously trained
rf_model2.fit(X_train, y_train)

# We will instantiate two wrappers:
# - `predictor1` will wrap `rf_model1`.
# - `predictor2` will wrap `rf_model2`. Also, we don't want to retrain our model,
#   so we will specify it to the constructor by setting the argument
#   `is_trained` to True. If `fit` is called, it will be skipped.
# The argument `is_trained` defaults to False.
predictor1 = BasePredictor(rf_model1, is_trained=False)
predictor2 = BasePredictor(rf_model2, is_trained=True)

# Fit `predictor2` to the training data.
# No need to call fit on `predictor2`. But if you do, it will be skipped.
predictor1.fit(X_train,y_train)

# Predict on X_new
y_pred1 = predictor1.predict(X_new)
y_pred2 = predictor2.fit(X_new)

Keras classification example:

from deel.puncc.api.prediction import BasePredictor
import tensorflow as tf
import numpy as np

# Generate data
X_train = np.random.uniform(0, 10, 1000)
X_train = np.expand_dims(X_train,-1)
X_new = np.random.uniform(0, 10, 100)
y_train = np.array([1 if x>5 else 0 for x in X_train])
y_train = tf.keras.utils.to_categorical(y_train)

# Instantiate a classifier as sequential model and add two dense layers
cl_model = tf.keras.Sequential()
cl_model.add(tf.keras.layers.Dense(10, activation="relu"))
cl_model.add(tf.keras.layers.Dense(2, activation="softmax"))

# The compile options need be passed to the constructor `BasePredictor`.
# The wrapper will call compile(**compile_kwargs) on an internal copy of the model.
# Our model is a classifier, we use categorical crossentropy as loss function.
compile_kwargs={"optimizer":"rmsprop", "loss":"categorical_crossentropy"}
predictor = BasePredictor(cl_model, is_trained=False, **compile_kwargs)

# The fit method is provided with a given training dataset (X,y) and
# with the train configuration of the underlying model. In the example below,
# we train the model over 5 epochs on batches containing 128 examples.
predictor.fit(X_train, y_train, **{"epochs":5,"batch_size":128})

# The `BasePredictor.predict` method enables to pass keyword arguments
# to the `predict`call of the underlying model (e.g., verbose).
y_pred = predictor.predict(X_new, **{"verbose":1})
copy()๏ƒ

Returns a copy of the predictor. The underlying model is either cloned (Keras model) or deepcopied (sklearn and similar models).

Returns:

copy of the predictor.

Return type:

BasePredictor

Raises:

RuntimeError โ€“ copy unsupported for provided models.

fit(X, y=None, **kwargs)๏ƒ

Fit model to the training data.

Parameters:

  • X (Iterable) โ€“ train features.

  • y (Optional[Iterable]) โ€“ train labels. Defaults to None (unsupervised).

  • kwargs โ€“ keyword arguments to be passed to the call fit() on the underlying model \(\hat{f}\).

Return type:

None

Note

For more details, check this code snippets.

get_is_trained()๏ƒ

Get flag that informs if the model is pre-trained.

Return type:

bool

predict(X, **kwargs)๏ƒ

Compute predictions on new examples.

Parameters:

  • X (Iterable) โ€“ new examplesโ€™ features.

  • kwargs (dict) โ€“ predict configuration to be passed to the predict method of the underlying model \(\hat{f}\).

Returns:

predictions \(\hat{f}(X)\) associated to the new examples X.

Return type:

ndarray

Note

For more details, check this code snippets.

class prediction.DualPredictor(models, is_trained=[False, False], compile_args=[{}, {}])๏ƒ

Bases: object

Wrapper of two joint point prediction models \((\hat{f_1},\hat{f_2})\). The prediction \(\hat{y}\) of a DualPredictor is a tuple \(\hat{y}=(\hat{y}_1,\hat{y}_2)\), where \(\hat{y}_1\) (resp. \(\hat{y}_2\)) is the prediction of \(\hat{f}_1\) (resp. \(\hat{f}_2\)).

Enables to standardize the interface of predictors and to expose generic fit(), predict() and copy() methods.

Parameters:

  • model (List[Any]) โ€“ list of two prediction models \([\hat{f_1},\hat{f_2}]\).

  • is_trained (List[bool]) โ€“ list of boolean flag that informs if the models are pre-trained. True value will skip the fitting of the corresponding model.

  • compile_kwargs (List) โ€“ list of keyword arguments to be used if needed to compile the underlying models.

Joint conditional mean / quantile regression example:

from deel.puncc.api.prediction import DualPredictor
import tensorflow_addons as tfa
import tensorflow as tf
from sklearn.ensemble import RandomForestRegressor

# Instantiate conditional mean model
rf_model = RandomForestRegressor(n_estimators=100)

# Instantiate 90-th quantile model
q_model = tf.keras.Sequential()
q_model.add(tf.keras.layers.Dense(10, activation="relu"))
q_model.add(tf.keras.layers.Dense(1, activation="relu"))
pinball_loss=tfa.losses.PinballLoss(tau=.9)
compile_kwargs={'optimizer':'sgd', 'loss':pinball_loss}

# The compile options need be passed to the constructor `DualPredictor`.
# The wrapper will call compile on the internal copy of each model if needed.
# Our predictor combines two regressors, of which only the second needs to be compiled.
tf_predictor = DualPredictor(models=[rf_model, q_model],
                             compile_args=[{}, compile_kwargs])

# The fit method is provided with a given training dataset (X,y) and
# the train configurations of the underlying models. In the example below,
# no specific configuration is passed to `rf_model` whereas we want to
# train the `q_model` over 10 epochs on batches containing 32 examples.
predictor.fit(X_train, y_train, [{}, {"epochs":10,"batch_size":32}])

# The `DualPredictor.predict` method enables to pass keyword arguments
# to the `predict` call of the underlying models. In the example below,
# we want to turn off the verbosity of `q_model.predict`.
# Besides, `y_pred` consists of a couple (y1, y2) for each new example.
# If `X_new` is a (n,m) matrix, the shape of `y_pred` will be (n, 2).
y_pred = predictor.predict(X_new, [{}, {"verbose":0}])
copy()๏ƒ

Returns a copy of the predictor. The underlying models are either cloned (Keras model) or deepcopied (sklearn and similar models).

Returns:

copy of the predictor.

Return type:

DualPredictor

Raises:

RuntimeError โ€“ copy unsupported for provided models.

fit(X, y, dictargs=[{}, {}])๏ƒ

Fit model to the training data.

Parameters:

  • X (Iterable) โ€“ train features.

  • y (Iterable) โ€“ train labels.

  • dictargs (List[dict[str]]) โ€“ list of fit configurations to be passed to the fit method of the underlying models \(\hat{f}_1\) and \(\hat{f}_2\), respectively.

Return type:

None

Note

For more details, check this code snippet.

get_is_trained()๏ƒ

Get flag that informs if the models are pre-trained. Returns True only when both models are pretrained.

Return type:

bool

predict(X, dictargs=[{}, {}])๏ƒ

Compute predictions on new examples.

Parameters:

  • X (Iterable) โ€“ new examplesโ€™ features.

  • kwargs (dict) โ€“ list of predict configurations to be passed to the predict method of the underlying models \(\hat{f}_1\) and \(\hat{f}_2\), respectively.

Returns:

predictions \(\hat{y}=\hat{f}(X)\) associated to the new examples X. For an instance \(X_i\), the prediction consists of a couple \(\hat{f}(X_i)=(\hat{f}_1(X_i), \hat{f}_2(X_i))\).

Return type:

Tuple[ndarray]

Raises:

NotImplementedError โ€“ predicted values not formated as numpy ndarrays.

Note

For more details, check this code snippet.

class prediction.IdPredictor(model=None, **kwargs)๏ƒ

Bases: BasePredictor

Subclass of BasePredictor to directly wrap existing predictions. The predictions are directly returned without any modification.

Parameters:

model โ€“ model to be wrapped.

Conformal regression example:

import numpy as np
from deel.puncc.api.prediction import IdPredictor
from deel.puncc.regression import SplitCP
from deel.puncc.metrics import regression_mean_coverage, regression_sharpness
from deel.puncc.plotting import plot_prediction_intervals

# Generate data
X = np.linspace(0, 20, 5000)
# randomly shuffle the data
np.random.shuffle(X)
X_calib, X_test = X[:4000], X[1000:]


# Define the real target function
def real_f(X):
    return 2 * X + np.random.randn(len(X)) * X * 0.5


# Target values for the calibration and new data
y_calib = real_f(X_calib)
y_test = real_f(X_test)


# Suppose we can obtain predictions from an API call
def api_call(X):
    return 2 * X  # This is a remote model that estimates of the target


# The model can still be conformalized using the calibration data as follows
## 1. Instantiate the Predictor wrapper, which serves as a container to host the predictions
dummy_predictor = IdPredictor()

## 2. CP method initialization
split_cp = SplitCP(
    dummy_predictor, train=False
)  # train=False to avoid trying to retrain the model internally

## 3. Request predictions on the calibration set
y_pred = api_call(X_calib)

## 4. The call to fit computes the nonconformity scores on the
## calibration set. Instead of features, we need to provide the
## predictions as X_calib along with the true target values.
split_cp.fit(X_calib=y_pred, y_calib=y_calib)

## 5. The predict method infers prediction intervals with respect to
## the significance level alpha = 10%. Make sure you provide the
## point predictions on the test set.
y_pred, y_pred_lower, y_pred_upper = split_cp.predict(
    api_call(X_test), alpha=0.1
)

# Compute marginal coverage and average width of the prediction intervals
coverage = regression_mean_coverage(y_test, y_pred_lower, y_pred_upper)
width = regression_sharpness(
    y_pred_lower=y_pred_lower, y_pred_upper=y_pred_upper
)
print(f"Marginal coverage: {np.round(coverage, 2)}")
print(f"Average width: {np.round(width, 2)}")

ax = plot_prediction_intervals(
    y_test, y_pred_lower, y_pred_upper, X=X_test, y_pred=y_pred
)
copy()๏ƒ

Returns a copy of the predictor. The underlying model is either cloned (Keras model) or deepcopied (sklearn and similar models).

Returns:

copy of the predictor.

Return type:

BasePredictor

Raises:

RuntimeError โ€“ copy unsupported for provided models.

fit(X, y=None, **kwargs)๏ƒ

Fit method is not supported for IdPredictor. Raises a RuntimeError.

Return type:

None

get_is_trained()๏ƒ

Get flag that informs if the model is pre-trained.

Return type:

bool

predict(X)๏ƒ

Returns the input argument as output data.

Parameters:

X (Iterable) โ€“ predictions.

Returns:

predictions.

Return type:

np.ndarray

predict_with_model(X)๏ƒ

Predicts the output using the wrapped model.

Parameters:

X (Iterable) โ€“ the input features to build predictions.

Returns:

predictions.

Return type:

np.ndarray

class prediction.MeanVarPredictor(models, is_trained=[False, False], compile_args=[{}, {}])๏ƒ

Bases: DualPredictor

Subclass of DualPredictor to specifically wrap a conditional mean estimator \(\hat{\mu}\) and a conditional dispersion estimator \(\hat{\sigma}\).

Specifically, the dispersion model \(\hat{\sigma}\) is trained on the mean absolute deviation of \(\hat{\mu}\)โ€™s predictions from the true labels \(y\). Given two training algorithms \({\cal A}_1\) and \({\cal A}_2\) and a training dataset \((X_{train}, y_{train})\):

\[\hat{\mu} \Leftarrow {\cal A}_1(X_{train}, y_{train})\]
\[\hat{\sigma} \Leftarrow {\cal A}_2(X_{train}, |\hat{\mu}(X_{train})-y_{train}|)\]
param List[Any] model:

list of two prediction models \([\hat{\mu},\hat{\sigma}]\).

param List[bool] is_trained:

list of boolean flag that informs if the models are pre-trained. True value will skip the fitting of the corresponding model.

param List compile_kwargs:

list of keyword arguments to be used if needed to compile the underlying models \(\hat{\mu}\) and \(\hat{\sigma}\), respectively.

Here follows an example of wrapping conditional mean and dispersion models:

from deel.puncc.api.prediction import MeanVarPredictor
from sklearn.ensemble import RandomForestRegressor

# Instantiate conditional mean model
mu_model = linear_model.LinearRegression()
# Instantiate conditional dispersion model
sigma_model = RandomForestRegressor(
    n_estimators=100, random_state=random_seed
)

# The instantiation of a :class:`MeanVarPredictor` is simple as the
# selected models do not require any compilation
mean_var_predictor = MeanVarPredictor([mu_model, sigma_model])

# The fit method is provided with a given training dataset (X,y).
# We do not choose any specific the train configurations.
mean_var_predictor.fit(X_train, y_train)

# The method `predict` yields `y_pred` that consists of a couple
# (y1, y2) for each new example.
# If `X_new` is a (n,m) matrix, the shape of `y_pred` will be (n, 2).
y_pred = mean_var_predictor.predict(X_new)

To see an example how to pass compilation/fit/predict configurations as arguments to the underlying models, check this code snippet.

copy()๏ƒ

Returns a copy of the predictor. The underlying models are either cloned (Keras model) or deepcopied (sklearn and similar models).

Returns:

copy of the predictor.

Return type:

DualPredictor

Raises:

RuntimeError โ€“ copy unsupported for provided models.

fit(X, y, dictargs=[{}, {}])๏ƒ

Fit models to the training data. The dispersion model \(\hat{\sigma}\) is trained on the mean absolute deviation of \(\hat{mu}\)โ€™s predictions \(\hat{\mu}\) from the true labels \(y\).

Parameters:

  • X (Iterable) โ€“ train features.

  • y (Iterable) โ€“ train labels.

  • dictargs (List[dict]) โ€“ list of fit configurations to be passed to the fit method of the underlying models \(\hat{\mu}\) and \(\hat{\sigma}\), respectively.

Return type:

None

get_is_trained()๏ƒ

Get flag that informs if the models are pre-trained. Returns True only when both models are pretrained.

Return type:

bool

predict(X, dictargs=[{}, {}])๏ƒ

Compute predictions on new examples.

Parameters:

  • X (Iterable) โ€“ new examplesโ€™ features.

  • kwargs (dict) โ€“ list of predict configurations to be passed to the predict method of the underlying models \(\hat{f}_1\) and \(\hat{f}_2\), respectively.

Returns:

predictions \(\hat{y}=\hat{f}(X)\) associated to the new examples X. For an instance \(X_i\), the prediction consists of a couple \(\hat{f}(X_i)=(\hat{f}_1(X_i), \hat{f}_2(X_i))\).

Return type:

Tuple[ndarray]

Raises:

NotImplementedError โ€“ predicted values not formated as numpy ndarrays.

Note

For more details, check this code snippet.