Score forecasts against reality; expose forecast trajectories

There was no evidence of how accurate the thermal model is: the only
number was an in-sample 5-minute error.

At every poll, record each room's reading, the outdoor temperature, the
radiator-lag estimate and the model parameters. 15, 30, 60 and 120
minutes later, replay the room from that poll with the parameters it
had then (out of sample), the outdoor temperature held constant and the
radiator lag that actually followed from the relay's real history, and
compare with the reading now. A no-change forecast is scored alongside.
Only spans with regular polls and a well-known lag are scored. 7 days
of per-room statistics are saved with the model (optional "evaluation"
payload key; about 18 KB at most).

Four new sensors, Heating MPC Forecast Error 15/30/60/120 min, show the
mean absolute error with bias, persistence error, skill and per-room
detail. On the synthetic lagged plant the 60-minute error is 0.03 °C
against 0.39 °C for persistence.

Room forecast entities gain forecast_15min and plan_15min attributes
and are labelled counterfactual. Volatile attributes are unrecorded.
The model is now saved after scoring, so saved statistics are current.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tests/ha/test_lifecycle.py

@@ -24,7 +24,7 @@ async def test_setup_reload_unload(hass, entry, inputs):
     assert coordinator.config_entry is entry
     registry = er.async_get(hass)
     entities = er.async_entries_for_config_entry(registry, entry.entry_id)
-    assert len(entities) == 5
+    assert len(entities) == 9  # 5 core + 4 forecast-error sensors
     ids = {e.unique_id: e.entity_id for e in entities}
     assert coordinator.data["status"] == "learning"
     status = hass.states.get(ids[f"{entry.entry_id}_status"])
@@ -55,7 +55,7 @@ async def test_options_reload_removes_only_old_room(hass, entry, inputs):
     assert hass.data[DOMAIN][entry.entry_id].rooms == ["sensor.new"]
     assert registry.async_get(old) is None
     assert registry.async_get_entity_id("sensor", DOMAIN, f"{entry.entry_id}_sensor.new_predicted_60min")
-    assert len(er.async_entries_for_config_entry(registry, entry.entry_id)) == 5
+    assert len(er.async_entries_for_config_entry(registry, entry.entry_id)) == 9
 
 
 async def test_storage_restore_and_remove(hass, entry, inputs):
@@ -69,6 +69,12 @@ async def test_storage_restore_and_remove(hass, entry, inputs):
     restored = hass.data[DOMAIN][entry.entry_id]
     assert restored.learners["sensor.room"].count == 250
     assert restored.data["status"] == "ready"
+    forecast_id = er.async_get(hass).async_get_entity_id(
+        "sensor", DOMAIN, f"{entry.entry_id}_sensor.room_predicted_60min")
+    forecast = hass.states.get(forecast_id)
+    assert len(forecast.attributes["forecast_15min"]) == len(forecast.attributes["plan_15min"]) == 4
+    assert forecast.attributes["forecast_15min"][3] == float(forecast.state)
+    assert {"forecast_15min", "plan_15min"} <= forecast.state_info["unrecorded_attributes"]
     assert await hass.config_entries.async_remove(entry.entry_id)
     await hass.async_block_till_done()
     assert await Store(hass, 1, f"{DOMAIN}.{entry.entry_id}").async_load() is None
@@ -210,3 +216,25 @@ async def test_removed_room_model_survives_options_round_trip(hass, entry, input
         entry, options={"room_entities": ["sensor.room", "sensor.new"]})
     await hass.async_block_till_done()
     assert hass.data[DOMAIN][entry.entry_id].learners["sensor.room"].count == 250
+
+
+async def test_forecast_error_sensors_score_scheduled_polls(hass, entry, inputs, freezer):
+    coordinator = await setup(hass, entry)
+    registry = er.async_get(hass)
+    ids = {horizon: registry.async_get_entity_id("sensor", DOMAIN, f"{entry.entry_id}_forecast_error_{horizon}min")
+           for horizon in (15, 30, 60, 120)}
+    assert all(hass.states.get(entity).state == "unknown" for entity in ids.values())
+    future = dt_util.utcnow()
+    for _ in range(3):  # 15 minutes of regular polls
+        future += timedelta(minutes=5, seconds=1)
+        freezer.move_to(future)
+        async_fire_time_changed(hass, future)
+        await hass.async_block_till_done()
+    state = hass.states.get(ids[15])
+    assert coordinator.data["forecast_accuracy"][15]["samples"] == 1
+    assert float(state.state) == coordinator.data["forecast_accuracy"][15]["mae"] > 0
+    assert state.attributes["samples"] == 1
+    assert state.attributes["rooms"]["sensor.room"]["samples"] == 1
+    assert state.attributes["unit_of_measurement"] == "°C"
+    assert "rooms" in state.state_info["unrecorded_attributes"]
+    assert hass.states.get(ids[30]).state == "unknown"

tests/test_coordinator.py

@@ -423,7 +423,7 @@ class BoundaryTests(unittest.IsolatedAsyncioTestCase):
             for entity, value in (('sensor.room', temp), ('sensor.outside', outside)):
                 self.states[entity].state = f'{value:.5f}'
                 self.states[entity].last_reported = self.now
-            await self.sample()
+            data = await self.sample()
             command = temp < 20.7 or (on and temp < 21.3)
             if command != on:  # relay switches 120 s after this poll
                 self.switch('on' if command else 'off', 180)
@@ -436,6 +436,11 @@ class BoundaryTests(unittest.IsolatedAsyncioTestCase):
         self.assertGreater(model.on_count, 100)
         self.assertAlmostEqual(model.theta[1], b, delta=0.1 * b)
         self.assertAlmostEqual(model.theta[0], a, delta=0.25 * a)
+        # Walk-forward replay with the real relay history beats persistence clearly.
+        accuracy = data['forecast_accuracy'][60]
+        self.assertGreater(accuracy['samples'], 1400)
+        self.assertLess(accuracy['mae'], 0.06)
+        self.assertGreater(accuracy['skill'], 0.85)
 
     async def test_several_relay_changes_between_polls_are_integrated(self):
         advance = coordinator.advance_lag
@@ -654,3 +659,30 @@ class BoundaryTests(unittest.IsolatedAsyncioTestCase):
             self.assertIs(type(data['count']), int)
             float(data['error_sum'])
             [float(v) for v in data['theta'] + [v for row in data['p'] for v in row]]
+
+    async def test_forecasts_are_scored_saved_and_restored(self):
+        for _ in range(5):
+            self.fresh('sensor.room', 'sensor.outside')
+            data = await self.sample()
+        accuracy = data['forecast_accuracy']
+        self.assertEqual((accuracy[15]['samples'], accuracy[30]['samples'], accuracy[120]['samples']),
+                         (2, 0, 0))
+        # The default parameters predict cooling while the room held 20 °C.
+        self.assertLess(accuracy[15]['rooms']['sensor.room']['bias'], 0)
+        self.assertEqual(accuracy[15]['persistence_mae'], 0)
+        payload = self.c._store.async_save.call_args.args[0]
+        restored = coordinator.HeatingMPCCoordinator(self.hass, self.entry)
+        restored._store.async_load.return_value = payload
+        await restored.async_restore_model()
+        self.assertEqual(restored._evaluator.summary(['sensor.room']), self.c._evaluator.summary(['sensor.room']))
+        # A poll without a usable relay state breaks the replay window.
+        self.states['switch.boiler'].state = 'unavailable'
+        self.fresh('sensor.room', 'sensor.outside')
+        await self.sample()
+        self.states['switch.boiler'].state = 'off'
+        for _ in range(3):
+            self.fresh('sensor.room', 'sensor.outside')
+            data = await self.sample()
+        self.assertEqual(data['forecast_accuracy'][15]['samples'], 2)
+        self.fresh('sensor.room', 'sensor.outside')
+        self.assertEqual((await self.sample())['forecast_accuracy'][15]['samples'], 3)

tests/test_evaluation.py

@@ -0,0 +1,97 @@
+"""Walk-forward forecast evaluation, without Home Assistant."""
+import sys
+import types
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+COMPONENTS = Path(__file__).resolve().parents[1] / 'custom_components' / 'heating_mpc'
+# Fake only the package container, to avoid importing Home Assistant from __init__.
+sys.modules.setdefault('heating_mpc', types.ModuleType('heating_mpc'))
+sys.modules['heating_mpc'].__path__ = [str(COMPONENTS)]
+from heating_mpc import evaluation as ev  # noqa: E402
+
+THETA = (0.004, 0.12, 0.0005)
+START = datetime(2026, 10, 26, tzinfo=timezone.utc)
+
+
+def plant(polls, theta=THETA, model_theta=None, outside=5.0, lag_at=lambda k: 1.0 if k % 30 < 12 else 0.0,
+          evaluator=None, uncertainty_at=lambda k: 0.0, start=START, step=300):
+    """Feed polls of a room that follows the 5-minute model exactly with `theta`."""
+    a, b, c = theta
+    evaluator = evaluator or ev.ForecastEvaluator(0.1)
+    temp = 20.0
+    for k in range(polls):
+        lag = lag_at(k)
+        if k:
+            temp += a * (outside - temp) + b * lag + c
+        evaluator.observe(start + timedelta(seconds=k * step), {'sensor.room': temp}, outside, lag,
+                          uncertainty_at(k), {'sensor.room': model_theta or theta})
+    return evaluator
+
+
+class EvaluationTests(unittest.TestCase):
+    def test_exact_model_scores_zero_error_and_beats_persistence(self):
+        summary = plant(60).summary(['sensor.room'])
+        for horizon, expected_samples in ((15, 57), (30, 54), (60, 48), (120, 36)):
+            with self.subTest(horizon=horizon):
+                stats = summary[horizon]
+                self.assertEqual(stats['samples'], expected_samples)
+                self.assertEqual(stats['rooms']['sensor.room']['samples'], expected_samples)
+                self.assertAlmostEqual(stats['mae'], 0, places=6)
+                self.assertGreater(stats['persistence_mae'], 0.05)
+                self.assertEqual(stats['skill'], 1)
+
+    def test_parameters_from_the_forecast_start_are_used(self):
+        # A model without heat gain under-predicts warming: negative bias.
+        stats = plant(10, model_theta=(0.004, 0.0, 0.0005), lag_at=lambda k: 1.0).summary(['sensor.room'])
+        self.assertLess(stats[15]['bias'], -0.3)
+        self.assertEqual(stats[15]['bias'], -stats[15]['mae'])
+
+    def test_replay_matches_hand_calculation(self):
+        temp = 20.0
+        for _ in range(3):
+            temp += 0.004 * (5 - temp) + 0.12 * 0.5 + 0.0005
+        self.assertAlmostEqual(ev.replay(20.0, 5.0, THETA, [(0.5, 300)] * 3), temp)
+        # Jitter is normalized like learning: 330 s counts as 1.1 steps.
+        self.assertAlmostEqual(ev.replay(20.0, 5.0, (0, 0.1, 0), [(1.0, 330)]), 20.11)
+
+    def test_breaks_uncertain_lag_and_missing_rooms_are_not_scored(self):
+        evaluator = plant(4)
+        self.assertEqual(evaluator.summary(['sensor.room'])[15]['samples'], 1)
+        evaluator.reset_window()
+        plant(3, evaluator=evaluator, start=START + timedelta(minutes=20))
+        self.assertEqual(evaluator.summary(['sensor.room'])[15]['samples'], 1)
+        # Irregular spacing restarts the window as well.
+        jittery = plant(8, step=400)
+        self.assertEqual(jittery.summary(['sensor.room'])[15]['samples'], 0)
+        uncertain = plant(8, uncertainty_at=lambda k: 0.5 if k == 4 else 0.0)
+        self.assertEqual(uncertain.summary(['sensor.room'])[15]['samples'], 2)  # k = 3 and 7 only
+        self.assertEqual(plant(8).summary(['sensor.other'])[15]['samples'], 0)
+
+    def test_window_keeps_seven_days(self):
+        evaluator = plant(10)
+        before = evaluator.summary(['sensor.room'])[15]['samples']
+        plant(10, evaluator=evaluator, start=START + timedelta(days=7))
+        self.assertEqual(evaluator.summary(['sensor.room'])[15]['samples'], 7)
+        self.assertEqual(before, 7)
+
+    def test_saved_statistics_round_trip_and_reject_garbage(self):
+        evaluator = plant(10)
+        restored = ev.ForecastEvaluator(0.1)
+        restored.load(evaluator.as_dict())
+        self.assertEqual(restored.summary(['sensor.room']), evaluator.summary(['sensor.room']))
+        day = START.date().isoformat()
+        restored.load({'days': {
+            'not-a-day': {'15': {'sensor.room': [1, 0.0, 0.1, 0.1]}},
+            day: {'45': {'sensor.room': [1, 0.0, 0.1, 0.1]},
+                  '15': {'sensor.a': [1, 0.0, -0.1, 0.1], 'sensor.b': [True, 0, 0, 0],
+                         'sensor.c': [1, float('nan'), 0, 0], 'sensor.d': [2, 0.1, 0.2, 0.4]}}}})
+        self.assertEqual(restored.as_dict(), {'days': {day: {'15': {'sensor.d': [2, 0.1, 0.2, 0.4]}}}})
+        for garbage in (None, [], {'days': []}):
+            restored.load(garbage)
+            self.assertEqual(restored.as_dict(), {'days': {}})
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_physics.py

@@ -76,7 +76,9 @@ class ModelTests(unittest.TestCase):
                 temps, lag = m.simulate_step(temps, 5, lag, bit, learners)
             if index == 3:
                 self.assertEqual(out['predicted_60m']['r'], round(temps['r'], 3))
+            self.assertEqual(out['trajectory_15min']['r'][index], round(temps['r'], 3))
         self.assertEqual(out['predicted_end']['r'], round(temps['r'], 3))
+        self.assertEqual(len(out['trajectory_15min']['r']), 6)
 
     def test_jitter_normalized_and_rejection_counted(self):
         nominal, delayed = m.Learner(), m.Learner()