Skip to content

edsteva.models.base

BaseModel

Base class for Models

ATTRIBUTE DESCRIPTION
_coefs

The list of the Model coefficients

TYPE: List[str]

estimates

Available with the fit() method

TYPE: pd.DataFrame

_metrics

Available with the fit() method

The list of computed metrics if any

TYPE: List[str]

params

Available with the fit() method

Ths list of extra keyword parameters used.

TYPE: List[str]

Source code in edsteva/models/base.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
class BaseModel(metaclass=ABCMeta):

    """Base class for Models

    Attributes
    ----------
     _coefs: List[str]
        The list of the Model coefficients
    estimates: pd.DataFrame
        Available with the [``fit()``][edsteva.models.base.BaseModel.fit] method
    _metrics: List[str]
        Available with the [``fit()``][edsteva.models.base.BaseModel.fit] method

        The list of computed metrics if any
    params: List[str]
        Available with the [``fit()``][edsteva.models.base.BaseModel.fit] method

        Ths list of extra keyword parameters used.
    """

    def __init__(
        self,
        algo: str,
        coefs: List[str],
        default_metrics: List[str],
    ):
        self._algo = algo
        self._coefs = coefs
        self._default_metrics = default_metrics
        self._viz_config = {}

    def is_computed_estimates(self) -> None:
        """Raises an error if the Probe has not been fitted properly"""
        if hasattr(self, "estimates"):
            if isinstance(self.estimates, pd.DataFrame):
                if len(self.estimates) == 0:
                    raise Exception(
                        "Estimates are empty, please review the process method or your arguments"
                    )
            else:
                raise Exception(
                    "The fit process must return a Pandas Dataframe and not {}".format(
                        type(self.estimates).__name__
                    )
                )

        else:
            raise Exception(
                "Model has not been fitted, please use the fit method as follow: Model.fit()"
            )

    @abstractmethod
    def fit_process(
        self,
        predictor: pd.DataFrame,
        index: List[str] = None,
        **kwargs,
    ):
        """Fit the Probe in order to obtain estimates"""

    @abstractmethod
    def predict_process(
        self,
        prediction: pd.DataFrame,
        **kwargs,
    ):
        """Compute the predicted Probe"""

    def fit(
        self,
        probe: BaseProbe,
        metric_functions: List[str] = None,
        start_date: str = None,
        end_date: str = None,
        with_cache: bool = True,
        **kwargs,
    ) -> None:
        """Fit the model to the probe instance

        Parameters
        ----------
        probe : BaseProbe
            Target variable to be fitted
        metric_functions : List[str], optional
            Metrics to apply on the fitted Probe. By default it will apply the default metric specified in the model.

            **EXAMPLE**: `[error, error_after_t0]`
        start_date : str, optional
            **EXAMPLE**: `"2019-05-01"`
        end_date : str, optional
            **EXAMPLE**: `"2021-07-01"`

        Examples
        --------
        ```python
        from edsteva.models.step_function import StepFunction

        step_function_model = StepFunction()
        step_function_model.fit(probe)
        step_function_model.estimates.head()
        ```

        | care_site_level          | care_site_id | stay_type    | t_0        | c_0   | error |
        | :----------------------- | :----------- | :----------- | :--------- | :---- | :---- |
        | Unité Fonctionnelle (UF) | 8312056386   | 'Urg_Hospit' | 2019-05-01 | 0.397 | 0.040 |
        | Unité Fonctionnelle (UF) | 8312056386   | 'All'        | 2011-04-01 | 0.583 | 0.028 |
        | Pôle/DMU                 | 8312027648   | 'Urg_Hospit' | 2021-03-01 | 0.677 | 0.022 |
        | Pôle/DMU                 | 8312027648   | 'All'        | 2018-08-01 | 0.764 | 0.014 |
        | Hôpital                  | 8312022130   | 'Urg_Hospit' | 2022-02-01 | 0.652 | 0.027 |
        """
        if isinstance(probe, BaseProbe):
            probe.is_computed_probe()
        else:
            raise TypeError(
                "Unsupported type {} for probe.".format(type(probe).__name__)
            )

        predictor = filter_table_by_date(
            table=probe.predictor,
            table_name="predictor",
            start_date=start_date,
            end_date=end_date,
        )
        index = probe._index

        estimates = self.fit_process(
            predictor=predictor,
            index=index,
            **kwargs,
        )

        metrics_df = self._compute_metrics(
            predictor=predictor,
            estimates=estimates,
            index=index,
            metric_functions=metric_functions,
        )

        if metrics_df is not None:
            self._metrics = list(metrics_df.columns.difference(index))
            self.estimates = estimates.merge(metrics_df, on=index)

        else:
            self.estimates = estimates

        self.is_computed_estimates()
        self.params = kwargs
        if with_cache:
            self.cache_estimates()

    def reset_estimates(
        self,
    ) -> None:
        """Reset the estimates to its initial state"""
        self.estimates = self._cache_estimates.copy()

    def cache_estimates(
        self,
    ) -> None:
        """Cache the predictor"""
        self._cache_estimates = self.estimates.copy()
        logger.info(
            "Cache the estimates, you can reset the estimates to this state with the method reset_estimates"
        )

    def predict(
        self,
        probe: BaseProbe,
    ) -> pd.DataFrame:
        """Computes the predicted probe by using the estimates

        Parameters
        ----------
        probe : BaseProbe
            Target variable to be predicted

        Examples
        --------
        ```python
        from edsteva.models.step_function import StepFunction

        step_function_model.predict(visit).head()
        ```

        | care_site_level          | care_site_id | stay_type    | date       | n_visit | c     | c_fit |
        | :----------------------- | :----------- | :----------- | :--------- | :------ | :---- | :---- |
        | Unité Fonctionnelle (UF) | 8312056386   | 'Urg_Hospit' | 2019-05-01 | 233.0   | 0.841 | 0.758 |
        | Unité Fonctionnelle (UF) | 8312056386   | 'All'        | 2021-04-01 | 393.0   | 0.640 | 0.758 |
        | Pôle/DMU                 | 8312027648   | 'Urg_Hospit' | 2011-03-01 | 204.0   | 0.497 | 0     |
        | Pôle/DMU                 | 8312027648   | 'All'        | 2018-08-01 | 22.0    | 0.784 | 0.874 |
        | Hôpital                  | 8312022130   | 'Urg_Hospit' | 2022-02-01 | 9746.0  | 0.974 | 0.912 |

        """

        predictor = probe.predictor
        index = probe._index

        return self.predict_process(predictor=predictor, index=index)

    def load(self, path=None) -> None:
        """Loads a Model from local

        Parameters
        ----------
        path : str, optional
            **EXAMPLE**: `"my_folder/my_file.html"`

        Examples
        -------
        ```python
        from edsteva.probes import VisitProbe

        probe_path = "my_path/visit.pkl"

        visit = VisitProbe()
        visit.load(path=probe_path)
        ```

        """

        path = path or self._get_path()
        loaded_model = load_object(path)
        self.__dict__ = loaded_model.__dict__.copy()
        self.path = path

    def save(self, path: str = None, name: str = None) -> bool:
        """Saves computed Model instance

        Parameters
        ----------
        path : str, optional
            **EXAMPLE**: `"my_folder/my_file.html"`
        name : str, optional
            **EXAMPLE**: `"fitted_visit"`

        Examples
        -------
        ```python
        from edsteva.probes import VisitProbe

        probe_path = "my_path/visit.pkl"

        visit = VisitProbe()
        visit.compute(data)
        visit.save(path=probe_path)
        ```

        """

        self.is_computed_estimates()

        if name:
            self.name = name
        if not path:
            path = self._get_path()

        self.path = path
        save_object(self, path)

    def delete(self, path: str = None) -> bool:
        """Delete the saved Model instance

        Parameters
        ----------
        path : str, optional
            **EXAMPLE**: `"my_folder/my_file.html"`
        """
        if not path:
            path = self.path

        delete_object(self, path)

    def _get_path(self):
        base_path = CACHE_DIR / "edsteva" / "models"
        if hasattr(self, "name"):
            filename = f"{self.name.lower()}.pickle"
        else:
            filename = f"{type(self).__name__.lower()}.pickle"
        return base_path / filename

    def _compute_metrics(
        self,
        predictor: pd.DataFrame,
        estimates: pd.DataFrame,
        index: List[str],
        metric_functions: List[str] = None,
    ):
        if metric_functions is None:
            if hasattr(self, "_default_metrics") and self._default_metrics:
                metric_functions = self._default_metrics
            else:
                return None
        if isinstance(metric_functions, str):
            metric_functions = [metric_functions]
        metrics_df = []
        for metric_function in metric_functions:
            metrics_df.append(
                metrics.get(metric_function)(
                    predictor=predictor, estimates=estimates, index=index
                )
            )
        return reduce(lambda left, right: left.merge(right, on=index), metrics_df)

    def is_predictable_probe(
        self,
        predictor: pd.DataFrame,
        index: List[str],
    ) -> pd.DataFrame:
        """Raises an error if the model has not been fitted on the input predictor.

        Parameters
        ----------
        predictor : pd.DataFrame
            Target DataFrame to be predicted
        index : List[str]
            List of the columns given by Probe._index

        Returns
        -------
        pd.DataFrame
            Predictor along with the fitted estimates

        Raises
        ------
        Exception
            Some indexes have no associated estimates, the model must be fitted on an adequate probe
        """
        prediction = predictor.merge(
            self.estimates, on=index, how="left", validate="many_to_one", indicator=True
        )
        if (prediction["_merge"] == "both").all():
            return prediction.drop(columns="_merge")

        raise Exception(
            "Some indexes have no associated estimates, the model must be fitted on an adequate probe"
        )

is_computed_estimates 

is_computed_estimates() -> None

Raises an error if the Probe has not been fitted properly

Source code in edsteva/models/base.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def is_computed_estimates(self) -> None:
    """Raises an error if the Probe has not been fitted properly"""
    if hasattr(self, "estimates"):
        if isinstance(self.estimates, pd.DataFrame):
            if len(self.estimates) == 0:
                raise Exception(
                    "Estimates are empty, please review the process method or your arguments"
                )
        else:
            raise Exception(
                "The fit process must return a Pandas Dataframe and not {}".format(
                    type(self.estimates).__name__
                )
            )

    else:
        raise Exception(
            "Model has not been fitted, please use the fit method as follow: Model.fit()"
        )

fit_process abstractmethod 

fit_process(
    predictor: pd.DataFrame,
    index: List[str] = None,
    **kwargs
)

Fit the Probe in order to obtain estimates

Source code in edsteva/models/base.py 
66
67
68
69
70
71
72
73
@abstractmethod
def fit_process(
    self,
    predictor: pd.DataFrame,
    index: List[str] = None,
    **kwargs,
):
    """Fit the Probe in order to obtain estimates"""

predict_process abstractmethod 

predict_process(prediction: pd.DataFrame, **kwargs)

Compute the predicted Probe

Source code in edsteva/models/base.py 
75
76
77
78
79
80
81
@abstractmethod
def predict_process(
    self,
    prediction: pd.DataFrame,
    **kwargs,
):
    """Compute the predicted Probe"""

fit 

fit(
    probe: BaseProbe,
    metric_functions: List[str] = None,
    start_date: str = None,
    end_date: str = None,
    with_cache: bool = True,
    **kwargs
) -> None

Fit the model to the probe instance

PARAMETER DESCRIPTION
probe

Target variable to be fitted

TYPE: BaseProbe

metric_functions

Metrics to apply on the fitted Probe. By default it will apply the default metric specified in the model.

EXAMPLE: [error, error_after_t0]

TYPE: List[str] DEFAULT: None

start_date

EXAMPLE: "2019-05-01"

TYPE: str DEFAULT: None

end_date

EXAMPLE: "2021-07-01"

TYPE: str DEFAULT: None

Examples:

from edsteva.models.step_function import StepFunction

step_function_model = StepFunction()
step_function_model.fit(probe)
step_function_model.estimates.head()
care_site_level care_site_id stay_type t_0 c_0 error
Unité Fonctionnelle (UF) 8312056386 'Urg_Hospit' 2019-05-01 0.397 0.040
Unité Fonctionnelle (UF) 8312056386 'All' 2011-04-01 0.583 0.028
Pôle/DMU 8312027648 'Urg_Hospit' 2021-03-01 0.677 0.022
Pôle/DMU 8312027648 'All' 2018-08-01 0.764 0.014
Hôpital 8312022130 'Urg_Hospit' 2022-02-01 0.652 0.027
Source code in edsteva/models/base.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def fit(
    self,
    probe: BaseProbe,
    metric_functions: List[str] = None,
    start_date: str = None,
    end_date: str = None,
    with_cache: bool = True,
    **kwargs,
) -> None:
    """Fit the model to the probe instance

    Parameters
    ----------
    probe : BaseProbe
        Target variable to be fitted
    metric_functions : List[str], optional
        Metrics to apply on the fitted Probe. By default it will apply the default metric specified in the model.

        **EXAMPLE**: `[error, error_after_t0]`
    start_date : str, optional
        **EXAMPLE**: `"2019-05-01"`
    end_date : str, optional
        **EXAMPLE**: `"2021-07-01"`

    Examples
    --------
    ```python
    from edsteva.models.step_function import StepFunction

    step_function_model = StepFunction()
    step_function_model.fit(probe)
    step_function_model.estimates.head()
    ```

    | care_site_level          | care_site_id | stay_type    | t_0        | c_0   | error |
    | :----------------------- | :----------- | :----------- | :--------- | :---- | :---- |
    | Unité Fonctionnelle (UF) | 8312056386   | 'Urg_Hospit' | 2019-05-01 | 0.397 | 0.040 |
    | Unité Fonctionnelle (UF) | 8312056386   | 'All'        | 2011-04-01 | 0.583 | 0.028 |
    | Pôle/DMU                 | 8312027648   | 'Urg_Hospit' | 2021-03-01 | 0.677 | 0.022 |
    | Pôle/DMU                 | 8312027648   | 'All'        | 2018-08-01 | 0.764 | 0.014 |
    | Hôpital                  | 8312022130   | 'Urg_Hospit' | 2022-02-01 | 0.652 | 0.027 |
    """
    if isinstance(probe, BaseProbe):
        probe.is_computed_probe()
    else:
        raise TypeError(
            "Unsupported type {} for probe.".format(type(probe).__name__)
        )

    predictor = filter_table_by_date(
        table=probe.predictor,
        table_name="predictor",
        start_date=start_date,
        end_date=end_date,
    )
    index = probe._index

    estimates = self.fit_process(
        predictor=predictor,
        index=index,
        **kwargs,
    )

    metrics_df = self._compute_metrics(
        predictor=predictor,
        estimates=estimates,
        index=index,
        metric_functions=metric_functions,
    )

    if metrics_df is not None:
        self._metrics = list(metrics_df.columns.difference(index))
        self.estimates = estimates.merge(metrics_df, on=index)

    else:
        self.estimates = estimates

    self.is_computed_estimates()
    self.params = kwargs
    if with_cache:
        self.cache_estimates()

reset_estimates 

reset_estimates() -> None

Reset the estimates to its initial state

Source code in edsteva/models/base.py 
165
166
167
168
169
def reset_estimates(
    self,
) -> None:
    """Reset the estimates to its initial state"""
    self.estimates = self._cache_estimates.copy()

cache_estimates 

cache_estimates() -> None

Cache the predictor

Source code in edsteva/models/base.py 
171
172
173
174
175
176
177
178
def cache_estimates(
    self,
) -> None:
    """Cache the predictor"""
    self._cache_estimates = self.estimates.copy()
    logger.info(
        "Cache the estimates, you can reset the estimates to this state with the method reset_estimates"
    )

predict 

predict(probe: BaseProbe) -> pd.DataFrame

Computes the predicted probe by using the estimates

PARAMETER DESCRIPTION
probe

Target variable to be predicted

TYPE: BaseProbe

Examples:

from edsteva.models.step_function import StepFunction

step_function_model.predict(visit).head()
care_site_level care_site_id stay_type date n_visit c c_fit
Unité Fonctionnelle (UF) 8312056386 'Urg_Hospit' 2019-05-01 233.0 0.841 0.758
Unité Fonctionnelle (UF) 8312056386 'All' 2021-04-01 393.0 0.640 0.758
Pôle/DMU 8312027648 'Urg_Hospit' 2011-03-01 204.0 0.497 0
Pôle/DMU 8312027648 'All' 2018-08-01 22.0 0.784 0.874
Hôpital 8312022130 'Urg_Hospit' 2022-02-01 9746.0 0.974 0.912
Source code in edsteva/models/base.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
def predict(
    self,
    probe: BaseProbe,
) -> pd.DataFrame:
    """Computes the predicted probe by using the estimates

    Parameters
    ----------
    probe : BaseProbe
        Target variable to be predicted

    Examples
    --------
    ```python
    from edsteva.models.step_function import StepFunction

    step_function_model.predict(visit).head()
    ```

    | care_site_level          | care_site_id | stay_type    | date       | n_visit | c     | c_fit |
    | :----------------------- | :----------- | :----------- | :--------- | :------ | :---- | :---- |
    | Unité Fonctionnelle (UF) | 8312056386   | 'Urg_Hospit' | 2019-05-01 | 233.0   | 0.841 | 0.758 |
    | Unité Fonctionnelle (UF) | 8312056386   | 'All'        | 2021-04-01 | 393.0   | 0.640 | 0.758 |
    | Pôle/DMU                 | 8312027648   | 'Urg_Hospit' | 2011-03-01 | 204.0   | 0.497 | 0     |
    | Pôle/DMU                 | 8312027648   | 'All'        | 2018-08-01 | 22.0    | 0.784 | 0.874 |
    | Hôpital                  | 8312022130   | 'Urg_Hospit' | 2022-02-01 | 9746.0  | 0.974 | 0.912 |

    """

    predictor = probe.predictor
    index = probe._index

    return self.predict_process(predictor=predictor, index=index)

load 

load(path = None) -> None

Loads a Model from local

PARAMETER DESCRIPTION
path

EXAMPLE: "my_folder/my_file.html"

TYPE: str DEFAULT: None

Examples:

from edsteva.probes import VisitProbe

probe_path = "my_path/visit.pkl"

visit = VisitProbe()
visit.load(path=probe_path)
Source code in edsteva/models/base.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
def load(self, path=None) -> None:
    """Loads a Model from local

    Parameters
    ----------
    path : str, optional
        **EXAMPLE**: `"my_folder/my_file.html"`

    Examples
    -------
    ```python
    from edsteva.probes import VisitProbe

    probe_path = "my_path/visit.pkl"

    visit = VisitProbe()
    visit.load(path=probe_path)
    ```

    """

    path = path or self._get_path()
    loaded_model = load_object(path)
    self.__dict__ = loaded_model.__dict__.copy()
    self.path = path

save 

save(path: str = None, name: str = None) -> bool

Saves computed Model instance

PARAMETER DESCRIPTION
path

EXAMPLE: "my_folder/my_file.html"

TYPE: str DEFAULT: None

name

EXAMPLE: "fitted_visit"

TYPE: str DEFAULT: None

Examples:

from edsteva.probes import VisitProbe

probe_path = "my_path/visit.pkl"

visit = VisitProbe()
visit.compute(data)
visit.save(path=probe_path)
Source code in edsteva/models/base.py 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
def save(self, path: str = None, name: str = None) -> bool:
    """Saves computed Model instance

    Parameters
    ----------
    path : str, optional
        **EXAMPLE**: `"my_folder/my_file.html"`
    name : str, optional
        **EXAMPLE**: `"fitted_visit"`

    Examples
    -------
    ```python
    from edsteva.probes import VisitProbe

    probe_path = "my_path/visit.pkl"

    visit = VisitProbe()
    visit.compute(data)
    visit.save(path=probe_path)
    ```

    """

    self.is_computed_estimates()

    if name:
        self.name = name
    if not path:
        path = self._get_path()

    self.path = path
    save_object(self, path)

delete 

delete(path: str = None) -> bool

Delete the saved Model instance

PARAMETER DESCRIPTION
path

EXAMPLE: "my_folder/my_file.html"

TYPE: str DEFAULT: None

Source code in edsteva/models/base.py 
274
275
276
277
278
279
280
281
282
283
284
285
def delete(self, path: str = None) -> bool:
    """Delete the saved Model instance

    Parameters
    ----------
    path : str, optional
        **EXAMPLE**: `"my_folder/my_file.html"`
    """
    if not path:
        path = self.path

    delete_object(self, path)

is_predictable_probe 

is_predictable_probe(
    predictor: pd.DataFrame, index: List[str]
) -> pd.DataFrame

Raises an error if the model has not been fitted on the input predictor.

PARAMETER DESCRIPTION
predictor

Target DataFrame to be predicted

TYPE: pd.DataFrame

index

List of the columns given by Probe._index

TYPE: List[str]

RETURNS DESCRIPTION
pd.DataFrame

Predictor along with the fitted estimates
RAISES DESCRIPTION
Exception

Some indexes have no associated estimates, the model must be fitted on an adequate probe
Source code in edsteva/models/base.py 
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
def is_predictable_probe(
    self,
    predictor: pd.DataFrame,
    index: List[str],
) -> pd.DataFrame:
    """Raises an error if the model has not been fitted on the input predictor.

    Parameters
    ----------
    predictor : pd.DataFrame
        Target DataFrame to be predicted
    index : List[str]
        List of the columns given by Probe._index

    Returns
    -------
    pd.DataFrame
        Predictor along with the fitted estimates

    Raises
    ------
    Exception
        Some indexes have no associated estimates, the model must be fitted on an adequate probe
    """
    prediction = predictor.merge(
        self.estimates, on=index, how="left", validate="many_to_one", indicator=True
    )
    if (prediction["_merge"] == "both").all():
        return prediction.drop(columns="_merge")

    raise Exception(
        "Some indexes have no associated estimates, the model must be fitted on an adequate probe"
    )