buzeraandClaude Opus 5.5
b1577a0c78
Validation: the walk-forward evaluator now replays all zones together with the full model (electrical heat, neighbour exchange, ventilation and recovery) using the parameters frozen at the forecast start and the electrical heat and window openings that actually happened, and scores the base 3-parameter model on the same spans. Forecast Error sensors add base_model_mae and improvement_vs_base (share of the base model's error the new features remove; 0 = no gain, negative = worse). Saved 0.3 statistics load with their own error as the base column. On synthetic apartments, 60-minute error full vs base: gaming office 0.031 vs 0.182 °C (83 % better), bedroom with ventilation 0.026 vs 0.218 °C (88 %), two coupled rooms 0.024 vs 0.037 °C (34 %; gone by 120 min, where the constant-outdoor assumption dominates). The same ventilation week without window sensors shows no improvement, as it should. These check the method, not real-world accuracy. Performance: every term of a zone's 5-minute change is affine in the zone temperatures and the radiator lag, and which layer applies depends only on which inputs exist. compile_step_model reads the coefficients off the real model per zone and step, so the plan search does a few multiply-adds per step. It matches the direct model to 1e-12 and makes a 180-minute, 5-zone search 4.5x faster (0.73 s here, close to the old base-only search). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
120 lines
6.1 KiB
Python
120 lines
6.1 KiB
Python
"""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_full_model_replays_actual_inputs_and_is_compared_with_base(self):
|
|
base = (0.004, 0.12, 0.0005)
|
|
full = {'base': base, 'uses_extended': True,
|
|
'extended': (['loss', 'heat', 'bias', 'internal'], base + (0.15,)),
|
|
'ventilation': (['vent_natural'], (0.01,))}
|
|
evaluator, temp = ev.ForecastEvaluator(0.1), 20.0
|
|
for k in range(40):
|
|
internal = 0.4 if (k // 8) % 2 else 0.0 # gaming on and off
|
|
opening = 1.0 if k % 10 in (3, 4) else 0.0 # short ventilation
|
|
if k:
|
|
temp += 0.004 * (5 - temp) + 0.12 * 0.5 + 0.0005 + 0.15 * internal + 0.01 * opening * (5 - temp)
|
|
evaluator.observe(START + timedelta(seconds=300 * k), {'r': temp}, 5.0, 0.5, 0.0, {'r': base},
|
|
models={'r': full},
|
|
inputs={'r': {'internal': internal, 'ventilation': {'opening': opening}}})
|
|
stats = evaluator.summary(['r'])[60]
|
|
self.assertAlmostEqual(stats['mae'], 0, places=6)
|
|
self.assertGreater(stats['base_model_mae'], 0.05)
|
|
self.assertEqual(stats['improvement_vs_base'], 1.0)
|
|
|
|
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],
|
|
'sensor.e': [2, 0.1, 0.2, 0.4, 0.3], 'sensor.f': [2, 0.1, 0.2, 0.4, -1]}}}})
|
|
# A 0.3 bucket has no base-model column; its model was the base model then.
|
|
self.assertEqual(restored.as_dict(), {'days': {day: {'15': {
|
|
'sensor.d': [2, 0.1, 0.2, 0.4, 0.2], 'sensor.e': [2, 0.1, 0.2, 0.4, 0.3]}}}})
|
|
for garbage in (None, [], {'days': []}):
|
|
restored.load(garbage)
|
|
self.assertEqual(restored.as_dict(), {'days': {}})
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|