6941005590

Committed 21 Nov 2023 08:13AM UTC coverage: 83.443% (+0.04%) from 83.403%

Build # 6941005590

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #240 from WenjieDu/dev

Adding other options for the first step imputation in LOCF

Run Details

43 of 43 new or added lines in 1 file covered. (100.0%)

5 existing lines in 2 files now uncovered.

4440 of 5321 relevant lines covered (83.44%)

5.0 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.33

/pypots/imputation/locf/model.py

"""
The implementation of LOCF (Last Observed Carried Forward) for the partially-observed time-series imputation task.

"""

# Created by Wenjie Du <wenjay.du@gmail.com>
# License: BSD-3-Clause

import warnings
from typing import Union, Optional

import numpy as np
import torch

from ..base import BaseImputer
from ...utils.logging import logger


class LOCF(BaseImputer):
    """LOCF (Last Observed Carried Forward) imputation method. A naive imputation method that fills missing values
    with the last observed value. Simple but commonly used in practice.

    Parameters
    ----------
    first_step_imputation : str, default='backward'
        With LOCF, the observed values are carried forward to impute the missing ones. But if the first value
        is missing, there is no value to carry forward. This parameter is used to determine the strategy to
        impute the missing values at the beginning of the time-series sequence after LOCF is applied.
        It can be one of ['backward', 'zero', 'mean', 'nan'].
        If 'nan', the missing values at the sequence beginning will be left as NaNs.
        If 'zero', the missing values at the sequence beginning will be imputed with 0.
        If 'backward', the missing values at the beginning of the time-series sequence will be imputed with the
        first observed value in the sequence, i.e. the first observed value will be carried backward to impute
        the missing values at the beginning of the sequence.
        If 'mean', the missing values at the sequence beginning will be imputed with the overall mean
        values of features in the dataset.

    """

    def __init__(
        self,
        first_step_imputation: str = "backward",
        device: Optional[Union[str, torch.device, list]] = None,
    ):
        super().__init__(device=device)
        assert first_step_imputation in ["nan", "zero", "backward", "mean"]
        self.first_step_imputation = first_step_imputation

    def fit(
        self,
        train_set: Union[dict, str],
        val_set: Optional[Union[dict, str]] = None,
        file_type: str = "h5py",
    ) -> None:
        """Train the imputer on the given data.

        Warnings
        --------
        LOCF does not need to run fit().
        Please run func ``predict()`` directly.

        """
        warnings.warn(
            "LOCF (Last Observed Carried Forward) imputation class has no parameter to train. "
            "Please run func impute(X) directly."
        )

    def _locf_numpy(
        self,
        X: np.ndarray,
    ) -> np.ndarray:
        """Numpy implementation of LOCF.

        Parameters
        ----------
        X : np.ndarray,
            Time series containing missing values (NaN) to be imputed.

        Returns
        -------
        X_imputed : array,
            Imputed time series.

        Notes
        -----
        This implementation gets inspired by the question on StackOverflow:
        https://stackoverflow.com/questions/41190852/most-efficient-way-to-forward-fill-nan-values-in-numpy-array

        """
        trans_X = X.transpose((0, 2, 1))
        mask = np.isnan(trans_X)
        n_samples, n_steps, n_features = mask.shape
        idx = np.where(~mask, np.arange(n_features), 0)
        idx = np.maximum.accumulate(idx, axis=2)

        collector = []
        for x, i in zip(trans_X, idx):
            collector.append(x[np.arange(n_steps)[:, None], i])
        X_imputed = np.asarray(collector)
        X_imputed = X_imputed.transpose((0, 2, 1))

        # If there are values still missing, they are missing at the beginning of the time-series sequence.
        if np.isnan(X_imputed).any():
            if self.first_step_imputation == "nan":
                pass
            elif self.first_step_imputation == "zero":
                X_imputed = np.nan_to_num(X_imputed, nan=0)
            elif self.first_step_imputation == "backward":
                X_imputed_transpose = np.copy(X_imputed)
                # imputed by last observation carried backward (LOCB)
                X_imputed_transpose = np.flip(X_imputed_transpose, axis=1)
                X_LOCB = self._locf_numpy(X_imputed_transpose)
                assert not np.isnan(X_LOCB).any(), "NaN still exists in X_LOCB"
                X_imputed = np.flip(X_LOCB, axis=1)
            elif self.first_step_imputation == "mean":
                bz, n_steps, n_features = X_imputed.shape
                X_imputed_squeezed = np.copy(X_imputed).reshape(-1, n_features)
                mean_values = np.nanmean(X_imputed_squeezed, axis=0)
                for i, v in enumerate(mean_values):
                    X_imputed[:, :, i] = np.nan_to_num(X_imputed[:, :, i], nan=v)

        return X_imputed

    def _locf_torch(
        self,
        X: torch.Tensor,
    ) -> torch.Tensor:
        """Torch implementation of LOCF.

        Parameters
        ----------
        X : tensor,
            Time series containing missing values (NaN) to be imputed.

        Returns
        -------
        X_imputed : tensor,
            Imputed time series.
        """
        trans_X = X.permute((0, 2, 1))
        mask = torch.isnan(trans_X)
        n_samples, n_steps, n_features = mask.shape
        idx = torch.where(~mask, torch.arange(n_features, device=mask.device), 0)
        idx = np.maximum.accumulate(idx, axis=2)

        collector = []
        for x, i in zip(trans_X, idx):
            collector.append(x[torch.arange(n_steps)[:, None], i])
        X_imputed = torch.stack(collector)
        X_imputed = X_imputed.permute((0, 2, 1))

        # If there are values still missing, they are missing at the beginning of the time-series sequence.
        if torch.isnan(X_imputed).any():
            if self.first_step_imputation == "nan":
                pass
            elif self.first_step_imputation == "zero":
                X_imputed = torch.nan_to_num(X_imputed, nan=0)
            elif self.first_step_imputation == "backward":
                X_imputed_transpose = X_imputed.clone()
                # imputed by last observation carried backward (LOCB)
                X_imputed_transpose = torch.flip(X_imputed_transpose, dims=[1])
                X_LOCB = self._locf_torch(X_imputed_transpose)
                assert not torch.isnan(X_LOCB).any(), "NaN still exists in X_LOCB"
                X_imputed = torch.flip(X_LOCB, dims=[1])
            elif self.first_step_imputation == "mean":
                bz, n_steps, n_features = X_imputed.shape
                X_imputed_squeezed = X_imputed.clone().reshape(-1, n_features)
                mean_values = torch.nanmean(X_imputed_squeezed, dim=0)
                for i, v in enumerate(mean_values):
                    X_imputed[:, :, i] = torch.nan_to_num(X_imputed[:, :, i], nan=v)

        return X_imputed

    def predict(
        self,
        test_set: Union[dict, str],
        file_type: str = "h5py",
    ) -> dict:
        """Make predictions for the input data with the trained model.

        Parameters
        ----------
        test_set : dict or str
            The dataset for model validating, should be a dictionary including keys as 'X',
            or a path string locating a data file supported by PyPOTS (e.g. h5 file).
            If it is a dict, X should be array-like of shape [n_samples, sequence length (time steps), n_features],
            which is time-series data for validating, can contain missing values, and y should be array-like of shape
            [n_samples], which is classification labels of X.
            If it is a path string, the path should point to a data file, e.g. a h5 file, which contains
            key-value pairs like a dict, and it has to include keys as 'X' and 'y'.

        file_type : str
            The type of the given file if test_set is a path string.

        Returns
        -------
        result_dict: dict
            Prediction results in a Python Dictionary for the given samples.
            It should be a dictionary including keys as 'imputation', 'classification', 'clustering', and 'forecasting'.
            For sure, only the keys that relevant tasks are supported by the model will be returned.
        """
        assert not isinstance(test_set, str)
        X = test_set["X"]

        assert len(X.shape) == 3, (
            f"Input X should have 3 dimensions [n_samples, n_steps, n_features], "
            f"but the actual shape of X: {X.shape}"
        )
        if isinstance(X, list):
            X = np.asarray(X)

        if isinstance(X, np.ndarray):
            imputed_data = self._locf_numpy(X)
        elif isinstance(X, torch.Tensor):
            imputed_data = self._locf_torch(X)
        else:
            raise TypeError(
                "X must be type of list/np.ndarray/torch.Tensor, " f"but got {type(X)}"
            )

        result_dict = {
            "imputation": imputed_data,
        }
        return result_dict

    def impute(
        self,
        X: Union[dict, str],
        file_type="h5py",
    ) -> np.ndarray:
        """Impute missing values in the given data with the trained model.

        Warnings
        --------
        The method impute is deprecated. Please use `predict()` instead.

        Parameters
        ----------
        X :
            The data samples for testing, should be array-like of shape [n_samples, sequence length (time steps),
            n_features], or a path string locating a data file, e.g. h5 file.

        file_type :
            The type of the given file if X is a path string.

        Returns
        -------
        array-like, shape [n_samples, sequence length (time steps), n_features],
            Imputed data.
        """
        logger.warning(
            "🚨DeprecationWarning: The method impute is deprecated. Please use `predict` instead."
        )
        results_dict = self.predict(X, file_type=file_type)
        return results_dict["imputation"]

1	"""	6✔
2	The implementation of LOCF (Last Observed Carried Forward) for the partially-observed time-series imputation task.
3
4	"""
5
6	# Created by Wenjie Du <wenjay.du@gmail.com>
7	# License: BSD-3-Clause
8
9	import warnings	6✔
10	from typing import Union, Optional	6✔
11
12	import numpy as np	6✔
13	import torch	6✔
14
15	from ..base import BaseImputer	6✔
16	from ...utils.logging import logger	6✔
17
18
19	class LOCF(BaseImputer):	6✔
20	"""LOCF (Last Observed Carried Forward) imputation method. A naive imputation method that fills missing values	6✔
21	with the last observed value. Simple but commonly used in practice.
22
23	Parameters
24	----------
25	first_step_imputation : str, default='backward'
26	With LOCF, the observed values are carried forward to impute the missing ones. But if the first value
27	is missing, there is no value to carry forward. This parameter is used to determine the strategy to
28	impute the missing values at the beginning of the time-series sequence after LOCF is applied.
29	It can be one of ['backward', 'zero', 'mean', 'nan'].
30	If 'nan', the missing values at the sequence beginning will be left as NaNs.
31	If 'zero', the missing values at the sequence beginning will be imputed with 0.
32	If 'backward', the missing values at the beginning of the time-series sequence will be imputed with the
33	first observed value in the sequence, i.e. the first observed value will be carried backward to impute
34	the missing values at the beginning of the sequence.
35	If 'mean', the missing values at the sequence beginning will be imputed with the overall mean
36	values of features in the dataset.
37
38	"""
39
40	def __init__(	6✔
41	self,
42	first_step_imputation: str = "backward",
43	device: Optional[Union[str, torch.device, list]] = None,
44	):
45	super().__init__(device=device)	6✔
46	assert first_step_imputation in ["nan", "zero", "backward", "mean"]	6✔
47	self.first_step_imputation = first_step_imputation	6✔
48
49	def fit(	6✔
50	self,
51	train_set: Union[dict, str],
52	val_set: Optional[Union[dict, str]] = None,
53	file_type: str = "h5py",
54	) -> None:
55	"""Train the imputer on the given data.
56
57	Warnings
58	--------
59	LOCF does not need to run fit().
60	Please run func ``predict()`` directly.
61
62	"""
63	warnings.warn(	×
64	"LOCF (Last Observed Carried Forward) imputation class has no parameter to train. "
65	"Please run func impute(X) directly."
66	)
67
68	def _locf_numpy(	6✔
69	self,
70	X: np.ndarray,
71	) -> np.ndarray:
72	"""Numpy implementation of LOCF.
73
74	Parameters
75	----------
76	X : np.ndarray,
77	Time series containing missing values (NaN) to be imputed.
78
79	Returns
80	-------
81	X_imputed : array,
82	Imputed time series.
83
84	Notes
85	-----
86	This implementation gets inspired by the question on StackOverflow:
87	https://stackoverflow.com/questions/41190852/most-efficient-way-to-forward-fill-nan-values-in-numpy-array
88
89	"""
90	trans_X = X.transpose((0, 2, 1))	6✔
91	mask = np.isnan(trans_X)	6✔
92	n_samples, n_steps, n_features = mask.shape	6✔
93	idx = np.where(~mask, np.arange(n_features), 0)	6✔
94	idx = np.maximum.accumulate(idx, axis=2)	6✔
95
96	collector = []	6✔
97	for x, i in zip(trans_X, idx):	6✔
98	collector.append(x[np.arange(n_steps)[:, None], i])	6✔
99	X_imputed = np.asarray(collector)	6✔
100	X_imputed = X_imputed.transpose((0, 2, 1))	6✔
101
102	# If there are values still missing, they are missing at the beginning of the time-series sequence.
103	if np.isnan(X_imputed).any():	6✔
104	if self.first_step_imputation == "nan":	6✔
105	pass	6✔
106	elif self.first_step_imputation == "zero":	6✔
107	X_imputed = np.nan_to_num(X_imputed, nan=0)	6✔
108	elif self.first_step_imputation == "backward":	6✔
109	X_imputed_transpose = np.copy(X_imputed)	6✔
110	# imputed by last observation carried backward (LOCB)
111	X_imputed_transpose = np.flip(X_imputed_transpose, axis=1)	6✔
112	X_LOCB = self._locf_numpy(X_imputed_transpose)	6✔
113	assert not np.isnan(X_LOCB).any(), "NaN still exists in X_LOCB"	6✔
114	X_imputed = np.flip(X_LOCB, axis=1)	6✔
115	elif self.first_step_imputation == "mean":	6✔
116	bz, n_steps, n_features = X_imputed.shape	6✔
117	X_imputed_squeezed = np.copy(X_imputed).reshape(-1, n_features)	6✔
118	mean_values = np.nanmean(X_imputed_squeezed, axis=0)	6✔
119	for i, v in enumerate(mean_values):	6✔
120	X_imputed[:, :, i] = np.nan_to_num(X_imputed[:, :, i], nan=v)	6✔
121
122	return X_imputed	6✔
123
124	def _locf_torch(	6✔
125	self,
126	X: torch.Tensor,
127	) -> torch.Tensor:
128	"""Torch implementation of LOCF.
129
130	Parameters
131	----------
132	X : tensor,
133	Time series containing missing values (NaN) to be imputed.
134
135	Returns
136	-------
137	X_imputed : tensor,
138	Imputed time series.
139	"""
140	trans_X = X.permute((0, 2, 1))	6✔
141	mask = torch.isnan(trans_X)	6✔
142	n_samples, n_steps, n_features = mask.shape	6✔
143	idx = torch.where(~mask, torch.arange(n_features, device=mask.device), 0)	6✔
144	idx = np.maximum.accumulate(idx, axis=2)	6✔
145
146	collector = []	6✔
147	for x, i in zip(trans_X, idx):	6✔
148	collector.append(x[torch.arange(n_steps)[:, None], i])	6✔
149	X_imputed = torch.stack(collector)	6✔
150	X_imputed = X_imputed.permute((0, 2, 1))	6✔
151
152	# If there are values still missing, they are missing at the beginning of the time-series sequence.
153	if torch.isnan(X_imputed).any():	6✔
154	if self.first_step_imputation == "nan":	6✔
155	pass	6✔
156	elif self.first_step_imputation == "zero":	6✔
157	X_imputed = torch.nan_to_num(X_imputed, nan=0)	6✔
158	elif self.first_step_imputation == "backward":	6✔
159	X_imputed_transpose = X_imputed.clone()	6✔
160	# imputed by last observation carried backward (LOCB)
161	X_imputed_transpose = torch.flip(X_imputed_transpose, dims=[1])	6✔
162	X_LOCB = self._locf_torch(X_imputed_transpose)	6✔
163	assert not torch.isnan(X_LOCB).any(), "NaN still exists in X_LOCB"	6✔
164	X_imputed = torch.flip(X_LOCB, dims=[1])	6✔
165	elif self.first_step_imputation == "mean":	6✔
166	bz, n_steps, n_features = X_imputed.shape	6✔
167	X_imputed_squeezed = X_imputed.clone().reshape(-1, n_features)	6✔
168	mean_values = torch.nanmean(X_imputed_squeezed, dim=0)	6✔
169	for i, v in enumerate(mean_values):	6✔
170	X_imputed[:, :, i] = torch.nan_to_num(X_imputed[:, :, i], nan=v)	6✔
171
172	return X_imputed	6✔
173
174	def predict(	6✔
175	self,
176	test_set: Union[dict, str],
177	file_type: str = "h5py",
178	) -> dict:
179	"""Make predictions for the input data with the trained model.
180
181	Parameters
182	----------
183	test_set : dict or str
184	The dataset for model validating, should be a dictionary including keys as 'X',
185	or a path string locating a data file supported by PyPOTS (e.g. h5 file).
186	If it is a dict, X should be array-like of shape [n_samples, sequence length (time steps), n_features],
187	which is time-series data for validating, can contain missing values, and y should be array-like of shape
188	[n_samples], which is classification labels of X.
189	If it is a path string, the path should point to a data file, e.g. a h5 file, which contains
190	key-value pairs like a dict, and it has to include keys as 'X' and 'y'.
191
192	file_type : str
193	The type of the given file if test_set is a path string.
194
195	Returns
196	-------
197	result_dict: dict
198	Prediction results in a Python Dictionary for the given samples.
199	It should be a dictionary including keys as 'imputation', 'classification', 'clustering', and 'forecasting'.
200	For sure, only the keys that relevant tasks are supported by the model will be returned.
201	"""
202	assert not isinstance(test_set, str)	6✔
203	X = test_set["X"]	6✔
204
205	assert len(X.shape) == 3, (	6✔
206	f"Input X should have 3 dimensions [n_samples, n_steps, n_features], "
207	f"but the actual shape of X: {X.shape}"
208	)
209	if isinstance(X, list):	6✔
210	X = np.asarray(X)	×
211
212	if isinstance(X, np.ndarray):	6✔
213	imputed_data = self._locf_numpy(X)	6✔
214	elif isinstance(X, torch.Tensor):	6✔
215	imputed_data = self._locf_torch(X)	6✔
216	else:
217	raise TypeError(	×
218	"X must be type of list/np.ndarray/torch.Tensor, " f"but got {type(X)}"
219	)
220
221	result_dict = {	6✔
222	"imputation": imputed_data,
223	}
224	return result_dict	6✔
225
226	def impute(	6✔
227	self,
228	X: Union[dict, str],
229	file_type="h5py",
230	) -> np.ndarray:
231	"""Impute missing values in the given data with the trained model.
232
233	Warnings
234	--------
235	The method impute is deprecated. Please use `predict()` instead.
236
237	Parameters
238	----------
239	X :
240	The data samples for testing, should be array-like of shape [n_samples, sequence length (time steps),
241	n_features], or a path string locating a data file, e.g. h5 file.
242
243	file_type :
244	The type of the given file if X is a path string.
245
246	Returns
247	-------
248	array-like, shape [n_samples, sequence length (time steps), n_features],
249	Imputed data.
250	"""
UNCOV 251	logger.warning(	×
252	"🚨DeprecationWarning: The method impute is deprecated. Please use `predict` instead."
253	)
UNCOV 254	results_dict = self.predict(X, file_type=file_type)	×
UNCOV 255	return results_dict["imputation"]	×

WenjieDu / PyPOTS / 6941005590

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous