Skip to content

Commit

Permalink
ENH: add gp cross conformuty score
Browse files Browse the repository at this point in the history
  • Loading branch information
vincentblot28 committed Aug 1, 2023
1 parent 6f3f458 commit 501d28f
Show file tree
Hide file tree
Showing 5 changed files with 676 additions and 56 deletions.
3 changes: 1 addition & 2 deletions mapie/conformity_scores/conformity_scores.py
Original file line number Diff line number Diff line change
Expand Up @@ -183,7 +183,6 @@ def get_conformity_scores(
X: ArrayLike,
y: ArrayLike,
y_pred: ArrayLike,
y_std: Union[ArrayLike, None]
) -> NDArray:
"""
Get the conformity score considering the symmetrical property if so.
Expand All @@ -204,7 +203,7 @@ def get_conformity_scores(
NDArray of shape (n_samples,)
Conformity scores.
"""
conformity_scores = self.get_signed_conformity_scores(X, y, y_pred, y_std)
conformity_scores = self.get_signed_conformity_scores(X, y, y_pred)
if self.consistency_check:
self.check_consistency(X, y, y_pred, conformity_scores)
if self.sym:
Expand Down
138 changes: 130 additions & 8 deletions mapie/conformity_scores/residual_conformity_scores.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
from mapie._typing import ArrayLike, NDArray

from mapie.conformity_scores import ConformityScore
from mapie.estimator.interface import EnsembleEstimator


class AbsoluteConformityScore(ConformityScore):
Expand Down Expand Up @@ -589,7 +590,6 @@ def get_signed_conformity_scores(
X: ArrayLike,
y: ArrayLike,
y_pred: ArrayLike,
y_std: Union[ArrayLike, None]
) -> NDArray:
"""
Computes the signed conformity score = (y - y_pred) / r_pred.
Expand Down Expand Up @@ -621,7 +621,9 @@ def get_signed_conformity_scores(
else:
cal_indexes = full_indexes

_, std_estim = self.residual_estimator.predict(X[cal_indexes], return_std=True)
_, std_estim = self.residual_estimator.predict(
X[cal_indexes], return_std=True
)
residuals_pred = np.maximum(
std_estim,
self.eps
Expand Down Expand Up @@ -662,12 +664,10 @@ def get_estimation_distribution(
``conformity_scores`` can be either the conformity scores or
the quantile of the conformity scores.
"""
# import pdb; pdb.set_trace()
_, r_pred = self.residual_estimator_.predict(X, return_std=True)
# import pdb; pdb.set_trace()
print(conformity_scores)
return np.add(y_pred, np.multiply(conformity_scores, r_pred.reshape(-1, 1)))

return np.add(
y_pred, np.multiply(conformity_scores, r_pred.reshape(-1, 1))
)


class GPCrossConformityScore(ConformityScore):
Expand All @@ -684,7 +684,7 @@ class GPCrossConformityScore(ConformityScore):
def __init__(
self,
) -> None:
super().__init__(sym=True, consistency_check=True)
super().__init__(sym=True, consistency_check=False)

def get_signed_conformity_scores(
self,
Expand All @@ -701,6 +701,40 @@ def get_signed_conformity_scores(
y_std = np.maximum(self.eps, y_std)
return np.subtract(y, y_pred) / y_std

def get_conformity_scores(
self,
X: ArrayLike,
y: ArrayLike,
y_pred: ArrayLike,
y_std: Union[ArrayLike, None]
) -> NDArray:
"""
Get the conformity score considering the symmetrical property if so.
Parameters
----------
X: NDArray of shape (n_samples, n_features)
Observed feature values.
y: NDArray of shape (n_samples,)
Observed target values.
y_pred: NDArray of shape (n_samples,)
Predicted target values.
Returns
-------
NDArray of shape (n_samples,)
Conformity scores.
"""
conformity_scores = self.get_signed_conformity_scores(
X, y, y_pred, y_std
)
if self.consistency_check:
self.check_consistency(X, y, y_pred, conformity_scores)
if self.sym:
conformity_scores = np.abs(conformity_scores)
return conformity_scores

def get_estimation_distribution(
self,
Expand All @@ -718,3 +752,91 @@ def get_estimation_distribution(
the quantile of the conformity scores.
"""
return np.add(y_pred, conformity_scores)

def get_bounds(
self,
X: ArrayLike,
estimator: EnsembleEstimator,
conformity_scores: NDArray,
alpha_np: NDArray,
ensemble: bool,
method: str
) -> Tuple[NDArray, NDArray, NDArray]:
"""
Compute bounds of the prediction intervals from the observed values,
the estimator of type ``EnsembleEstimator`` and the conformity scores.
Parameters
----------
X: ArrayLike of shape (n_samples, n_features)
Observed feature values.
estimator: EnsembleEstimator
Estimator that is fitted to predict y from X.
conformity_scores: ArrayLike of shape (n_samples,)
Conformity scores.
alpha_np: NDArray of shape (n_alpha,)
NDArray of floats between ``0`` and ``1``, represents the
uncertainty of the confidence interval.
ensemble: bool
Boolean determining whether the predictions are ensembled or not.
method: str
Method to choose for prediction interval estimates.
The ``"plus"`` method implies that the quantile is calculated
after estimating the bounds, whereas the other methods
(among the ``"naive"``, ``"base"`` or ``"minmax"`` methods,
for example) do the opposite.
Returns
-------
Tuple[NDArray, NDArray, NDArray]
- The predictions itself. (y_pred) of shape (n_samples,).
- The lower bounds of the prediction intervals of shape
(n_samples, n_alpha).
- The upper bounds of the prediction intervals of shape
(n_samples, n_alpha).
"""
y_pred, y_pred_low, y_pred_up, y_std_multi = estimator.predict(
X, ensemble
)
signed = -1 if self.sym else 1
conformity_scores = conformity_scores * y_std_multi
if method == "plus":
alpha_low = alpha_np if self.sym else alpha_np / 2
alpha_up = 1 - alpha_np if self.sym else 1 - alpha_np / 2

conformity_scores_low = self.get_estimation_distribution(
X, y_pred_low, signed * conformity_scores
)
conformity_scores_up = self.get_estimation_distribution(
X, y_pred_up, conformity_scores
)
bound_low = self.get_quantile(
conformity_scores_low, alpha_low, axis=1, method="lower"
)
bound_up = self.get_quantile(
conformity_scores_up, alpha_up, axis=1, method="higher"
)
else:
quantile_search = "higher" if self.sym else "lower"
alpha_low = 1 - alpha_np if self.sym else alpha_np / 2
alpha_up = 1 - alpha_np if self.sym else 1 - alpha_np / 2

quantile_low = self.get_quantile(
conformity_scores, alpha_low, axis=0, method=quantile_search
)
quantile_up = self.get_quantile(
conformity_scores, alpha_up, axis=0, method="higher"
)
bound_low = self.get_estimation_distribution(
X, y_pred_low, signed * quantile_low
)
bound_up = self.get_estimation_distribution(
X, y_pred_up, quantile_up
)

return y_pred, bound_low, bound_up
Loading

0 comments on commit 501d28f

Please sign in to comment.