Billing fix: don't bill periods from stale/out-of-range DSMR readings

Root cause (found in the live DB): register_at() returned the latest reading
for any boundary beyond the DSMR data range (a future instant or a long gap),
so future periods got start==end -> delta 0 and were written as degraded=False
'ok' rows with cost 0. recompute_range() over a future end then filled the whole
month with such fake rows, which then occupied the slots so the per-minute tick
(immutability guard skips non-degraded rows) never recomputed the real data.

- register_at: ignore a reading older than one period (15min) before the
  boundary -> returns None -> period is marked degraded instead of a fake 0.
- recompute_range: only process closed periods (t1 <= now); never future ones.
- Tests cover staleness window, out-of-range -> degraded, recompute skips
  future, and a regression that normal closed periods are unaffected.

app/services/energy_cost.py

@@ -71,6 +71,15 @@ logger = logging.getLogger(__name__)
 _PERIOD_MINUTES = 15
 _LOOKBACK_DAYS = 7  # maximum lookback window for compute_closed_periods
 
+# Maximum age a DSMR reading may have relative to the boundary being queried.
+# Under normal operation DSMR readings arrive every ~10 seconds, so a reading
+# more than one period (15 minutes) old at the boundary indicates either a data
+# gap or — critically — a *future* boundary being resolved against the last
+# historical reading.  In both cases the reading is considered stale and
+# ``register_at`` returns None, letting the period be marked degraded instead of
+# producing a spurious zero-delta "successful" row.
+_READING_MAX_STALENESS = timedelta(minutes=_PERIOD_MINUTES)
+
 # DSMR payload register keys (cumulative kWh, JSON string values).
 _KEY_D1 = "electricity_delivered_1"   # delivered low-tariff  (dal / _1)
 _KEY_D2 = "electricity_delivered_2"   # delivered high-tariff (normal / _2)
@@ -148,6 +157,13 @@ def register_at(session: Session, boundary: datetime) -> dict[str, Decimal] | No
     if row is None:
         return None
 
+    # Freshness guard: reject readings that are too old relative to *boundary*.
+    # ``recorded_at`` is stored as a naive UTC datetime in SQLite; attach UTC
+    # tzinfo before comparing with *boundary* (which is always tz-aware UTC) to
+    # avoid an "offset-naive vs offset-aware" TypeError.
+    if _as_utc(row.recorded_at) < _as_utc(boundary) - _READING_MAX_STALENESS:
+        return None
+
     payload = row.payload or {}
     try:
         d1_raw = payload[_KEY_D1]
@@ -437,18 +453,26 @@ def recompute_range(session: Session, start: datetime, end: datetime) -> int:
     """
     t0 = floor_to_quarter(_as_utc(start))
     end_utc = _as_utc(end)
+    now = datetime.now(UTC)
 
     written = 0
     while t0 < end_utc:
-        try:
+        t1 = t0 + timedelta(minutes=_PERIOD_MINUTES)
-            did_write = compute_period(session, t0, overwrite=True)
+        # Only recompute periods that have already closed (t1 ≤ now).  Even
-            if did_write:
+        # when the caller passes a future *end*, we must not write speculative
-                written += 1
+        # "future" rows: register_at would resolve to the latest historical
-        except Exception:
+        # reading for both boundaries, producing a zero-delta fake-success row
-            logger.exception(
+        # that blocks the real computation once the period actually closes.
-                "recompute_range: unexpected error for t0=%s — continuing.",
+        if t1 <= now:
-                t0.isoformat(),
+            try:
-            )
+                did_write = compute_period(session, t0, overwrite=True)
+                if did_write:
+                    written += 1
+            except Exception:
+                logger.exception(
+                    "recompute_range: unexpected error for t0=%s — continuing.",
+                    t0.isoformat(),
+                )
         t0 += timedelta(minutes=_PERIOD_MINUTES)
 
     session.commit()

tests/test_energy_cost.py

@@ -1195,3 +1195,223 @@ class TestRecomputeRange:
         assert row_after.r2_kwh == 0.0
         assert row_after.contract_version_id is None
         assert row_after.pricing == {}
+
+
+# ---------------------------------------------------------------------------
+# Bug-fix regression tests: register_at freshness + recompute_range future guard
+# ---------------------------------------------------------------------------
+
+
+class TestRegisterAtFreshness:
+    """Tests for the staleness guard added to register_at (Fix A).
+
+    Normal DSMR readings arrive every ~10 seconds.  A reading more than
+    ``_READING_MAX_STALENESS`` (15 minutes) old at a given boundary is treated
+    as absent so the period is marked degraded rather than producing a
+    zero-delta fake-success row.
+    """
+
+    def test_stale_reading_returns_none(self, energy_db: Session) -> None:
+        """A reading older than 15 min before the boundary must be rejected."""
+        reading_time = _ts(10, 0)  # 10:00
+        boundary = _ts(10, 30)     # 10:30 — 30 min later (> 15 min staleness)
+        _make_reading(energy_db, recorded_at=reading_time, source_id=1)
+        energy_db.commit()
+
+        result = register_at(energy_db, boundary)
+        assert result is None, (
+            "register_at must return None when the closest reading is more than "
+            "15 minutes before the boundary"
+        )
+
+    def test_fresh_reading_within_window_returned(self, energy_db: Session) -> None:
+        """A reading within 15 min of the boundary must be returned normally."""
+        reading_time = _ts(10, 5)  # 10:05
+        boundary = _ts(10, 15)     # 10:15 — only 10 min gap (within window)
+        _make_reading(energy_db, recorded_at=reading_time,
+                      d1="30000.0", d2="15000.0", r1="1000.0", r2="500.0",
+                      source_id=1)
+        energy_db.commit()
+
+        result = register_at(energy_db, boundary)
+        assert result is not None, (
+            "register_at must return the reading when it is within the 15-min staleness window"
+        )
+        assert result["d1"] == Decimal("30000.0")
+
+    def test_exact_15min_boundary_is_accepted(self, energy_db: Session) -> None:
+        """A reading exactly 15 min before the boundary sits at the edge — accepted."""
+        reading_time = _ts(10, 0)  # 10:00
+        boundary = _ts(10, 15)     # 10:15 — exactly 15 min gap
+        _make_reading(energy_db, recorded_at=reading_time,
+                      d1="40000.0", d2="20000.0", r1="2000.0", r2="1000.0",
+                      source_id=1)
+        energy_db.commit()
+
+        result = register_at(energy_db, boundary)
+        # boundary - reading == 15 min == staleness limit → NOT stale (strict <)
+        assert result is not None
+
+    def test_future_boundary_returns_none(self, energy_db: Session) -> None:
+        """A far-future boundary must return None (not the latest historical reading).
+
+        This is the root cause of the production bug: without the freshness
+        guard, register_at(far_future) returned the last available reading,
+        causing both period boundaries to resolve to the same row and producing
+        a zero-delta fake-success period.
+        """
+        # Insert a reading at a realistic "now" time.
+        reading_time = _ts(10, 0)  # 10:00 on 2026-06-23
+        _make_reading(energy_db, recorded_at=reading_time, source_id=1)
+        energy_db.commit()
+
+        # Query with a far-future boundary (e.g. end of month, same day +8 hours)
+        far_future_boundary = _ts(18, 0)  # 18:00 — 8 hours later
+        result = register_at(energy_db, far_future_boundary)
+        assert result is None, (
+            "register_at must return None for a far-future boundary rather than "
+            "the latest historical reading"
+        )
+
+
+class TestFuturePeriodDegraded:
+    """Ensure that a period whose boundaries fall entirely outside DSMR data
+    is written as degraded=True (not as a fake zero-cost success).
+
+    This covers the production scenario where recompute_range was called with
+    a future end date, causing compute_period to be called on not-yet-closed
+    periods.  With Fix A (register_at freshness) the boundary lookups return
+    None → degraded; with Fix B (recompute_range t1<=now guard) such periods
+    are skipped entirely.  This test exercises the Fix A path directly via
+    compute_period(overwrite=True) on a future-like period with no nearby data.
+    """
+
+    def test_out_of_range_period_is_degraded(self, energy_db: Session) -> None:
+        """A period whose boundaries have no fresh DSMR readings → degraded=True.
+
+        We insert a reading at 10:00 and call compute_period for a period
+        starting at 14:00 (4 hours later).  The boundary readings (14:00 and
+        14:15) are both more than 15 min from any available data, so
+        register_at returns None for both → degraded row written.
+        """
+        contract = _make_contract(energy_db, kind="manual", active=True)
+        _make_version(energy_db, contract, _MANUAL_VALUES, effective_from=_ts(0, 0))
+        # Only a reading at 10:00 — far from the 14:00/14:15 boundaries.
+        _make_reading(energy_db, recorded_at=_ts(10, 0), source_id=1)
+        energy_db.commit()
+
+        future_t0 = _ts(14, 0)
+        result = compute_period(energy_db, future_t0, overwrite=True)
+        assert result is True  # a record was written
+
+        row = energy_db.execute(
+            select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == future_t0)
+        ).scalar_one()
+        assert row.degraded is True, (
+            "compute_period must write degraded=True when both boundaries lack fresh readings"
+        )
+        assert row.import_cost == 0.0
+        assert row.net_cost == 0.0
+
+
+class TestRecomputeRangeNoFuturePeriods:
+    """Verify that recompute_range never writes periods whose t1 > now (Fix B)."""
+
+    def test_recompute_range_skips_future_periods(self, energy_db: Session) -> None:
+        """When end is in the future, recompute_range must not write any future rows.
+
+        This is the direct test for Fix B.  We call recompute_range with a
+        future end datetime; the only periods that may be written are those
+        where t1 <= now.  No row with period_start > now should appear in the DB.
+        """
+        contract = _make_contract(energy_db, kind="manual", active=True)
+        _make_version(energy_db, contract, _MANUAL_VALUES, effective_from=_ts(0, 0))
+        # No readings needed — we only care that future rows are NOT written.
+        energy_db.commit()
+
+        now = datetime.now(UTC)
+        far_future_end = now + timedelta(days=7)
+        # Use a past start so there are some candidate periods.
+        past_start = floor_to_quarter(now - timedelta(minutes=30))
+
+        recompute_range(energy_db, past_start, far_future_end)
+
+        # Assert no row has period_start beyond now.
+        rows = energy_db.execute(select(EnergyCostPeriod)).scalars().all()
+        future_rows = [
+            r for r in rows
+            if (r.period_start if r.period_start.tzinfo else r.period_start.replace(tzinfo=UTC)) > now
+        ]
+        assert future_rows == [], (
+            f"recompute_range must not write future periods; "
+            f"found {len(future_rows)} row(s) with period_start > now"
+        )
+
+    def test_recompute_range_chain_no_slot_squatting(self, energy_db: Session) -> None:
+        """Simulate the full production failure scenario and verify it is fixed.
+
+        Scenario:
+        1. recompute_range is called with end=far_future (reproduces the bug trigger).
+        2. compute_closed_periods is then called (reproduces the normal tick).
+        3. After the fix, step 1 must not have written any future rows, so the
+           tick in step 2 is free to compute periods normally.
+
+        We only verify that no future rows exist after step 1 (Fix B), because
+        that is the core slot-squatting prevention.
+        """
+        contract = _make_contract(energy_db, kind="manual", active=True)
+        # Contract effective from far past so all periods in scope have a version.
+        _make_version(
+            energy_db, contract, _MANUAL_VALUES,
+            effective_from=datetime(2020, 1, 1, tzinfo=UTC),
+        )
+        energy_db.commit()
+
+        now = datetime.now(UTC)
+        far_future = now + timedelta(days=30)
+        past_start = floor_to_quarter(now - timedelta(hours=1))
+
+        # Step 1: buggy call with future end.
+        recompute_range(energy_db, past_start, far_future)
+
+        # Verify no future rows were squatted.
+        all_rows = energy_db.execute(select(EnergyCostPeriod)).scalars().all()
+        future_rows = [
+            r for r in all_rows
+            if (r.period_start if r.period_start.tzinfo else r.period_start.replace(tzinfo=UTC)) > now
+        ]
+        assert future_rows == [], (
+            f"After recompute_range with future end, found {len(future_rows)} future row(s). "
+            "These would block the real tick from computing those periods."
+        )
+
+
+class TestNormalPeriodsUnaffected:
+    """Regression: the freshness guard must not break computation of normal
+    closed periods where readings are available close to the boundaries.
+    """
+
+    def test_normal_closed_period_still_computes_ok(self, energy_db: Session) -> None:
+        """A normal period with readings seconds before each boundary → non-degraded."""
+        contract = _make_contract(energy_db, kind="manual", active=True)
+        _make_version(energy_db, contract, _MANUAL_VALUES, effective_from=_ts(0, 0))
+
+        # Readings placed very close to boundaries (as DSMR normally delivers them).
+        # t0=10:00, reading at 09:59:50 (10 s before); t1=10:15, reading at 10:14:55 (5 s before)
+        _make_reading(energy_db, recorded_at=_ts(9, 59, 50),
+                      d1=_START_D1, d2=_START_D2, r1=_START_R1, r2=_START_R2, source_id=1)
+        _make_reading(energy_db, recorded_at=_ts(10, 14, 55),
+                      d1=_END_D1, d2=_END_D2, r1=_END_R1, r2=_END_R2, source_id=2)
+        energy_db.commit()
+
+        result = compute_period(energy_db, _T0)
+        assert result is True
+
+        row = energy_db.execute(
+            select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
+        ).scalar_one()
+        assert row.degraded is False, (
+            "A normal period with readings just before each boundary must not be degraded"
+        )
+        # import_cost matches hand-calc: same deltas as standard scenario
+        assert abs(row.import_cost - 0.4101) < 1e-6