diff --git a/docs/api/processors.rst b/docs/api/processors.rst
index 0243890c7..f4f2996fb 100644
--- a/docs/api/processors.rst
+++ b/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
\ No newline at end of file
+    processors/pyhealth.processors.GraphProcessor
+    processors/pyhealth.processors.utils
\ No newline at end of file
diff --git a/docs/api/processors/pyhealth.processors.TimeseriesBinaryLabelProcessor.rst b/docs/api/processors/pyhealth.processors.TimeseriesBinaryLabelProcessor.rst
new file mode 100644
index 000000000..378971fb3
--- /dev/null
+++ b/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:
diff --git a/docs/api/processors/pyhealth.processors.TimeseriesLabelProcessor.rst b/docs/api/processors/pyhealth.processors.TimeseriesLabelProcessor.rst
new file mode 100644
index 000000000..009a0850c
--- /dev/null
+++ b/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:
diff --git a/docs/api/processors/pyhealth.processors.TimeseriesRegressionLabelProcessor.rst b/docs/api/processors/pyhealth.processors.TimeseriesRegressionLabelProcessor.rst
new file mode 100644
index 000000000..0aa0cd59a
--- /dev/null
+++ b/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:
diff --git a/docs/api/processors/pyhealth.processors.utils.rst b/docs/api/processors/pyhealth.processors.utils.rst
new file mode 100644
index 000000000..8045fe3da
--- /dev/null
+++ b/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:
diff --git a/pyhealth/processors/__init__.py b/pyhealth/processors/__init__.py
index 5fd0a12ba..b0738b05a 100644
--- a/pyhealth/processors/__init__.py
+++ b/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",
diff --git a/pyhealth/processors/temporal_timeseries_processor.py b/pyhealth/processors/temporal_timeseries_processor.py
index b38a84295..cbd3626c9 100644
--- a/pyhealth/processors/temporal_timeseries_processor.py
+++ b/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):
diff --git a/pyhealth/processors/tensor_processor.py b/pyhealth/processors/tensor_processor.py
index 3c15cffe7..ad036029a 100644
--- a/pyhealth/processors/tensor_processor.py
+++ b/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,8 +126,25 @@ 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:
         """
@@ -93,9 +152,12 @@ def size(self) -> int | None:
 
         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:
         """
diff --git a/pyhealth/processors/timeseries_processor.py b/pyhealth/processors/timeseries_processor.py
index 3d1e253a4..ef5a34ddd 100644
--- a/pyhealth/processors/timeseries_processor.py
+++ b/pyhealth/processors/timeseries_processor.py
@@ -6,6 +6,13 @@
 
 from . import register_processor
 from .base_processor import FeatureProcessor
+from .utils import (
+    NORMALIZATIONS,
+    Normalization,
+    fit_normalization,
+    grid_steps,
+    sample_on_grid,
+)
 
 
 @register_processor("timeseries")
@@ -19,10 +26,27 @@ class TimeseriesProcessor(FeatureProcessor):
 
     Processing:
         1. Uniform sampling at fixed intervals.
-        2. Imputation for missing values.
+        2. Optional normalization, with statistics learned in ``fit()`` from
+           the observed (non-missing) sampled values.
+        3. Optional missing-value indicators, one per feature, 1 where the
+           value was missing before imputation.
+        4. Imputation for missing values.
+
+    With ``normalization="zscore"`` and ``impute_strategy="forward_fill"``,
+    values missing before the first observation are filled with 0, which is
+    the training mean after normalization.
 
     Output:
-        - torch.Tensor of shape (S, F), where S is the number of sampled time steps.
+        - torch.Tensor of shape (S, F), or (S, 2F) with missing-value
+          indicators appended, where S is the number of sampled time steps.
+
+    Args:
+        sampling_rate: Interval of the sampled time steps.
+        impute_strategy: ``"forward_fill"`` (values before the first
+            observation are filled with 0) or ``"zero"``.
+        normalization: ``None``, ``"zscore"`` (zero mean, unit variance) or
+            ``"minmax"`` (scaled to [0, 1] on the fitted range).
+        missing_indicators: Whether to append the missing-value indicators.
 
     Examples:
         >>> from datetime import datetime, timedelta
@@ -36,19 +60,34 @@ class TimeseriesProcessor(FeatureProcessor):
     """
 
     stores_tensor = True
+    # Defaults for processors pickled before these options existed.
+    normalization: Normalization = None
+    missing_indicators: bool = False
 
     def __init__(
         self,
         sampling_rate: timedelta = timedelta(hours=1),
         impute_strategy: str = "forward_fill",
+        normalization: Normalization = None,
+        missing_indicators: bool = False,
     ):
+        if normalization not in NORMALIZATIONS:
+            raise ValueError(f"normalization must be one of {NORMALIZATIONS}")
         # Configurable sampling rate and imputation method
         self.sampling_rate = sampling_rate
         self.impute_strategy = impute_strategy
+        self.normalization = normalization
+        self.missing_indicators = missing_indicators
+        self.learns_statistics = normalization is not None
         self.n_features = None
+        self.center: list[float] = []
+        self.scale: list[float] = []
 
     def fit(self, samples: Any, field: str) -> None:
-        """Fit the processor by determining n_features from the first valid sample.
+        """Fit the processor.
+
+        Determines n_features from the first valid sample and, with
+        ``normalization``, learns the per-feature statistics from all samples.
 
         Args:
             samples: Iterable of sample dictionaries.
@@ -66,32 +105,42 @@ def fit(self, samples: Any, field: str) -> None:
                     self.n_features = 1
                     break
 
+        if self.normalization is not None:
+            self.center, self.scale = fit_normalization(
+                (
+                    self._sample(*sample[field])
+                    for sample in samples
+                    if sample.get(field) is not None and len(sample[field][0]) > 0
+                ),
+                self.normalization,
+            )
+
+    def _sample(
+        self, timestamps: list[datetime], values: np.ndarray
+    ) -> np.ndarray:
+        return sample_on_grid(timestamps, values, self.sampling_rate)
+
     def process(self, value: Tuple[List[datetime], np.ndarray]) -> torch.Tensor:
         timestamps, values = value
 
         if len(timestamps) == 0:
             raise ValueError("Timestamps list is empty.")
 
-        values = np.asarray(values)
-        num_features = values.shape[1]
-
         # Step 1: Uniform sampling
-        start_time = timestamps[0]
-        end_time = timestamps[-1]
-        total_steps = int((end_time - start_time) / self.sampling_rate) + 1
-
-        sampled_times = [
-            start_time + i * self.sampling_rate for i in range(total_steps)
-        ]
-        sampled_values = np.full((total_steps, num_features), np.nan)
-
-        # Map original timestamps to indices in the sampled grid
-        for t, v in zip(timestamps, values):
-            idx = int((t - start_time) / self.sampling_rate)
-            if 0 <= idx < total_steps:
-                sampled_values[idx] = v
-
-        # Step 2: Imputation
+        sampled_values = self._sample(timestamps, values)
+        total_steps, num_features = sampled_values.shape
+
+        # Step 2: Normalization
+        if self.normalization is not None:
+            sampled_values = (
+                sampled_values - np.asarray(self.center)
+            ) / np.asarray(self.scale)
+
+        # Step 3: Missing-value indicators, before imputation
+        if self.missing_indicators:
+            indicators = np.isnan(sampled_values).astype(float)
+
+        # Step 4: Imputation
         if self.impute_strategy == "forward_fill":
             for f in range(num_features):
                 last_value = 0.0
@@ -108,11 +157,15 @@ def process(self, value: Tuple[List[datetime], np.ndarray]) -> torch.Tensor:
         if self.n_features is None:
             self.n_features = sampled_values.shape[1]
 
+        if self.missing_indicators:
+            sampled_values = np.concatenate([sampled_values, indicators], axis=1)
         return torch.tensor(sampled_values, dtype=torch.float)
 
     def size(self):
         # Size equals number of features, unknown until first process
-        return self.n_features
+        if self.n_features is None:
+            return None
+        return self.n_features * (2 if self.missing_indicators else 1)
 
     def is_token(self) -> bool:
         """Time series values are continuous, not discrete tokens."""
@@ -132,5 +185,128 @@ def spatial(self) -> tuple[bool, ...]:
     def __repr__(self):
         return (
             f"TimeSeriesProcessor(sampling_rate={self.sampling_rate}, "
-            f"impute_strategy='{self.impute_strategy}')"
+            f"impute_strategy='{self.impute_strategy}', "
+            f"normalization={self.normalization!r}, "
+            f"missing_indicators={self.missing_indicators})"
+        )
+
+
+class TimeseriesLabelProcessor(FeatureProcessor):
+    """Base processor for labels given per time step.
+
+    The label is a time series ``(timestamps, values)``, placed on the same
+    kind of uniform grid as :class:`TimeseriesProcessor`: a label at time
+    ``t`` goes to step ``floor((t - t0) / sampling_rate)``, where ``t0`` is the
+    first label timestamp. Steps that no label falls into, or whose label is
+    NaN, are masked out and never imputed. Two labels in the same step raise
+    a ``ValueError`` rather than one silently replacing the other.
+
+    The processor does not see the input time series, so the caller must give
+    the labels the same first timestamp and ``sampling_rate`` as the input for
+    each label to line up with one input step. A label series longer than its
+    input is rejected by the model's loss.
+
+    Output:
+        - ``(value, mask)``: two float tensors of shape (S,), where ``mask``
+          is 1 for steps with a label. Padding added when batching has mask 0.
+
+    The registered names ``"timeseries_binary"`` and
+    ``"timeseries_regression"`` are model modes of their own, so modules that
+    assume one prediction per sample (calibration, prediction sets,
+    interpretability) do not treat these labels as ``"binary"`` or
+    ``"regression"``.
+
+    Args:
+        sampling_rate: Interval of the sampled time steps.
+
+    Examples:
+        >>> from datetime import datetime, timedelta
+        >>> from pyhealth.processors import TimeseriesBinaryLabelProcessor
+        >>> t0 = datetime(2020, 1, 1)
+        >>> times = [t0, t0 + timedelta(hours=2)]
+        >>> TimeseriesBinaryLabelProcessor().process((times, [0, 1]))
+        (tensor([0., 0., 1.]), tensor([1., 0., 1.]))
+    """
+
+    def __init__(self, sampling_rate: timedelta = timedelta(hours=1)):
+        self.sampling_rate = sampling_rate
+
+    def process(
+        self, value: tuple[list[datetime], np.ndarray]
+    ) -> tuple[torch.Tensor, torch.Tensor]:
+        timestamps, labels = value
+        if len(timestamps) == 0:
+            raise ValueError("Timestamps list is empty.")
+        steps, counts = np.unique(
+            grid_steps(timestamps, self.sampling_rate), return_counts=True
+        )
+        if (counts > 1).any():
+            step = int(steps[counts > 1][0])
+            raise ValueError(
+                f"{int(counts.max())} labels fall into time step {step}; "
+                "give at most one label per step of the sampling rate."
+            )
+        sampled = sample_on_grid(timestamps, labels, self.sampling_rate)
+        if sampled.shape[1] != 1:
+            raise ValueError("Expected one label per time step.")
+        sampled = sampled[:, 0]
+        mask = ~np.isnan(sampled)
+        self._check(sampled[mask])
+        return (
+            torch.tensor(np.where(mask, sampled, 0.0), dtype=torch.float),
+            torch.tensor(mask, dtype=torch.float),
         )
+
+    def _check(self, labels: np.ndarray) -> None:
+        pass
+
+    def size(self) -> int:
+        return 1
+
+    def is_token(self) -> bool:
+        return False
+
+    def schema(self) -> tuple[str, ...]:
+        return ("value", "mask")
+
+    def dim(self) -> tuple[int, ...]:
+        return (1, 1)
+
+    def spatial(self) -> tuple[bool, ...]:
+        return (True,)
+
+    def __repr__(self):
+        return f"{type(self).__name__}(sampling_rate={self.sampling_rate})"
+
+
+@register_processor("timeseries_binary")
+class TimeseriesBinaryLabelProcessor(TimeseriesLabelProcessor):
+    """Binary (0/1) label per time step, see :class:`TimeseriesLabelProcessor`.
+
+    Examples:
+        >>> from datetime import datetime, timedelta
+        >>> from pyhealth.processors import TimeseriesBinaryLabelProcessor
+        >>> t0 = datetime(2020, 1, 1)
+        >>> times = [t0 + timedelta(hours=h) for h in range(3)]
+        >>> TimeseriesBinaryLabelProcessor().process((times, [0, float("nan"), 1]))
+        (tensor([0., 0., 1.]), tensor([1., 0., 1.]))
+    """
+
+    def _check(self, labels: np.ndarray) -> None:
+        if not np.isin(labels, (0.0, 1.0)).all():
+            raise ValueError("Binary labels must be 0 or 1.")
+
+
+@register_processor("timeseries_regression")
+class TimeseriesRegressionLabelProcessor(TimeseriesLabelProcessor):
+    """Real-valued label per time step, see :class:`TimeseriesLabelProcessor`.
+
+    Examples:
+        >>> from datetime import datetime, timedelta
+        >>> from pyhealth.processors import TimeseriesRegressionLabelProcessor
+        >>> t0 = datetime(2020, 1, 1)
+        >>> times = [t0 + timedelta(hours=h) for h in range(3)]
+        >>> TimeseriesRegressionLabelProcessor().process((times, [2.5, float("nan"), 0.5]))
+        (tensor([2.5000, 0.0000, 0.5000]), tensor([1., 0., 1.]))
+    """
+
diff --git a/pyhealth/processors/utils.py b/pyhealth/processors/utils.py
new file mode 100644
index 000000000..d1fc2fc2f
--- /dev/null
+++ b/pyhealth/processors/utils.py
@@ -0,0 +1,141 @@
+"""Utilities shared by processors."""
+
+from collections.abc import Iterable
+from datetime import datetime, timedelta
+from typing import Literal
+
+import numpy as np
+
+Normalization = Literal["zscore", "minmax"] | None
+NORMALIZATIONS = (None, "zscore", "minmax")
+
+
+def fit_normalization(
+    arrays: Iterable[np.ndarray], method: Normalization
+) -> tuple[list[float], list[float]]:
+    """Learns per-feature ``(center, scale)`` from the observed values.
+
+    Missing values (``NaN``) are ignored. ``"zscore"`` uses the mean and the
+    population standard deviation, ``"minmax"`` the minimum and the range. A
+    feature that is never observed or constant gets a scale of 1.
+
+    Args:
+        arrays: Arrays of shape ``(N, F)``, streamed one at a time.
+        method: ``"zscore"`` or ``"minmax"``.
+
+    Returns:
+        The center and scale of each feature, as lists of floats.
+
+    Examples:
+        >>> import numpy as np
+        >>> from pyhealth.processors.utils import fit_normalization
+        >>> fit_normalization([np.array([[1.0], [3.0]]), np.array([[np.nan]])], "zscore")
+        ([2.0], [1.0])
+    """
+    count = total_mean = total_m2 = low = high = None
+    for array in arrays:
+        array = np.asarray(array, dtype=np.float64)
+        if count is None:
+            n_features = array.shape[1]
+            count = np.zeros(n_features)
+            total_mean = np.zeros(n_features)
+            total_m2 = np.zeros(n_features)
+            low = np.full(n_features, np.inf)
+            high = np.full(n_features, -np.inf)
+        observed = ~np.isnan(array)
+        n = observed.sum(axis=0)
+        if not n.any():
+            continue
+        with np.errstate(invalid="ignore", divide="ignore"):
+            mean = np.where(n > 0, np.nansum(array, axis=0) / n, 0.0)
+            m2 = np.nansum((array - mean) ** 2, axis=0)
+        # Chan et al. parallel update of the running mean and squared deviations.
+        combined = count + n
+        delta = mean - total_mean
+        with np.errstate(invalid="ignore", divide="ignore"):
+            total_mean = np.where(
+                combined > 0, total_mean + delta * n / combined, 0.0
+            )
+            total_m2 = total_m2 + m2 + np.where(
+                combined > 0, delta**2 * count * n / combined, 0.0
+            )
+        count = combined
+        low = np.fmin(low, np.nanmin(np.where(observed, array, np.inf), axis=0))
+        high = np.fmax(high, np.nanmax(np.where(observed, array, -np.inf), axis=0))
+
+    if count is None:
+        return [], []
+    if method == "zscore":
+        center = total_mean
+        with np.errstate(invalid="ignore", divide="ignore"):
+            scale = np.sqrt(np.where(count > 0, total_m2 / count, 0.0))
+    elif method == "minmax":
+        center = np.where(count > 0, low, 0.0)
+        scale = np.where(count > 0, high - low, 0.0)
+    else:
+        raise ValueError(f"normalization must be one of {NORMALIZATIONS}")
+    scale = np.where(scale > 0, scale, 1.0)
+    return center.tolist(), scale.tolist()
+
+
+def grid_steps(timestamps: list[datetime], sampling_rate: timedelta) -> np.ndarray:
+    """Step of each timestamp on a uniform grid starting at the first one.
+
+    A timestamp ``t`` goes to step ``floor((t - t0) / sampling_rate)``.
+
+    Args:
+        timestamps: Times, the first one starting the grid.
+        sampling_rate: Interval of the grid.
+
+    Returns:
+        Integer array with the step of each timestamp.
+
+    Examples:
+        >>> from datetime import datetime, timedelta
+        >>> from pyhealth.processors.utils import grid_steps
+        >>> t0 = datetime(2020, 1, 1)
+        >>> times = [t0, t0 + timedelta(minutes=90), t0 + timedelta(hours=3)]
+        >>> grid_steps(times, timedelta(hours=1)).tolist()
+        [0, 1, 3]
+    """
+    start_time = timestamps[0]
+    return np.array(
+        [int((t - start_time) / sampling_rate) for t in timestamps], dtype=int
+    )
+
+
+def sample_on_grid(
+    timestamps: list[datetime], values: np.ndarray, sampling_rate: timedelta
+) -> np.ndarray:
+    """Places values on a uniform time grid starting at the first timestamp.
+
+    Each value goes to the step of its timestamp (see :func:`grid_steps`);
+    when several values fall into the same step, the last one is kept.
+
+    Args:
+        timestamps: Time of each row of ``values``.
+        values: Array of shape ``(T,)`` or ``(T, F)``.
+        sampling_rate: Interval of the grid.
+
+    Returns:
+        Array of shape ``(S, F)``, NaN where the grid has no value.
+
+    Examples:
+        >>> from datetime import datetime, timedelta
+        >>> from pyhealth.processors.utils import sample_on_grid
+        >>> t0 = datetime(2020, 1, 1)
+        >>> sample_on_grid([t0, t0 + timedelta(hours=2)], [1.0, 3.0], timedelta(hours=1))
+        array([[ 1.],
+               [nan],
+               [ 3.]])
+    """
+    values = np.asarray(values, dtype=float)
+    if values.ndim == 1:
+        values = values[:, None]
+    steps = grid_steps(timestamps, sampling_rate)
+    total_steps = int((timestamps[-1] - timestamps[0]) / sampling_rate) + 1
+    sampled_values = np.full((total_steps, values.shape[1]), np.nan)
+    for step, value in zip(steps, values):
+        if 0 <= step < total_steps:
+            sampled_values[step] = value
+    return sampled_values
diff --git a/tests/core/test_per_step_labels.py b/tests/core/test_per_step_labels.py
new file mode 100644
index 000000000..4268be3b0
--- /dev/null
+++ b/tests/core/test_per_step_labels.py
@@ -0,0 +1,104 @@
+import unittest
+from datetime import datetime, timedelta
+
+import numpy as np
+import torch
+
+from pyhealth.datasets import create_sample_dataset, get_dataloader
+from pyhealth.models import RNN
+from pyhealth.processors import (
+    TimeseriesBinaryLabelProcessor,
+    TimeseriesRegressionLabelProcessor,
+)
+from pyhealth.trainer import Trainer
+
+T0 = datetime(2020, 1, 1)
+NAN = float("nan")
+
+
+def _times(n):
+    return [T0 + timedelta(hours=h) for h in range(n)]
+
+
+def _samples(label_fn):
+    rng = np.random.default_rng(0)
+    samples = []
+    for i, n in enumerate([5, 3, 6, 4]):
+        samples.append(
+            {
+                "patient_id": f"p{i}",
+                "dynamic": (_times(n), rng.normal(size=(n, 2))),
+                "static": [float(i), 1.0],
+                "label": (_times(n), [label_fn(i, h) for h in range(n)]),
+            }
+        )
+    return samples
+
+
+def _dataset(output="timeseries_binary", label_fn=lambda i, h: float((i + h) % 2)):
+    return create_sample_dataset(
+        samples=_samples(label_fn),
+        input_schema={"dynamic": "timeseries", "static": "tensor"},
+        output_schema={"label": output},
+        dataset_name="per_step",
+    )
+
+
+class TestTimeseriesLabelProcessors(unittest.TestCase):
+    def test_value_and_mask(self):
+        value, mask = TimeseriesBinaryLabelProcessor().process(
+            (_times(4), [0, NAN, 1, 1])
+        )
+        np.testing.assert_array_equal(value.numpy(), [0, 0, 1, 1])
+        np.testing.assert_array_equal(mask.numpy(), [1, 0, 1, 1])
+
+    def test_grid_follows_sampling_rate(self):
+        times = [T0, T0 + timedelta(hours=3)]
+        value, mask = TimeseriesRegressionLabelProcessor().process((times, [2.5, 4.0]))
+        np.testing.assert_array_equal(value.numpy(), [2.5, 0, 0, 4.0])
+        np.testing.assert_array_equal(mask.numpy(), [1, 0, 0, 1])
+
+    def test_binary_rejects_other_values(self):
+        with self.assertRaises(ValueError):
+            TimeseriesBinaryLabelProcessor().process((_times(2), [0, 2]))
+
+    def test_timestamps_between_steps_are_floored(self):
+        times = [T0, T0 + timedelta(minutes=90)]
+        value, mask = TimeseriesRegressionLabelProcessor().process((times, [1.0, 2.0]))
+        np.testing.assert_array_equal(value.numpy(), [1.0, 2.0])
+        np.testing.assert_array_equal(mask.numpy(), [1, 1])
+
+    def test_two_labels_in_one_step_are_rejected(self):
+        times = [T0, T0 + timedelta(minutes=20), T0 + timedelta(hours=1)]
+        with self.assertRaisesRegex(ValueError, "time step 0"):
+            TimeseriesBinaryLabelProcessor().process((times, [0, 1, 0]))
+
+
+class TestPerStepEndToEnd(unittest.TestCase):
+    """The label processors with a model and the Trainer.
+
+    Model-level behaviour (masked loss, causality, bidirectional guard) is
+    covered with a mock label processor in test_timeseries_modes.py.
+    """
+
+    def test_label_processors_train_and_evaluate(self):
+        for mode, metric in (
+            ("timeseries_binary", "roc_auc"),
+            ("timeseries_regression", "mae"),
+        ):
+            dataset = _dataset(mode)
+            model = RNN(dataset=dataset, hidden_dim=8, embedding_dim=8)
+            self.assertEqual(model.mode, mode)
+            batch = next(iter(get_dataloader(dataset, batch_size=4)))
+            out = model(**batch)
+            self.assertEqual(tuple(out["y_prob"].shape), (4, 6, 1))
+            self.assertTrue(torch.isfinite(out["loss"]))
+
+            trainer = Trainer(model=model, metrics=[metric], enable_logging=False)
+            loader = get_dataloader(dataset, batch_size=3)
+            y_true, _, _ = trainer.inference(loader)
+            self.assertEqual(sorted(len(y) for y in y_true), [3, 4, 5, 6])
+            self.assertIn(metric, trainer.evaluate(loader))
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/core/test_processor_normalization.py b/tests/core/test_processor_normalization.py
new file mode 100644
index 000000000..b546e51d0
--- /dev/null
+++ b/tests/core/test_processor_normalization.py
@@ -0,0 +1,131 @@
+import unittest
+from datetime import datetime, timedelta
+
+import numpy as np
+import torch
+
+from pyhealth.processors import (
+    TemporalTimeseriesProcessor,
+    TensorProcessor,
+    TimeseriesProcessor,
+)
+from pyhealth.processors.utils import fit_normalization, grid_steps, sample_on_grid
+
+T0 = datetime(2020, 1, 1)
+NAN = float("nan")
+
+
+def _series(values):
+    return ([T0 + timedelta(hours=h) for h in range(len(values))], values)
+
+
+class TestFitNormalization(unittest.TestCase):
+    def test_zscore_matches_numpy_across_chunks(self):
+        rng = np.random.default_rng(0)
+        data = rng.normal(5.0, 2.0, size=(100, 3))
+        data[rng.random(data.shape) < 0.3] = np.nan
+        center, scale = fit_normalization(np.array_split(data, 7), "zscore")
+        np.testing.assert_allclose(center, np.nanmean(data, axis=0))
+        np.testing.assert_allclose(scale, np.nanstd(data, axis=0))
+
+    def test_minmax(self):
+        data = np.array([[1.0, NAN], [3.0, 2.0], [NAN, 6.0]])
+        center, scale = fit_normalization([data[:1], data[1:]], "minmax")
+        self.assertEqual(center, [1.0, 2.0])
+        self.assertEqual(scale, [2.0, 4.0])
+
+    def test_constant_and_unobserved_features_get_unit_scale(self):
+        data = np.array([[4.0, NAN], [4.0, NAN]])
+        center, scale = fit_normalization([data], "zscore")
+        self.assertEqual(center, [4.0, 0.0])
+        self.assertEqual(scale, [1.0, 1.0])
+
+    def test_unknown_method(self):
+        with self.assertRaises(ValueError):
+            fit_normalization([np.zeros((1, 1))], "robust")
+
+
+class TestGrid(unittest.TestCase):
+    def test_grid_steps_floor_timestamps(self):
+        times = [T0, T0 + timedelta(minutes=59), T0 + timedelta(minutes=61)]
+        self.assertEqual(grid_steps(times, timedelta(hours=1)).tolist(), [0, 0, 1])
+
+    def test_sample_on_grid_keeps_the_last_value_of_a_step(self):
+        times = [T0, T0 + timedelta(minutes=30), T0 + timedelta(hours=2)]
+        out = sample_on_grid(times, [1.0, 2.0, 3.0], timedelta(hours=1))
+        np.testing.assert_array_equal(out[:, 0], [2.0, NAN, 3.0])
+
+    def test_temporal_timeseries_processor_uses_the_shared_grid(self):
+        value = ([T0, T0 + timedelta(hours=2)], np.array([[1.0], [3.0]]))
+        p = TemporalTimeseriesProcessor()
+        p.fit([{"x": value}], "x")
+        out = p.process(value)
+        np.testing.assert_array_equal(out["value"].numpy()[:, 0], [1.0, 1.0, 3.0])
+        np.testing.assert_array_equal(out["time"].numpy(), [0.0, 1.0, 2.0])
+
+
+class TestTimeseriesProcessorOptions(unittest.TestCase):
+    def test_defaults_are_unchanged(self):
+        p = TimeseriesProcessor()
+        value = _series([[NAN, 1.0], [2.0, NAN]])
+        p.fit([{"x": value}], "x")
+        self.assertFalse(p.learns_statistics)
+        self.assertEqual(p.size(), 2)
+        np.testing.assert_array_equal(
+            p.process(value).numpy(), [[0.0, 1.0], [2.0, 1.0]]
+        )
+
+    def test_zscore_normalization_with_indicators(self):
+        train = [{"x": _series([[1.0], [NAN], [3.0]])}, {"x": _series([[5.0]])}]
+        p = TimeseriesProcessor(normalization="zscore", missing_indicators=True)
+        p.fit(train, "x")
+        self.assertTrue(p.learns_statistics)
+        self.assertEqual(p.size(), 2)
+        mean, std = 3.0, np.std([1.0, 3.0, 5.0])
+
+        out = p.process(_series([[NAN], [5.0], [NAN]])).numpy()
+        # Leading gap -> 0 (the training mean), then forward-filled.
+        np.testing.assert_allclose(out[:, 0], [0.0, (5.0 - mean) / std, (5.0 - mean) / std])
+        np.testing.assert_array_equal(out[:, 1], [1.0, 0.0, 1.0])
+
+    def test_invalid_normalization(self):
+        with self.assertRaises(ValueError):
+            TimeseriesProcessor(normalization="robust")
+
+
+class TestTensorProcessorOptions(unittest.TestCase):
+    def test_defaults_are_unchanged(self):
+        p = TensorProcessor()
+        p.fit([{"x": [1.0, NAN]}], "x")
+        self.assertFalse(p.learns_statistics)
+        out = p.process([1.0, NAN])
+        self.assertTrue(torch.isnan(out[1]))
+
+    def test_normalization_indicators_and_zero_fill(self):
+        train = [{"x": [1.0, 10.0]}, {"x": [3.0, NAN]}]
+        p = TensorProcessor(
+            normalization="zscore", missing_indicators=True, impute_strategy="zero"
+        )
+        p.fit(train, "x")
+        self.assertEqual(p.size(), 4)
+        out = p.process([NAN, 10.0])
+        np.testing.assert_allclose(out.numpy(), [0.0, 0.0, 1.0, 0.0])
+        out = p.process([3.0, NAN])
+        np.testing.assert_allclose(out.numpy(), [1.0, 0.0, 0.0, 1.0])
+
+    def test_scalar_shapes_match_dim(self):
+        for kwargs, shape in (
+            ({"normalization": "zscore"}, ()),
+            ({"missing_indicators": True}, (2,)),
+            ({"normalization": "zscore", "missing_indicators": True}, (2,)),
+        ):
+            p = TensorProcessor(**kwargs)
+            p.fit([{"x": 1.0}, {"x": 3.0}], "x")
+            out = p.process(2.0)
+            self.assertEqual(tuple(out.shape), shape)
+            self.assertEqual(p.dim(), (len(shape),))
+            self.assertEqual(len(p.spatial()), len(shape))
+
+
+if __name__ == "__main__":
+    unittest.main()