Some Formula tests are flaky
### What happened?

```
    async def test_3_phase_formulas(self, mocker: MockerFixture) -> None:
        """Test 3 phase formulas current formulas and their composition."""
        mockgrid = MockMicrogrid(grid_side_meter=False)
        mockgrid.add_batteries(3)
        mockgrid.add_ev_chargers(1)
        request_chan, channel_registry = await mockgrid.start(mocker)
        logical_meter = LogicalMeter(
            channel_registry,
            request_chan.new_sender(),
            microgrid.get().component_graph,
        )
        grid_current_recv = await logical_meter.grid_current()
        pool = EVChargerPool(channel_registry, request_chan.new_sender())
        ev_current_recv = await pool.total_current()
    
        engine = (grid_current_recv.clone() - ev_current_recv.clone()).build(
            "net_current"
        )
        net_current_recv = engine.new_receiver()
    
        await synchronize_receivers(
            [grid_current_recv, ev_current_recv, net_current_recv]
        )
    
        for _ in range(10):
            grid_amps = await grid_current_recv.receive()
            ev_amps = await ev_current_recv.receive()
            net_amps = await net_current_recv.receive()
    
>           assert grid_amps.value_p1 is not None and grid_amps.value_p1 > 0.0
E           assert (None is not None)
E            +  where None = Sample3Phase(timestamp=datetime.datetime(2023, 3, 10, 7, 9, 39, 78987, tzinfo=datetime.timezone.utc), value_p1=None, value_p2=None, value_p3=None).value_p1
```

### What did you expect instead?

We should fix them

### Affected version(s)

_No response_

### Affected part(s)

Unit, integration and performance tests (part:tests)

### Extra information

_No response_
diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md
index 2e596e0..bb7e4ac 100644
--- a/RELEASE_NOTES.md
+++ b/RELEASE_NOTES.md
@@ -12,12 +12,10 @@
   battery_power_receiver = microgrid.battery_pool().power.new_receiver()
   ```
 
-+ Formulas composition has changed (#327) -
-  - receivers from formulas are no longer composable.
-  - formula composition is now done by composing FormulaEngine instances.
-  - Automatic formulas from the logical meter and *pools, are now
-    properties, and return `FormulaEngine` instances, which can be
-    composed further, or can provide a receiver to fetch values.
+* Formulas composition has changed (#327)
+  * Receivers from formulas are no longer composable.
+  * Formula composition is now done by composing FormulaEngine instances.
+  * Automatic formulas from the logical meter and \*pools, are now properties, and return `FormulaEngine` instances, which can be composed further, or can provide a receiver to fetch values.
 
   ``` python
   grid_power_receiver = microgrid.logical_meter().grid_power.new_receiver()
@@ -30,26 +28,47 @@
   inverter_power_receiver = self._inverter_power.new_receiver()
   ```
 
-* Update BatteryStatus to mark battery with unknown capacity as not working (#263)
+* Update `BatteryStatus` to mark battery with unknown capacity as not working (#263)
+
 * The channels dependency was updated to v0.14.0 (#292)
+
 * Some properties for `PowerDistributingActor` results were renamed to be more consistent between `Success` and `PartialFailure`:
+
   * The `Success.used_batteries` property was renamed to `succeeded_batteries`.
   * The `PartialFailure.success_batteries` property was renamed to `succeeded_batteries`.
   * The `succeed_power` property was renamed to `succeeded_power` for both `Success` and `PartialFailure`.
-* Update MovingWindow to accept size parameter as timedelta instead of int (#269).
-  This change allows users to define the time span of the moving window more intuitively, representing the duration over which samples will be stored.
-* Add a resampler in the MovingWindow to control the granularity of the samples to be stored in the underlying buffer (#269).
-  Notice that the parameter `sampling_period` has been renamed to `input_sampling_period`
-  to better distinguish it from the sampling period parameter in the resampler.
+
 * The serialization feature for the ringbuffer was made more flexible. The `dump` and `load` methods can now work directly with a ringbuffer instance.
-* The `ResamplerConfig` now takes the resampling period as a `timedelta`. The configuration was renamed from `resampling_period_s` to `resampling_period` accordingly.
-* The `SourceProperties` of the resampler now uses a `timedelta` for the input sampling period. The attribute was renamed from `sampling_period_s` to `sampling_period` accordingly.
+
+* `MovingWindow`
+
+  * Accept the `size` parameter as `timedelta` instead of `int` (#269).
+
+    This change allows users to define the time span of the moving window more intuitively, representing the duration over which samples will be stored.
+
+  * The input data will be resampled if a `resampler_config` is passed (#269).
+
+    This allows controlling the granularity of the samples to be stored in the underlying buffer.
+
+    Note that the parameter `sampling_period` has been renamed to `input_sampling_period` to better distinguish it from the sampling period parameter in the `resampler_config`.
+
+* `Resampler`
+
+  * The `ResamplerConfig` now takes the resampling period as a `timedelta`. The configuration was renamed from `resampling_period_s` to `resampling_period` accordingly.
+
+  * The `SourceProperties` of the resampler now uses a `timedelta` for the input sampling period. The attribute was renamed from `sampling_period_s` to `sampling_period` accordingly.
+
+  * The periods are now aligned to the `UNIX_EPOCH` by default.
+
+    To use the old behaviour (aligning to the time the resampler was created), pass `align_to=None` to the `ResamplerConfig`.
 
 ## New Features
 
-* Automatic creation of core data-pipeline actors, to eliminate a lot
-  of boiler plate code.  This makes it much simpler to deploy apps
-  (#270).  For example:
+* The core data-pipeline actors are now created automatically (#270).
+
+  This eliminates a lot of boiler plate code and makes it much simpler to deploy apps.
+
+  For example:
 
   ``` python
   async def run():
@@ -59,9 +78,14 @@
       grid_power = microgrid.logical_meter().grid_power()
   ```
 
-* The `Result` class (and subclasses) for the `PowerDistributingActor` are now dataclasses, so logging them will produce a more detailed output.
+* The `Result` class (and subclasses) for the `PowerDistributingActor` are now `dataclass`es, so logging them will produce a more detailed output.
+
+* The `Resampler` can now can align the resampling period to an arbitrary `datetime`.
+
+  This can be configured via the new `align_to` option in the `ResamplerConfig`. By default the resampling period is aligned to the `UNIX_EPOCH`.
 
 ## Bug Fixes
 
-* Change PowerDistributor to use all batteries if none is working (#258)
-* Update the ordered ring buffer to fix the len() function so that it returns a value equal to or greater than zero, as expected (#274)
+* Change `PowerDistributor` to use all batteries when none are working (#258)
+
+* Update the ordered ring buffer to fix the `len()` function so that it returns a value equal to or greater than zero, as expected (#274)
diff --git a/src/frequenz/sdk/timeseries/__init__.py b/src/frequenz/sdk/timeseries/__init__.py
index 8f1dc70..b96d414 100644
--- a/src/frequenz/sdk/timeseries/__init__.py
+++ b/src/frequenz/sdk/timeseries/__init__.py
@@ -6,8 +6,35 @@ Handling of timeseries streams.
 
 A timeseries is a stream (normally an async iterator) of
 [`Sample`][frequenz.sdk.timeseries.Sample]s.
+
+# Periodicity and alignment
+
+All the data produced by this package is always periodic and aligned to the
+`UNIX_EPOCH` (by default).
+
+Classes normally take a (re)sampling period as and argument and, optionally, an
+`align_to` argument.
+
+This means timestamps are always separated exaclty by a period, and that this
+timestamp falls always at multiples of the period, starting at the `align_to`.
+
+This ensures that the data is predictable and consistent among restarts.
+
+Example:
+    If we have a period of 10 seconds, and are aligning to the UNIX
+    epoch. Assuming the following timeline starts in 1970-01-01 00:00:00
+    UTC and our current `now` is 1970-01-01 00:00:32 UTC, then the next
+    timestamp will be at 1970-01-01 00:00:40 UTC:
+
+    ```
+    align_to = 1970-01-01 00:00:00         next event = 1970-01-01 00:00:40
+    |                                       |
+    |---------|---------|---------|-|-------|---------|---------|---------|
+    0        10        20        30 |      40        50        60        70
+                                   now = 1970-01-01 00:00:32
+    ```
 """
 
-from ._base_types import Sample, Sample3Phase
+from ._base_types import UNIX_EPOCH, Sample, Sample3Phase
 
-__all__ = ["Sample", "Sample3Phase"]
+__all__ = ["Sample", "Sample3Phase", "UNIX_EPOCH"]
diff --git a/src/frequenz/sdk/timeseries/_base_types.py b/src/frequenz/sdk/timeseries/_base_types.py
index 0d76291..c5cc8d2 100644
--- a/src/frequenz/sdk/timeseries/_base_types.py
+++ b/src/frequenz/sdk/timeseries/_base_types.py
@@ -7,9 +7,12 @@ from __future__ import annotations
 
 import functools
 from dataclasses import dataclass, field
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import Callable, Iterator, Optional, overload
 
+UNIX_EPOCH = datetime.fromtimestamp(0.0, tz=timezone.utc)
+"""The UNIX epoch (in UTC)."""
+
 
 # Ordering by timestamp is a bit arbitrary, and it is not always what might be
 # wanted. We are using this order now because usually we need to do binary
diff --git a/src/frequenz/sdk/timeseries/_resampling.py b/src/frequenz/sdk/timeseries/_resampling.py
index 5bcf447..94c3252 100644
--- a/src/frequenz/sdk/timeseries/_resampling.py
+++ b/src/frequenz/sdk/timeseries/_resampling.py
@@ -17,6 +17,7 @@ from typing import AsyncIterator, Callable, Coroutine, Optional, Sequence
 
 from .._internal.asyncio import cancel_and_await
 from . import Sample
+from ._base_types import UNIX_EPOCH
 
 _logger = logging.getLogger(__name__)
 
@@ -191,6 +192,17 @@ class ResamplerConfig:
     It must be at bigger than `warn_buffer_len`.
     """
 
+    align_to: datetime | None = UNIX_EPOCH
+    """The time to align the resampling period to.
+
+    The resampling period will be aligned to this time, so the first resampled
+    sample will be at the first multiple of `resampling_period` starting from
+    `align_to`. It must be an aware datetime and can be in the future too.
+
+    If `align_to` is `None`, the resampling period will be aligned to the
+    time the resampler is created.
+    """
+
     def __post_init__(self) -> None:
         """Check that config values are valid.
 
@@ -231,6 +243,10 @@ class ResamplerConfig:
                 self.initial_buffer_len,
                 self.warn_buffer_len,
             )
+        if self.align_to is not None and self.align_to.tzinfo is None:
+            raise ValueError(
+                f"align_to ({self.align_to}) should be a timezone aware datetime"
+            )
 
 
 class SourceStoppedError(RuntimeError):
@@ -348,9 +364,7 @@ class Resampler:
         self._resamplers: dict[Source, _StreamingHelper] = {}
         """A mapping between sources and the streaming helper handling that source."""
 
-        self._window_end: datetime = (
-            datetime.now(timezone.utc) + self._config.resampling_period
-        )
+        self._window_end: datetime = self._calculate_window_end()
         """The time in which the current window ends.
 
         This is used to make sure every resampling window is generated at
@@ -358,6 +372,9 @@ class Resampler:
         never fire at the exact requested time, so if we don't use a precise
         time for the end of the window, the resampling windows we produce will
         have different sizes.
+
+        The window end will also be aligned to the `config.align_to` time, so
+        the window end is deterministic.
         """
 
     @property
@@ -491,6 +508,29 @@ class Resampler:
                 self._config.resampling_period,
             )
 
+    def _calculate_window_end(self) -> datetime:
+        """Calculate the end of the current resampling window.
+
+        The calculated resampling window end is a multiple of
+        `self._config.resampling_period` starting at `self._config.align_to`.
+
+        if `self._config.align_to` is `None`, the current time is used.
+
+        Returns:
+            The end of the current resampling window aligned to
+                `self._config.align_to`.
+        """
+        now = datetime.now(timezone.utc)
+        period = self._config.resampling_period
+        align_to = self._config.align_to
+
+        if align_to is None:
+            return now + period
+
+        elapsed = (now - align_to) % period
+
+        return now + period - elapsed
+
 
 class _ResamplingHelper:
     """Keeps track of *relevant* samples to pass them to the resampling function.
diff --git a/tests/timeseries/mock_microgrid.py b/tests/timeseries/mock_microgrid.py
index 7a6133a..a54d741 100644
--- a/tests/timeseries/mock_microgrid.py
+++ b/tests/timeseries/mock_microgrid.py
@@ -119,7 +119,14 @@ class MockMicrogrid:  # pylint: disable=too-many-instance-attributes
 
         # pylint: disable=protected-access
         _data_pipeline._DATA_PIPELINE = _data_pipeline._DataPipeline(
-            ResamplerConfig(resampling_period=timedelta(seconds=self._sample_rate_s))
+            ResamplerConfig(
+                resampling_period=timedelta(seconds=self._sample_rate_s),
+                # Align to the time the resampler is created to avoid flakiness
+                # in the tests, it seems test using the mock microgrid assume
+                # that the resampling window is aligned to the start of the
+                # test.
+                align_to=None,
+            )
         )
         # pylint: enable=protected-access
 
diff --git a/tests/timeseries/test_resampling.py b/tests/timeseries/test_resampling.py
index 969a12e..58a929a 100644
--- a/tests/timeseries/test_resampling.py
+++ b/tests/timeseries/test_resampling.py
@@ -34,7 +34,8 @@ from frequenz.sdk.timeseries._resampling import (
 
 from ..utils import a_sequence
 
-# pylint: disable=too-many-locals,redefined-outer-name
+# We relax some pylint checks as for tests they don't make a lot of sense.
+# pylint: disable=too-many-lines,disable=too-many-locals,redefined-outer-name
 
 
 # Setting 'autouse' has no effect as this method replaces the event loop for all tests in the file.
@@ -168,6 +169,78 @@ async def test_helper_buffer_too_big(
     assert helper._buffer.maxlen == DEFAULT_BUFFER_LEN_MAX
 
 
+@pytest.mark.parametrize(
+    "resampling_period_s,now,align_to,result",
+    (
+        (
+            1.0,
+            datetime(2020, 1, 1, 2, 3, 5, 300000, tzinfo=timezone.utc),
+            datetime(2020, 1, 1, tzinfo=timezone.utc),
+            datetime(2020, 1, 1, 2, 3, 6, tzinfo=timezone.utc),
+        ),
+        (
+            3.0,
+            datetime(2020, 1, 1, 2, 3, 5, 300000, tzinfo=timezone.utc),
+            datetime(2020, 1, 1, 0, 0, 5, tzinfo=timezone.utc),
+            datetime(2020, 1, 1, 2, 3, 8, tzinfo=timezone.utc),
+        ),
+        (
+            10.0,
+            datetime(2020, 1, 1, 2, 3, 5, 300000, tzinfo=timezone.utc),
+            datetime(2020, 1, 1, 0, 0, 5, tzinfo=timezone.utc),
+            datetime(2020, 1, 1, 2, 3, 15, tzinfo=timezone.utc),
+        ),
+        # Future align_to
+        (
+            10.0,
+            datetime(2020, 1, 1, 2, 3, 5, 300000, tzinfo=timezone.utc),
+            datetime(2020, 1, 1, 2, 3, 18, tzinfo=timezone.utc),
+            datetime(2020, 1, 1, 2, 3, 8, tzinfo=timezone.utc),
+        ),
+    ),
+)
+def test_calculate_window_end_trivial_cases(
+    fake_time: time_machine.Coordinates,
+    resampling_period_s: float,
+    now: datetime,
+    align_to: datetime,
+    result: datetime,
+) -> None:
+    """Test the calculation of the resampling window end for simple cases."""
+    resampling_period = timedelta(seconds=resampling_period_s)
+    resampler = Resampler(
+        ResamplerConfig(
+            resampling_period=resampling_period,
+            align_to=align_to,
+        )
+    )
+    fake_time.move_to(now)
+    # pylint: disable=protected-access
+    assert resampler._calculate_window_end() == result
+
+    # Repeat the test with align_to=None, so the result should be align to now
+    # instead
+    resampler_now = Resampler(
+        ResamplerConfig(
+            resampling_period=resampling_period,
+            align_to=now,
+        )
+    )
+    resampler_none = Resampler(
+        ResamplerConfig(
+            resampling_period=resampling_period,
+            align_to=None,
+        )
+    )
+    fake_time.move_to(now)
+    # pylint: disable=protected-access
+    assert (
+        resampler_now._calculate_window_end() == resampler_none._calculate_window_end()
+    )
+    # pylint: disable=protected-access
+    assert resampler_none._calculate_window_end() == now + resampling_period
+
+
 async def test_resampling_window_size_is_constant(
     fake_time: time_machine.Coordinates, source_chan: Broadcast[Sample]
 ) -> None:
['tests/timeseries/mock_microgrid.py', 'src/frequenz/sdk/timeseries/_resampling.py', 'src/frequenz/sdk/timeseries/_base_types.py', 'src/frequenz/sdk/timeseries/__init__.py', 'tests/timeseries/test_resampling.py', 'RELEASE_NOTES.md']