Add timeseries label processors and normalization options

docs/api/processors.rst

@@ -35,6 +35,8 @@ Available Processors
 - ``MultiClassLabelProcessor``: Multi-class classification labels
 - ``MultiLabelProcessor``: Multi-label classification (e.g., drug recommendation)
 - ``RegressionLabelProcessor``: Continuous regression targets
+- ``TimeseriesBinaryLabelProcessor``: Binary label per time step (``"timeseries_binary"``), returned as ``(value, mask)`` on the time grid of the input time series
+- ``TimeseriesRegressionLabelProcessor``: Real-valued label per time step (``"timeseries_regression"``)
 
 **Data Type Processors:**
 
@@ -43,9 +45,9 @@ Available Processors
 - ``TupleTimeTextProcessor``: For text paired with temporal information (time-aware text)
 - ``AudioProcessor``: For audio signal data
 - ``SignalProcessor``: For general signal data (e.g., EEG, ECG)
-- ``TimeseriesProcessor``: For time-series data
+- ``TimeseriesProcessor``: For time-series data; optional ``normalization`` (``"zscore"`` or ``"minmax"``, fitted on the training split) and ``missing_indicators``
 - ``TimeImageProcessor``: For time-stamped image sequences (e.g., serial X-rays)
-- ``TensorProcessor``: For pre-processed tensor data
+- ``TensorProcessor``: For pre-processed tensor data; optional ``normalization``, ``missing_indicators`` and ``impute_strategy="zero"``
 - ``RawProcessor``: Pass-through processor for raw data
 
 **Specialized Processors:**
@@ -548,6 +550,9 @@ API Reference
     processors/pyhealth.processors.MultiClassLabelProcessor
     processors/pyhealth.processors.MultiLabelProcessor
     processors/pyhealth.processors.RegressionLabelProcessor
+    processors/pyhealth.processors.TimeseriesLabelProcessor
+    processors/pyhealth.processors.TimeseriesBinaryLabelProcessor
+    processors/pyhealth.processors.TimeseriesRegressionLabelProcessor
     processors/pyhealth.processors.ImageProcessor
     processors/pyhealth.processors.TextProcessor
     processors/pyhealth.processors.TupleTimeTextProcessor
@@ -562,4 +567,5 @@ API Reference
     processors/pyhealth.processors.MultiHotProcessor
     processors/pyhealth.processors.StageNetProcessor
     processors/pyhealth.processors.StageNetTensorProcessor
-    processors/pyhealth.processors.GraphProcessor
+    processors/pyhealth.processors.GraphProcessor
+    processors/pyhealth.processors.utils

docs/api/processors/pyhealth.processors.TimeseriesBinaryLabelProcessor.rst

@@ -0,0 +1,9 @@
+pyhealth.processors.TimeseriesBinaryLabelProcessor
+==================================================
+
+Binary (0/1) label per time step (``"timeseries_binary"``).
+
+.. autoclass:: pyhealth.processors.TimeseriesBinaryLabelProcessor
+    :members:
+    :undoc-members:
+    :show-inheritance:

docs/api/processors/pyhealth.processors.TimeseriesLabelProcessor.rst

@@ -0,0 +1,9 @@
+pyhealth.processors.TimeseriesLabelProcessor
+============================================
+
+Base processor for labels given per time step, returning ``(value, mask)`` on the time grid of the input time series.
+
+.. autoclass:: pyhealth.processors.TimeseriesLabelProcessor
+    :members:
+    :undoc-members:
+    :show-inheritance:

docs/api/processors/pyhealth.processors.TimeseriesRegressionLabelProcessor.rst

@@ -0,0 +1,9 @@
+pyhealth.processors.TimeseriesRegressionLabelProcessor
+======================================================
+
+Real-valued label per time step (``"timeseries_regression"``).
+
+.. autoclass:: pyhealth.processors.TimeseriesRegressionLabelProcessor
+    :members:
+    :undoc-members:
+    :show-inheritance:

docs/api/processors/pyhealth.processors.utils.rst

@@ -0,0 +1,8 @@
+pyhealth.processors.utils
+=========================
+
+Helpers shared by processors: learning normalization statistics and placing
+values on a uniform time grid.
+
+.. automodule:: pyhealth.processors.utils
+    :members:

pyhealth/processors/__init__.py

@@ -44,7 +44,12 @@ def get_processor(name: str):
 )
 from .tensor_processor import TensorProcessor
 from .text_processor import TextProcessor
-from .timeseries_processor import TimeseriesProcessor
+from .timeseries_processor import (
+    TimeseriesBinaryLabelProcessor,
+    TimeseriesLabelProcessor,
+    TimeseriesProcessor,
+    TimeseriesRegressionLabelProcessor,
+)
 from .time_image_processor import TimeImageProcessor
 from .graph_processor import GraphProcessor
 from .audio_processor import AudioProcessor
@@ -77,7 +82,10 @@ def get_processor(name: str):
     "TensorProcessor",
     "TemporalTimeseriesProcessor",
     "TextProcessor",
+    "TimeseriesBinaryLabelProcessor",
+    "TimeseriesLabelProcessor",
     "TimeseriesProcessor",
+    "TimeseriesRegressionLabelProcessor",
     "TimeImageProcessor",
     "GraphProcessor",
     "AudioProcessor",

pyhealth/processors/temporal_timeseries_processor.py

@@ -16,6 +16,7 @@
 
 from . import register_processor
 from .base_processor import ModalityType, TemporalFeatureProcessor
+from .utils import sample_on_grid
 
 
 @register_processor("temporal_timeseries")
@@ -39,15 +40,18 @@ class TemporalTimeseriesProcessor(TemporalFeatureProcessor):
         sampling_rate: Uniform re-sampling interval.  Defaults to 1 hour.
         impute_strategy: Currently only ``"forward_fill"`` is supported.
 
-    Example::
-
-        proc = TemporalTimeseriesProcessor(sampling_rate=timedelta(hours=2))
-        from datetime import datetime, timedelta
-        ts  = [datetime(2023,1,1,0), datetime(2023,1,1,4), datetime(2023,1,1,8)]
-        val = np.array([[120.0, 80.0], [115.0, 78.0], [118.0, 82.0]])
-        out = proc.process_temporal((ts, val))
-        # out["value"].shape  → (5, 2)   ← 5 two-hour steps over 8 h
-        # out["time"].shape   → (5,)     ← [0., 2., 4., 6., 8.] hours
+    Examples:
+        >>> from datetime import datetime, timedelta
+        >>> import numpy as np
+        >>> from pyhealth.processors import TemporalTimeseriesProcessor
+        >>> proc = TemporalTimeseriesProcessor(sampling_rate=timedelta(hours=2))
+        >>> ts = [datetime(2023, 1, 1, 0), datetime(2023, 1, 1, 4), datetime(2023, 1, 1, 8)]
+        >>> val = np.array([[120.0, 80.0], [115.0, 78.0], [118.0, 82.0]])
+        >>> out = proc.process_temporal((ts, val))
+        >>> tuple(out["value"].shape)  # 5 two-hour steps over 8 hours
+        (5, 2)
+        >>> out["time"].tolist()
+        [0.0, 2.0, 4.0, 6.0, 8.0]
     """
 
     def __init__(
@@ -89,21 +93,8 @@ def process(self, value: Tuple[List[datetime], np.ndarray]) -> dict:
         if len(timestamps) == 0:
             raise ValueError("Timestamps list is empty.")
 
-        values = np.asarray(values, dtype=float)
-        if values.ndim == 1:
-            values = values[:, None]  # (T,) → (T, 1)
-
-        num_features = values.shape[1]
-        start_time = timestamps[0]
-        end_time = timestamps[-1]
-        total_steps = int((end_time - start_time) / self.sampling_rate) + 1
-
-        sampled_values = np.full((total_steps, num_features), np.nan)
-
-        for t, v in zip(timestamps, values):
-            idx = int((t - start_time) / self.sampling_rate)
-            if 0 <= idx < total_steps:
-                sampled_values[idx] = v
+        sampled_values = sample_on_grid(timestamps, values, self.sampling_rate)
+        total_steps, num_features = sampled_values.shape
 
         # Forward-fill imputation
         for f in range(num_features):

pyhealth/processors/tensor_processor.py

@@ -1,9 +1,11 @@
-from typing import Any, Dict, Iterable, Optional
+from collections.abc import Iterable
+from typing import Any, Literal
 
 import torch
 
 from . import register_processor
 from .base_processor import FeatureProcessor
+from .utils import NORMALIZATIONS, Normalization, fit_normalization
 
 
 @register_processor("tensor")
@@ -16,6 +18,11 @@ class TensorProcessor(FeatureProcessor):
 
     Processing:
         - Convert input directly to torch.Tensor using torch.tensor()
+        - Optionally, along the last dimension: normalization with statistics
+          learned in ``fit()`` from the observed (non-NaN) values,
+          missing-value indicators (1 where the value is NaN) appended, and
+          NaN filled with 0, in that order. After ``"zscore"``
+          normalization, 0 is the training mean.
 
     Output:
         - torch.Tensor with appropriate shape and dtype
@@ -31,11 +38,18 @@ class TensorProcessor(FeatureProcessor):
     """
 
     stores_tensor = True
+    # Defaults for processors pickled before these options existed.
+    normalization: Normalization = None
+    missing_indicators: bool = False
+    impute_strategy: Literal["zero"] | None = None
 
     def __init__(
         self,
         dtype: torch.dtype = torch.float32,
-        spatial_dims: Optional[tuple[bool, ...]] = None,
+        spatial_dims: tuple[bool, ...] | None = None,
+        normalization: Normalization = None,
+        missing_indicators: bool = False,
+        impute_strategy: Literal["zero"] | None = None,
     ):
         """
         Initialize the TensorProcessor.
@@ -45,13 +59,29 @@ def __init__(
                   Default is torch.float32.
             spatial_dims: Tuple of booleans indicating which dimensions are spatial.
                   If None, defaults to all False. Default is None.
+            normalization: ``None``, ``"zscore"`` (zero mean, unit variance)
+                  or ``"minmax"`` (scaled to [0, 1] on the fitted range),
+                  per entry of the last dimension. Default is None.
+            missing_indicators: Whether to append a missing-value indicator
+                  per entry of the last dimension. Default is False.
+            impute_strategy: ``None`` (keep NaN) or ``"zero"``. Default is None.
         """
+        if normalization not in NORMALIZATIONS:
+            raise ValueError(f"normalization must be one of {NORMALIZATIONS}")
+        if impute_strategy not in (None, "zero"):
+            raise ValueError("impute_strategy must be None or 'zero'")
         self.dtype = dtype
         self._n_dim = None
         self._feature_dim = None
         self._spatial_dims = spatial_dims
-
-    def fit(self, samples: Iterable[Dict[str, Any]], field: str) -> None:
+        self.normalization = normalization
+        self.missing_indicators = missing_indicators
+        self.impute_strategy = impute_strategy
+        self.learns_statistics = normalization is not None
+        self.center: list[float] = []
+        self.scale: list[float] = []
+
+    def fit(self, samples: Iterable[dict[str, Any]], field: str) -> None:
         """Infer n_dim and the feature width from the first valid sample.
 
         Args:
@@ -70,6 +100,18 @@ def fit(self, samples: Iterable[Dict[str, Any]], field: str) -> None:
                 self._feature_dim = tensor.shape[-1] if tensor.dim() > 0 else 1
                 break
 
+        if self.normalization is not None:
+            self.center, self.scale = fit_normalization(
+                (
+                    torch.as_tensor(sample[field], dtype=torch.float64)
+                    .reshape(-1, self._feature_dim)
+                    .numpy()
+                    for sample in samples
+                    if sample.get(field) is not None
+                ),
+                self.normalization,
+            )
+
     def process(self, value: Any) -> torch.Tensor:
         """
         Process a numerical value or list into a torch.Tensor.
@@ -84,18 +126,38 @@ def process(self, value: Any) -> torch.Tensor:
         # which can trigger a UserWarning. If value is already a tensor,
         # return a detached clone cast to the requested dtype.
         if isinstance(value, torch.Tensor):
-            return value.detach().clone().to(dtype=self.dtype)
-        return torch.tensor(value, dtype=self.dtype)
+            tensor = value.detach().clone().to(dtype=self.dtype)
+        else:
+            tensor = torch.tensor(value, dtype=self.dtype)
+
+        if self.normalization is not None:
+            # Shaped like the last dimension, so a scalar input stays a scalar.
+            shape = tensor.shape[-1:]
+            center = torch.tensor(self.center, dtype=self.dtype).reshape(shape)
+            scale = torch.tensor(self.scale, dtype=self.dtype).reshape(shape)
+            tensor = (tensor - center) / scale
+        if self.missing_indicators:
+            if tensor.dim() == 0:
+                tensor = tensor.reshape(1)
+            indicators = torch.isnan(tensor).to(self.dtype)
+        if self.impute_strategy == "zero":
+            tensor = torch.nan_to_num(tensor, nan=0.0)
+        if self.missing_indicators:
+            tensor = torch.cat([tensor, indicators], dim=-1)
+        return tensor
 
     def size(self) -> int | None:
         """
         Get the feature size of the processor.
 
         Returns:
             The width of the last dimension (1 for scalars) seen in ``fit()``,
-            or None before fitting.
+            doubled with missing-value indicators, or None before fitting.
         """
-        return getattr(self, "_feature_dim", None)
+        feature_dim = getattr(self, "_feature_dim", None)
+        if feature_dim is None or not self.missing_indicators:
+            return feature_dim
+        return feature_dim * 2
 
     def is_token(self) -> bool:
         """Whether the output tensor represents discrete token indices, inferred from dtype.
@@ -112,7 +174,8 @@ def dim(self) -> tuple[int, ...]:
         """Number of dimensions for the output tensor.
 
         Returns:
-            (n_dim,)
+            (n_dim,). With missing-value indicators, a scalar input becomes a
+            vector of 2, so n_dim is at least 1.
 
         Raises:
             NotImplementedError: If n_dim was not provided and fit() was not called.
@@ -122,7 +185,12 @@ def dim(self) -> tuple[int, ...]:
                 "TensorProcessor cannot determine n_dim automatically. "
                 "Call fit() first."
             )
-        return (self._n_dim,)
+        return (self._output_n_dim(),)
+
+    def _output_n_dim(self) -> int:
+        if self.missing_indicators:
+            return max(self._n_dim, 1)
+        return self._n_dim
 
     def spatial(self) -> tuple[bool, ...]:
         """Whether each dimension of the output tensor is spatial.
@@ -137,7 +205,7 @@ def spatial(self) -> tuple[bool, ...]:
                 "TensorProcessor cannot determine spatial dims. "
                 "Call fit() first."
             )
-        return tuple(False for _ in range(self._n_dim))
+        return tuple(False for _ in range(self._output_n_dim()))
 
     def __repr__(self) -> str:
         """