Coverage for uqmodels/modelization/UQEstimator.py: 63%
157 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-09-05 14:29 +0000
1################################################################################
2# Source that specify UQEstimator class : Estimator performing Uncertainty
3# quantification according to serveral paradigms
4from abc import ABCMeta, abstractmethod
5from typing import Callable, TypeVar
7import numpy as np
8from sklearn.base import BaseEstimator
9from sklearn.metrics import make_scorer
10from sklearn.preprocessing import StandardScaler
12import uqmodels.postprocessing.UQ_processing as UQ_proc
13from uqmodels.utils import aux_tuning
15# To do : Specify an UQmeasure object & specify the type_UQ list
17Array = TypeVar("Array")
18List_Estimators = TypeVar("List_Estimators")
19Estimator = TypeVar("Estimator")
20Dict_params = TypeVar("Dict_params")
22# Current support type :
23list_type_UQ = [
24 "None",
25 "var",
26 "res_var",
27 "2var",
28 "res_2var",
29 "var_A&E",
30 "quantile",
31 "res_quantile",
32]
33# To Do : add var_residuals, 2var_residuals, quantile_residuals
36def check_type_UQ(type_UQ):
37 if type_UQ in list_type_UQ:
38 None
39 else:
40 raise ValueError(type_UQ, " not in ", list_type_UQ)
43def get_UQEstimator_parameters(
44 model_parameters={},
45 factory_parameters={},
46 training_parameters=None,
47 type_output=None,
48 rescale=False,
49 random_state=None,
50 **kwargs,
51):
52 """Generate dict object containing UQEstimators parameters to provide to init().
54 Args:
55 model_parameters (dict, optional): UQestimators parameters. Defaults to {}.
56 factory_parameters (dict, optional): Factory parameters, to data preprocessing. Defaults to {}.
57 training_parameters (_type_, optional): Model training parameters for DEEP estimators. Defaults to None.
58 type_output (_type_, optional): Specification of deep learning output. Defaults to None.
59 rescale (bool, optional): Use or not internal rescale function Defaults to False.
60 random_state (bool): handle experimental random using seed.
61 Returns:
62 dict: UQEstimators parameters
63 """
64 UQEstimator_parameters = {
65 "factory_parameters": factory_parameters,
66 "model_parameters": model_parameters,
67 "factory_parameters": factory_parameters,
68 "training_parameters": training_parameters,
69 "type_output": type_output,
70 "rescale": rescale,
71 "random_state": random_state,
72 }
74 for key_arg in kwargs.keys():
75 UQEstimator_parameters[key_arg] = kwargs[key_arg]
77 return UQEstimator_parameters
80class UQEstimator(BaseEstimator, metaclass=ABCMeta):
81 """Abstract structure of a UQEstimator : Estimator (fit/predict) that
82 perform prediction(optionaly) and UQmeasure estimation
83 """
85 def __init__(
86 self,
87 name="UQEstimator",
88 type_UQ="None",
89 rescale=False,
90 type_UQ_params=None,
91 var_min=0.000001,
92 random_state=None,
93 ):
94 """UQestimator that perform prediction(optionaly) and UQmeasure estimation
96 Args:
97 name (str, optional): name. Defaults to 'UQEstimator'.
98 type_UQ (str, optional): nature of UQmeasure. Defaults to 'None'.
99 rescale (bool, optional): boolean flag that specify to use or not rescalling procedure. Defaults to False.
100 type_UQ_params (_type_, optional): additional information about UQMeasure parameters. Defaults to None.
101 random_state (bool): handle experimental random using seed.
102 """
103 self.name = name
104 self.type_UQ = type_UQ
105 self.type_UQ_params = type_UQ_params
106 self.rescale = rescale
107 self.random_state = random_state
108 self.var_min = var_min
109 self.is_fitted = False
111 self.len_y_shape = None
112 self.scaler = [
113 StandardScaler(with_mean=True, with_std=True),
114 StandardScaler(with_mean=True, with_std=True),
115 ]
116 self.random_state = random_state
118 def _format(self, X, y, type_transform=False, mode_UQ=False, skip_format=False):
119 """Data normalisation procedure (apply StandardScaler scikit learn) apply to features, target,
120 prediction and UQmeasure design to hold UQ for several type of UQ use renormalise_UQ function of UQ_processing
122 Args:
123 X (_type_): Features
124 y (_type_): Targets/Observation or prediction or UQmeasure)
125 type_transform: action : {fit_transform, : fit scaler and transform
126 transform, : use fitted scaler to transform
127 inverse_transform}
128 mode_UQ (bool, optional): true if format UQ measure
130 Returns:
131 X,y: Normalised features, and targets (or prediction or UQmeasure)
132 """
133 if skip_format:
134 return (X, y)
136 scalerX, scalerY = self.scaler
138 if isinstance(y, tuple):
139 y = np.array(list(y))
141 if type_transform == "fit_transform":
142 self.len_y_shape = len(y.shape)
144 if not (hasattr(self, "len_y_shape")):
145 if y is not None:
146 self.len_y_shape = len(y.shape)
148 if self.rescale:
149 if y is not None:
150 if len(y.shape) == 1:
151 y = y[:, None]
153 if type_transform == "fit_transform": # Fit X&Y Scaler
154 if X is not None:
155 X = self.scaler[0].fit_transform(X)
156 if y is not None:
157 y = self.scaler[1].fit_transform(y)
159 elif type_transform == "transform":
160 if X is not None:
161 X = self.scaler[0].transform(X)
162 if y is not None:
163 y = self.scaler[1].transform(y)
165 elif type_transform == "inverse_transform": # Inverse Transform pred or UQ
166 X_transformer = self.scaler[0].inverse_transform
167 Y_transformer = self.scaler[1].inverse_transform
168 if X is not None and not (len(X) == 0):
169 X = X_transformer(X)
171 if y is not None and not (len(y) == 0):
172 if mode_UQ: # UQ
173 y = UQ_proc.renormalise_UQ(
174 y, self.type_UQ, self.scaler[1], var_min=self.var_min
175 )
177 else:
178 y = Y_transformer(y)
180 if y is not None:
181 if self.len_y_shape == 1:
182 y = np.squeeze(y)
184 return (X, y)
186 @abstractmethod
187 def fit(self, X: np.array, y: np.array, skip_format=False, **kwargs) -> None:
188 """Fit UQestimator using training data.
189 Args:
190 X: train features
191 y: train targets/observations
192 """
194 @abstractmethod
195 def predict(self, X: np.array, skip_format=False, **kwargs):
196 """Compute prediction (or provide None) and UQ-measure
197 Args:
198 X: features
199 Returns:
200 pred, UQ_measure
201 """
203 def _tuning(
204 self,
205 estimator: Estimator,
206 X: Array,
207 y: Array,
208 n_esti: int = 100,
209 folds: int = 4,
210 score: str = "neg_mean_squared_error",
211 params=None,
212 **kwarg,
213 ) -> None:
214 """Fine-tune an estimator using a standards scikit learns gridsearch procedure on params (grid_parameters)
215 Args:
216 X: features
217 y: targets/observations
218 """
219 return aux_tuning(
220 estimator,
221 X,
222 y,
223 params,
224 score,
225 n_esti,
226 folds,
227 random_state=self.random_state,
228 )
230 def get_params(self, deep=False):
231 dict_params = {}
232 for key in self.__dict__.keys():
233 if hasattr(self.__getattribute__(key), "get_params"):
234 dict_params[key] = self.__getattribute__(key).get_params(deep=False)
235 else:
236 dict_params[key] = self.__getattribute__(key)
237 return dict_params
239 def factory(
240 self, X, y, type_transform="transform", only_fit_scaler=False, **kwargs
241 ):
242 if only_fit_scaler:
243 type_transform = "fit_transform"
244 if y is None:
245 X, _ = self._format(X, None, type_transform=type_transform)
246 else:
247 X, y = self._format(X, y, type_transform=type_transform)
248 return (X, y, None)
251class MeanVarUQEstimator(UQEstimator):
252 """Mean var UQ Estimator that estimate a mean values and estimate irreductible"""
254 def __init__(
255 self,
256 estimator=None,
257 estimator_var=None,
258 type_UQ="var",
259 name="MeanVarUQEstimator",
260 rescale=False,
261 var_min=0.00001,
262 random_state=None,
263 ):
264 """Initialization
266 Args:
267 name (str): Name
268 estimator (estimator): Target predictor
269 estimator_var (estimator): variance estimators
270 random_state (bool): handle experimental random using seed.
271 """
273 super().__init__(
274 name=name,
275 type_UQ=type_UQ,
276 rescale=rescale,
277 var_min=var_min,
278 random_state=random_state,
279 )
280 self.estimator = estimator
281 self.estimator_var = estimator_var
283 def fit(self, X, y, skip_format=False, **kwargs):
284 """Fit UQestimator using training data.
285 Args:
286 X: train features
287 y: train targets/observations
288 """
289 X, y = self._format(X, y, "fit_transform", skip_format=skip_format)
290 self.estimator.fit(X, y)
291 pred = self.estimator.predict(X)
292 residual = np.power(np.squeeze(y) - np.squeeze(pred), 2)
293 residual = residual.reshape(y.shape)
294 self.estimator_var.fit(X, residual)
295 self.is_fitted = True
297 def predict(self, X, skip_format=False, **kwargs):
298 """Compute prediction (or provide None) and UQ-measure
299 Args:
300 X: features
301 Returns:
302 pred, UQ_measure"""
304 X, _ = self._format(X, None, "fit", skip_format=skip_format)
306 pred = self.estimator.predict(X)
307 UQ = self.estimator_var.predict(X)
309 _, pred = self._format(None, pred, "inverse_transform")
310 _, UQ = self._format(None, UQ, "inverse_transform", mode_UQ=True)
311 return pred, UQ
313 def _format(
314 self,
315 X: np.array,
316 y: np.array,
317 type_transform: str,
318 mode_UQ: bool = False,
319 skip_format=False,
320 ):
321 """Data normalisation procedure : see UQModels _gormat
323 Args:
324 X (np.array): _description_
325 y (np.array): _description_
326 type_transform (str): _description_
327 mode_UQ (bool, optional): _description_. Defaults to False.
329 Returns:
330 _type_: _description_
331 """
332 X, y = super()._format(X, y, type_transform, mode_UQ, skip_format=skip_format)
333 return (X, y)
336class QuantileUQEstimator(UQEstimator):
337 def __init__(
338 self,
339 list_estimators,
340 list_alpha: list = [0.025, 0.5, 0.975],
341 type_UQ: str = "quantile",
342 name: str = "QuantileUQEstimator",
343 rescale=False,
344 var_min=0.00001,
345 random_state=None,
346 **kwargs,
347 ) -> None:
348 """Initialise all attributes of the UQEstimatorGBRQ class.
350 Args:
351 list_estimator: List of provided quatile estimators by default use GradientBoostingRegressor
352 as default estimator.
353 list_alpha: List of quantile values to estimates.
354 type_quantile : quantile or res_quantile : express if y provided is already a residual or None
355 list_alpha :list of alpha conf quantile to estimate
357 """
358 super().__init__(
359 name=name,
360 type_UQ=type_UQ,
361 rescale=rescale,
362 var_min=var_min,
363 random_state=random_state,
364 )
365 self.list_alpha = list_alpha
366 self.type_UQ_params = {"list_alpha": list_alpha}
367 self.list_estimators = list_estimators
369 def fit(self, X, y, skip_format=False, **kwargs):
370 X, y = self._format(X, y, "fit_transform", skip_format=skip_format)
371 for estimator in self.list_estimators:
372 estimator.fit(X, y)
373 self.is_fitted = True
375 def predict(self, X, skip_format=False, **kwargs):
376 pred = None
377 X, _ = self._format(X, None, "fit", skip_format=skip_format)
379 if 0.5 in self.type_UQ_params["list_alpha"]:
380 idx = self.type_UQ_params["list_alpha"].index(0.5)
381 pred = self.list_estimators[idx].predict(X)
383 UQ = []
384 for n, estimator in enumerate(self.list_estimators):
385 UQ.append(estimator.predict(X))
386 UQ = np.stack(UQ)
387 _, pred = self._format(None, pred, "inverse_transform", mode_UQ=False)
388 _, UQ = self._format(None, UQ, "inverse_transform", mode_UQ=True)
390 return pred, UQ
392 def _format(
393 self,
394 X: np.array,
395 y: np.array,
396 type_transform: str,
397 mode_UQ: bool = False,
398 skip_format=False,
399 ):
400 """Data normalisation procedure (apply StandardScaler scikit learn) apply to features, target,
401 prediction and UQmeasure.
402 Designed to hold UQ for several type of UQ use renormalise_UQ function of UQ_processing
404 Args:
405 X (_type_): Features
406 y (_type_): Targets/Observation or prediction or UQmeasure)
407 type_transform: action : {fit_transform, : fit scaler and transform
408 transform, : use fitted scaler to transform
409 inverse_transform}
410 mode_UQ (bool, optional): true if format UQ measure
412 Returns:
413 X,y: Normalised features, and targets (or prediction or UQmeasure)
414 """
415 X, y = super()._format(
416 X=X,
417 y=y,
418 type_transform=type_transform,
419 mode_UQ=mode_UQ,
420 skip_format=skip_format,
421 )
422 return (X, y)
424 def _tuning(
425 self,
426 X: Array,
427 y: Array,
428 n_esti: int = 100,
429 folds: int = 4,
430 params: Dict_params = None,
431 **kwarg,
432 ) -> None:
433 """Perform a random search tuning using a parameter grid."""
434 X, y = self._format(X, y, "fit_transform")
436 def make_q_loss(beta: float) -> Callable:
437 """Build the quantile loss calculation function.
439 Args:
440 beta: Must be between 0 and 1. Miss-coverage rate.
442 Returns:
443 The quantile loss function.
444 """
446 def quantile_loss(y_true: Array, pred: Array) -> float:
447 """Compute the quantile loss between a vector of
448 predicted values and one of ground truth values for y.
450 The beta parameter is set outside the function (as an
451 argument to the maker function).
453 Args:
454 y_true: vector of ground truth values
455 pred: vector of predicted values
457 Returns:
458 The loss score.
459 """
460 delta = y_true - pred
461 score = (
462 (beta - 1.0) * delta * (delta < 0) + beta * delta * (delta >= 0)
463 ).sum()
464 return score
466 return quantile_loss
468 for n, estimator in enumerate(self.list_estimators):
469 alpha = self.list_alpha[n]
470 if params is not None:
471 # Mean estimator tuning
472 if hasattr(self, "pretuned"):
473 if self.pretuned:
474 score = make_scorer(
475 make_q_loss(alpha),
476 greater_is_better=False,
477 )
478 self.list_estimators[n] = super()._tuning(
479 estimator=estimator,
480 X=X,
481 y=y,
482 params=params,
483 score=score,
484 n_esti=n_esti,
485 folds=folds,
486 )
488 # else no tuning