processors.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318
  1. # Copyright (c) 2024 PaddlePaddle Authors. All Rights Reserved.
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License");
  4. # you may not use this file except in compliance with the License.
  5. # You may obtain a copy of the License at
  6. #
  7. # http://www.apache.org/licenses/LICENSE-2.0
  8. #
  9. # Unless required by applicable law or agreed to in writing, software
  10. # distributed under the License is distributed on an "AS IS" BASIS,
  11. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. # See the License for the specific language governing permissions and
  13. # limitations under the License.
  14. from typing import Any, Dict, List
  15. import joblib
  16. import numpy as np
  17. import pandas as pd
  18. from ....utils.benchmark import benchmark
  19. from .funcs import load_from_dataframe, time_feature
  20. __all__ = [
  21. "BuildTSDataset",
  22. "TSCutOff",
  23. "TSNormalize",
  24. "TimeFeature",
  25. "TStoArray",
  26. "TStoBatch",
  27. ]
  28. @benchmark.timeit
  29. class TSCutOff:
  30. """Truncates time series data to a specified length for training.
  31. This class provides a method to truncate or cut off time series data
  32. to a specified input length, optionally skipping some initial data
  33. points. This is useful for preparing data for training models that
  34. require a fixed input size.
  35. """
  36. def __init__(self, size: Dict[str, int]):
  37. """Initializes the TSCutOff with size configurations.
  38. Args:
  39. size (Dict[str, int]): Dictionary containing size configurations,
  40. including 'in_chunk_len' for the input chunk length and
  41. optionally 'skip_chunk_len' for the number of initial data
  42. points to skip.
  43. """
  44. super().__init__()
  45. self.size = size
  46. def __call__(self, ts_list: List) -> List:
  47. """Applies the cut off operation to a list of time series.
  48. Args:
  49. ts_list (List): List of time series data frames to be truncated.
  50. Returns:
  51. List: List of truncated time series data frames.
  52. """
  53. return [self.cutoff(ts) for ts in ts_list]
  54. def cutoff(self, ts: Any) -> Any:
  55. """Truncates a single time series data frame to the specified length.
  56. This method truncates the time series data to the specified input
  57. chunk length, optionally skipping some initial data points. It raises
  58. a ValueError if the time series is too short.
  59. Args:
  60. ts: A single time series data frame to be truncated.
  61. Returns:
  62. Any: The truncated time series data frame.
  63. Raises:
  64. ValueError: If the time series length is less than the required
  65. minimum length (input chunk length plus any skip chunk length).
  66. """
  67. skip_len = self.size.get("skip_chunk_len", 0)
  68. if len(ts) < self.size["in_chunk_len"] + skip_len:
  69. raise ValueError(
  70. f"The length of the input data is {len(ts)}, but it should be at least {self.size['in_chunk_len'] + self.size['skip_chunk_len']} for training."
  71. )
  72. ts_data = ts[-(self.size["in_chunk_len"] + skip_len) :]
  73. return ts_data
  74. @benchmark.timeit
  75. class TSNormalize:
  76. """Normalizes time series data using a pre-fitted scaler.
  77. This class normalizes specified columns of time series data using a
  78. pre-fitted scaler loaded from a specified path. It supports normalization
  79. of both target and feature columns as specified in the parameters.
  80. """
  81. def __init__(self, scale_path: str, params_info: Dict[str, Any]):
  82. """Initializes the TSNormalize with a scaler and normalization parameters.
  83. Args:
  84. scale_path (str): Path to the pre-fitted scaler object file.
  85. params_info (Dict[str, Any]): Dictionary containing information
  86. about which columns to normalize, including 'target_cols'
  87. and 'feature_cols'.
  88. """
  89. super().__init__()
  90. self.scaler = joblib.load(scale_path)
  91. self.params_info = params_info
  92. def __call__(self, ts_list: List[pd.DataFrame]) -> List[pd.DataFrame]:
  93. """Applies normalization to a list of time series data frames.
  94. Args:
  95. ts_list (List[pd.DataFrame]): List of time series data frames to be normalized.
  96. Returns:
  97. List[pd.DataFrame]: List of normalized time series data frames.
  98. """
  99. return [self.tsnorm(ts) for ts in ts_list]
  100. def tsnorm(self, ts: pd.DataFrame) -> pd.DataFrame:
  101. """Normalizes specified columns of a single time series data frame.
  102. This method applies the scaler to normalize the specified target
  103. and feature columns of the time series.
  104. Args:
  105. ts (pd.DataFrame): A single time series data frame to be normalized.
  106. Returns:
  107. pd.DataFrame: The normalized time series data frame.
  108. """
  109. if self.params_info.get("target_cols", None) is not None:
  110. ts[self.params_info["target_cols"]] = self.scaler.transform(
  111. ts[self.params_info["target_cols"]]
  112. )
  113. if self.params_info.get("feature_cols", None) is not None:
  114. ts[self.params_info["feature_cols"]] = self.scaler.transform(
  115. ts[self.params_info["feature_cols"]]
  116. )
  117. return ts
  118. @benchmark.timeit
  119. class BuildTSDataset:
  120. """Constructs a time series dataset from a list of time series data frames."""
  121. def __init__(self, params_info: Dict[str, Any]):
  122. """Initializes the BuildTSDataset with parameters for dataset construction.
  123. Args:
  124. params_info (Dict[str, Any]): Dictionary containing parameters for
  125. constructing the time series dataset.
  126. """
  127. super().__init__()
  128. self.params_info = params_info
  129. def __call__(self, ts_list: List) -> List:
  130. """Applies the dataset construction to a list of time series.
  131. Args:
  132. ts_list (List): List of time series data frames.
  133. Returns:
  134. List: List of constructed time series datasets.
  135. """
  136. return [self.buildtsdata(ts) for ts in ts_list]
  137. def buildtsdata(self, ts) -> Any:
  138. """Builds a time series dataset from a single time series data frame.
  139. Args:
  140. ts: A single time series data frame.
  141. Returns:
  142. Any: A constructed time series dataset.
  143. """
  144. ts_data = load_from_dataframe(ts, **self.params_info)
  145. return ts_data
  146. @benchmark.timeit
  147. class TimeFeature:
  148. """Extracts time features from time series data for forecasting."""
  149. def __init__(
  150. self, params_info: Dict[str, Any], size: Dict[str, int], holiday: bool = False
  151. ):
  152. """Initializes the TimeFeature extractor.
  153. Args:
  154. params_info (Dict[str, Any]): Dictionary containing frequency information.
  155. size (Dict[str, int]): Dictionary containing the output chunk length.
  156. holiday (bool, optional): Whether to include holiday features. Defaults to False.
  157. """
  158. super().__init__()
  159. self.freq = params_info["freq"]
  160. self.size = size
  161. self.holiday = holiday
  162. def __call__(self, ts_list: List) -> List:
  163. """Applies time feature extraction to a list of time series.
  164. Args:
  165. ts_list (List): List of time series data frames.
  166. Returns:
  167. List: List of time series with extracted time features.
  168. """
  169. return [self.timefeat(ts) for ts in ts_list]
  170. def timefeat(self, ts: Dict[str, Any]) -> Any:
  171. """Extracts time features from a single time series data frame.
  172. Args:
  173. ts: A single time series data frame.
  174. Returns:
  175. Any: The time series with added time features.
  176. """
  177. if not self.holiday:
  178. ts = time_feature(
  179. ts,
  180. self.freq,
  181. ["hourofday", "dayofmonth", "dayofweek", "dayofyear"],
  182. self.size["out_chunk_len"],
  183. )
  184. else:
  185. ts = time_feature(
  186. ts,
  187. self.freq,
  188. [
  189. "minuteofhour",
  190. "hourofday",
  191. "dayofmonth",
  192. "dayofweek",
  193. "dayofyear",
  194. "monthofyear",
  195. "weekofyear",
  196. "holidays",
  197. ],
  198. self.size["out_chunk_len"],
  199. )
  200. return ts
  201. @benchmark.timeit
  202. class TStoArray:
  203. """Converts time series data into arrays for model input."""
  204. def __init__(self, input_data: Dict[str, Any]):
  205. """Initializes the TStoArray converter.
  206. Args:
  207. input_data (Dict[str, Any]): Dictionary specifying the input data format.
  208. """
  209. super().__init__()
  210. self.input_data = input_data
  211. def __call__(self, ts_list: List[Dict[str, Any]]) -> List[List[np.ndarray]]:
  212. """Converts a list of time series data frames into arrays.
  213. Args:
  214. ts_list (List[Dict[str, Any]]): List of time series data frames.
  215. Returns:
  216. List[List[np.ndarray]]: List of lists of arrays for each time series.
  217. """
  218. return [self.tstoarray(ts) for ts in ts_list]
  219. def tstoarray(self, ts: Dict[str, Any]) -> List[np.ndarray]:
  220. """Converts a single time series data frame into arrays.
  221. Args:
  222. ts (Dict[str, Any]): A single time series data frame.
  223. Returns:
  224. List[np.ndarray]: List of arrays representing the time series data.
  225. """
  226. ts_list = []
  227. input_name = list(self.input_data.keys())
  228. input_name.sort()
  229. for key in input_name:
  230. ts_list.append(np.array(ts[key]).astype("float32"))
  231. return ts_list
  232. @benchmark.timeit
  233. class TStoBatch:
  234. """Convert a list of time series into batches for processing.
  235. This class provides a method to convert a list of time series data into
  236. batches. Each time series in the list is assumed to be a sequence of
  237. equal-length arrays or DataFrames.
  238. """
  239. def __call__(self, ts_list: List[np.ndarray]) -> List[np.ndarray]:
  240. """Convert a list of time series into batches.
  241. This method stacks time series data along a new axis to create batches.
  242. It assumes that each time series in the list has the same length.
  243. Args:
  244. ts_list (List[np.ndarray]): A list of time series, where each
  245. time series is represented as a list or array of equal length.
  246. Returns:
  247. List[np.ndarray]: A list of batches, where each batch is a stacked
  248. array of time series data at the same index across all series.
  249. """
  250. n = len(ts_list[0])
  251. return [np.stack([ts[i] for ts in ts_list], axis=0) for i in range(n)]