"""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_rmse_effective_samples_and_quality(self): stats = plant(10, model_theta=(0.004, 0.0, 0.0005), lag_at=lambda k: 1.0).summary(['sensor.room']) s15 = stats[15] self.assertGreaterEqual(s15['rmse'], s15['mae']) # 7 overlapping 15-minute forecasts started 5 minutes apart: ~2.3 independent ones. self.assertEqual((s15['samples'], s15['effective_samples'], s15['quality']), (7, 2.3, 'insufficient')) self.assertEqual(ev.quality(5), 'indicative') self.assertEqual(ev.quality(30), 'ok') self.assertEqual(ev.effective_samples(120, 60, zones=2), 5.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]}}}, 'excluded': {day: {'15': {'sensor.d': [1, 2, 0.3], 'sensor.x': [1, -2, 0.3], 'sensor.y': [1, 2]}, '45': {'sensor.d': [1, 2, 0.3]}}}}) # A 0.3 bucket has no base-model column; its model was the base model then. # Neither 0.3 nor 0.4 buckets have squared errors: they do not count for RMSE. # 0.4 data has no excluded-span counts. self.assertEqual(restored.as_dict(), {'days': {day: {'15': { 'sensor.d': [2, 0.1, 0.2, 0.4, 0.2, 0.0, 0], 'sensor.e': [2, 0.1, 0.2, 0.4, 0.3, 0.0, 0]}}}, 'excluded': {day: {'15': {'sensor.d': [1, 2, 0.3]}}}}) self.assertIsNone(restored.summary(['sensor.d'])[15]['rmse']) self.assertEqual({k: restored.summary(['sensor.d'])[15][k] for k in ('incomplete', 'skipped', 'incomplete_base_model_mae')}, {'incomplete': 1, 'skipped': 2, 'incomplete_base_model_mae': 0.3}) for garbage in (None, [], {'days': []}): restored.load(garbage) self.assertEqual(restored.as_dict(), {'days': {}, 'excluded': {}}) T0 = START.timestamp() def run_ledger(plan, relay_on, temps, trajectory, start=20.0, polls=25, unknown_relay_at=None, skip=()): """One forecast at T0, then a poll every 5 minutes with the given relay state per poll.""" ledger = ev.OperationalForecastLedger() ledger.observe(T0, {'r': start}, None, None) ledger.record(T0, {'r': start}, {'r': trajectory}, plan, 'rev1', {'strategy': 'mpc_plan'}) for k in range(1, polls): if k in skip: continue on = None if k == unknown_relay_at else (300.0 if relay_on(k) else 0.0) ledger.observe(T0 + 300 * k, {'r': temps(k)}, on, 300.0) return ledger.summary(['r']) class OperationalForecastTests(unittest.TestCase): PLAN_OFF = [0] * 8 TRAJECTORY = [19.9, 19.8, 19.7, 19.6, 19.5, 19.4, 19.3, 19.2] # every 15 minutes def test_followed_plan_is_scored_at_every_horizon_against_persistence(self): stats = run_ledger(self.PLAN_OFF, lambda k: False, lambda k: 20.0 - k / 30, self.TRAJECTORY) for horizon, actual in ((15, 19.9), (30, 19.8), (60, 19.6), (120, 19.2)): with self.subTest(horizon=horizon): s = stats[horizon] self.assertEqual((s['samples'], s['matched'], s['counterfactual'], s['incomplete']), (1, 1, 0, 0)) self.assertAlmostEqual(s['mae'], 0, places=3) self.assertAlmostEqual(s['persistence_mae'], 20.0 - actual, places=3) self.assertEqual(s['improvement_vs_persistence'], 1.0) self.assertEqual(s['quality'], 'insufficient') # one forecast proves little def test_plan_not_followed_is_counterfactual_not_scored(self): stats = run_ledger([1] * 8, lambda k: False, lambda k: 20.0, self.TRAJECTORY) for horizon in (15, 30, 60, 120): self.assertEqual((stats[horizon]['samples'], stats[horizon]['counterfactual']), (0, 1)) self.assertIsNone(stats[horizon]['mae']) def test_plan_followed_for_a_while(self): # The thermostat switches the boiler on in the 4th block (after 45 minutes). stats = run_ledger(self.PLAN_OFF, lambda k: k > 9, lambda k: 20.0, self.TRAJECTORY) self.assertEqual([stats[h]['samples'] for h in (15, 30, 60, 120)], [1, 1, 0, 0]) self.assertEqual([stats[h]['counterfactual'] for h in (15, 30, 60, 120)], [0, 0, 1, 1]) def test_unknown_relay_history_or_missing_observations(self): stats = run_ledger(self.PLAN_OFF, lambda k: False, lambda k: 20.0, self.TRAJECTORY, unknown_relay_at=2) self.assertEqual((stats[15]['samples'], stats[15]['incomplete_reasons']['relay_history'], stats[60]['incomplete']), (0, 1, 1)) missing = run_ledger(self.PLAN_OFF, lambda k: False, lambda k: None if k == 12 else 20.0, self.TRAJECTORY) self.assertEqual((missing[60]['samples'], missing[60]['incomplete_reasons']['reading'], missing[30]['samples']), (0, 1, 1)) # No poll near the due time (polls 11 to 13 missing): the 60-minute point cannot be scored. gap = run_ledger(self.PLAN_OFF, lambda k: False, lambda k: 20.0, self.TRAJECTORY, skip=(11, 12, 13)) self.assertEqual((gap[60]['samples'], gap[60]['incomplete_reasons']['no_poll']), (0, 1)) def test_due_time_alignment(self): ledger = ev.OperationalForecastLedger() ledger.record(T0, {'r': 20.0}, {'r': self.TRAJECTORY}, self.PLAN_OFF, 'rev1', {}) for offset in (300, 600): ledger.observe(T0 + offset, {'r': 20.0}, 0.0, 300.0) ledger.observe(T0 + 900 - 200, {'r': 20.0}, 0.0, 100.0) # too early for +15 min self.assertEqual(ledger.summary(['r'])[15]['samples'], 0) ledger.observe(T0 + 900 + 100, {'r': 19.9}, 0.0, 300.0) # within half a poll of the due time self.assertEqual(ledger.summary(['r'])[15]['samples'], 1) def test_round_trip_forget_and_garbage(self): stats_ledger = ev.OperationalForecastLedger() stats_ledger.record(T0, {'r': 20.0}, {'r': self.TRAJECTORY}, [1] * 8, 'rev1', {}) stats_ledger.record(T0, {'r': 20.0}, {'r': self.TRAJECTORY}, self.PLAN_OFF, 'rev1', {}) for k in range(1, 4): stats_ledger.observe(T0 + 300 * k, {'r': 19.9}, 0.0, 300.0) restored = ev.OperationalForecastLedger(stats_ledger.as_dict()) self.assertEqual(restored.summary(['r']), stats_ledger.summary(['r'])) self.assertEqual(restored.summary(['r'])[15]['counterfactual'], 1) restored.forget('r') self.assertEqual(restored.summary(['r'])[15]['samples'], 0) empty = {'version': 2, 'days': {}, 'approximate': {}, 'verdicts': {}} for garbage in (None, [], {'version': 2, 'days': {'x': {}}, 'verdicts': {'2026-10-27': {'15': {'bogus': 3}}}}): self.assertEqual(ev.OperationalForecastLedger(garbage).as_dict(), empty) # 0.4.1 statistics accepted plans within 20 % per block: not comparable, dropped. old = {**stats_ledger.as_dict(), 'version': None} old.pop('version') self.assertEqual(ev.OperationalForecastLedger(old).as_dict(), empty) if __name__ == '__main__': unittest.main() BASE = (0.004, 0.12, 0.0005) def ventilated(observed, actual=None, wind=4.0, features=('vent_natural',), zones=('r',), internal=None, extended=None, lag_uncertain_at=(), missing_end_at=()): """Zones cooled by real ventilation; `observed[k]` is what the window sensor said for the interval ending at poll k (None: unknown). Returns the summary at 15 min.""" actual = actual if actual is not None else [1.0 if o is None else o for o in observed] snapshot = {'base': BASE, 'uses_extended': extended is not None, 'extended': extended, 'ventilation': (list(features), (0.02,) * len(features))} evaluator, temps = ev.ForecastEvaluator(0.1), {zone: 20.0 for zone in zones} for k, seen in enumerate(observed): if k: temps = {zone: t + 0.004 * (5 - t) + 0.12 * 0.5 + 0.0005 + 0.02 * actual[k] * (5 - t) * (1 + (wind or 0) * ('vent_wind' in features)) for zone, t in temps.items()} vent = {'opening': seen, 'facing': 0.0, 'other_opening': 0.0, 'recovery': 0.0, 't_ref': None, 'wind': wind} readings = {z: t for z, t in temps.items() if not (k in missing_end_at and z == zones[0])} evaluator.observe(START + timedelta(seconds=300 * k), readings, 5.0, 0.5, 0.5 if k in lag_uncertain_at else 0.0, {z: BASE for z in zones}, models={z: snapshot for z in zones}, inputs={z: {'internal': internal, 'ventilation': vent if z == zones[0] else {**vent, 'opening': actual[k]}} for z in zones}) return evaluator.summary(list(zones))[15] class IncompleteInputTests(unittest.TestCase): """Unknown ventilation or extended inputs never become Model Error samples.""" def test_window_unknown_for_the_whole_span_is_incomplete(self): stats = ventilated([0.0, None, None, None]) self.assertEqual((stats['samples'], stats['mae'], stats['incomplete']), (0, None, 1)) # The base model needs no window input: its error is still kept, apart. self.assertGreater(stats['incomplete_base_model_mae'], 0.1) def test_window_unknown_halfway_or_recovering_during_the_span(self): self.assertEqual(ventilated([0.0, 1.0, None, 1.0])['incomplete'], 1) # unavailable halfway stats = ventilated([0.0, None, 1.0, 1.0, 1.0, 1.0, 1.0]) # back after one poll # Spans ending at polls 3-6: only the first contains the unknown interval. self.assertEqual((stats['incomplete'], stats['samples']), (1, 3)) self.assertAlmostEqual(stats['mae'], 0, places=6) # the known spans are exact def test_one_zone_unknown_leaves_the_other_zone_scored(self): stats = ventilated([0.0, None, None, None], zones=('a', 'b')) self.assertEqual((stats['rooms']['a']['incomplete'], stats['rooms']['a']['samples']), (1, 0)) self.assertEqual((stats['rooms']['b']['incomplete'], stats['rooms']['b']['samples']), (0, 1)) def test_missing_wind_matters_only_while_a_window_is_open(self): features = ('vent_natural', 'vent_wind') open_span = ventilated([0.0, 1.0, 1.0, 1.0], wind=None, features=features) closed_span = ventilated([0.0, 0.0, 0.0, 0.0], wind=None, features=features) self.assertEqual((open_span['incomplete'], open_span['samples']), (1, 0)) self.assertEqual((closed_span['incomplete'], closed_span['samples']), (0, 1)) self.assertEqual(ventilated([0.0, 1.0, 1.0, 1.0], features=features)['samples'], 1) def test_unknown_electrical_heat_makes_the_extended_replay_incomplete(self): extended = (['loss', 'heat', 'bias', 'internal'], BASE + (0.15,)) stats = ventilated([0.0] * 4, extended=extended, internal=None) self.assertEqual((stats['incomplete'], stats['samples']), (1, 0)) self.assertIsNotNone(stats['incomplete_base_model_mae']) self.assertEqual(ventilated([0.0] * 4, extended=extended, internal=0.0)['samples'], 1) def test_a_neighbour_with_unknown_inputs_spoils_a_coupled_zone(self): coupled = {'base': BASE, 'uses_extended': True, 'ventilation': None, 'extended': (['loss', 'heat', 'bias', 'couple:a'], BASE + (0.01,))} plain = {'base': BASE, 'uses_extended': False, 'extended': None, 'ventilation': (['vent_natural'], (0.02,))} origin = {'temps': {'a': 20.0, 'b': 20.0}, 'outside': 5.0, 'models': {'a': plain, 'b': coupled}} path = [origin] + [{'time': START + timedelta(seconds=300 * k), 'lag': 0.5, 'inputs': {'a': {'ventilation': {'opening': None}}, 'b': {}}} for k in range(4)] origin['time'] = START - timedelta(seconds=300) _, incomplete = ev.replay_zones(origin, path) self.assertEqual(incomplete, {'a', 'b'}) path = [origin] + [{**poll, 'inputs': {'a': {'ventilation': {'opening': 0.0}}, 'b': {}}} for poll in path[1:]] self.assertEqual(ev.replay_zones(origin, path)[1], set()) def test_skipped_spans_are_counted_and_samples_stay_correct(self): stats = ventilated([0.0] * 9, lag_uncertain_at=(4,)) # Spans ending at 3..8: those crossing poll 4 (ending 4, 5, 6) are skipped. self.assertEqual((stats['samples'], stats['skipped'], stats['incomplete']), (3, 3, 0)) stats = ventilated([0.0] * 5, missing_end_at=(4,), zones=('a', 'b')) self.assertEqual((stats['rooms']['a']['samples'], stats['rooms']['a']['skipped']), (1, 1)) self.assertEqual(stats['rooms']['b']['samples'], 2) def ledger_with(plan, on_seconds, temps=None, uncertain=None, polls=None): """One forecast at T0; `on_seconds[k-1]` is the relay's on-time in the interval ending at poll k (None: unknown). Returns the ledger.""" ledger = ev.OperationalForecastLedger() trajectory = [20.0 + 0.1 * (b + 1) for b in range(len(plan))] ledger.observe(T0, {'r': 20.0}, None, None) ledger.record(T0, {'r': 20.0}, {'r': trajectory}, plan, 'rev1', {'strategy': 'mpc_plan', 'uncertain_inputs': uncertain or []}) for k in range(1, (polls or len(on_seconds)) + 1): on = on_seconds[k - 1] if k - 1 < len(on_seconds) else 0.0 ledger.observe(T0 + 300 * k, {'r': (temps or (lambda k: 20.0 + k / 30))(k)}, on, 300.0) return ledger def verdict(ledger, horizon): stats = ledger.summary(['r'])[horizon] for name in ('matched', 'approximate', 'counterfactual'): if stats[name]: return name return next(f'incomplete:{r}' for r, n in stats['incomplete_reasons'].items() if n) class PlanMatchTests(unittest.TestCase): """Forecasts are compared with reality only when the relay did what their plan assumed.""" ON, OFF = 300.0, 0.0 def test_exact_match_is_scored(self): ledger = ledger_with([1, 1, 0, 0], [self.ON] * 6 + [self.OFF] * 6) self.assertEqual([verdict(ledger, h) for h in (15, 30, 60)], ['matched'] * 3) self.assertEqual(ledger.deviation({'time': T0, 'plan': [1, 1, 0, 0]}, 60), (0.0, 0.0)) stats = ledger.summary(['r'])[60] self.assertEqual((stats['samples'], stats['approximate_stats']['samples']), (1, 0)) def test_small_transition_jitter_still_matches(self): # Planned off at 30 minutes; the relay switched off 40 s later. ledger = ledger_with([1, 1, 0, 0], [self.ON] * 6 + [40.0] + [self.OFF] * 5) self.assertEqual(ledger.deviation({'time': T0, 'plan': [1, 1, 0, 0]}, 60), (40.0, 0.0)) self.assertEqual(verdict(ledger, 60), 'matched') def test_twelve_of_fifteen_planned_minutes_is_only_approximate(self): ledger = ledger_with([1, 0], [self.ON, self.ON, 120.0, self.OFF, self.OFF, self.OFF]) self.assertEqual(verdict(ledger, 15), 'approximate') stats = ledger.summary(['r'])[15] # Not in the main figures; kept apart. self.assertEqual((stats['samples'], stats['mae'], stats['approximate']), (0, None, 1)) self.assertEqual(stats['approximate_stats']['samples'], 1) def test_same_on_time_at_another_time_is_counterfactual(self): # 15 minutes on as planned, but in the second block instead of the first. ledger = ledger_with([1, 0, 0, 0], [self.OFF] * 3 + [self.ON] * 3 + [self.OFF] * 6) self.assertEqual(ledger.deviation({'time': T0, 'plan': [1, 0, 0, 0]}, 30), (1800.0, 0.0)) self.assertEqual([verdict(ledger, h) for h in (15, 30, 60)], ['counterfactual'] * 3) def test_early_shutdown_and_unexpected_activation(self): early = ledger_with([1, 1, 1, 1], [self.ON] * 4 + [self.OFF] * 8) # off after 20 of 60 minutes self.assertEqual([verdict(early, h) for h in (15, 30, 60)], ['matched', 'counterfactual', 'counterfactual']) brief = ledger_with([0, 0, 0, 0], [self.OFF] * 4 + [240.0] + [self.OFF] * 7) # 4 minutes on self.assertEqual([verdict(brief, h) for h in (15, 30)], ['matched', 'approximate']) long = ledger_with([0, 0, 0, 0], [self.OFF] * 4 + [self.ON, self.ON] + [self.OFF] * 6) self.assertEqual(verdict(long, 30), 'counterfactual') def test_several_cycles_followed_closely_match(self): plan = [1, 0, 1, 0, 1, 0, 1, 0] relay = [] for block in plan: relay += [self.ON] * 3 if block else [30.0, self.OFF, self.OFF] # each off 30 s late ledger = ledger_with(plan, relay) self.assertEqual(ledger.deviation({'time': T0, 'plan': plan}, 120)[0], 120.0) # four transitions self.assertEqual(verdict(ledger, 60), 'matched') self.assertEqual(verdict(ledger, 120), 'matched') def test_missing_relay_history_is_incomplete_not_matched(self): ledger = ledger_with([0, 0, 0, 0], [self.OFF, None] + [self.OFF] * 10) self.assertEqual([verdict(ledger, h) for h in (15, 60)], ['incomplete:relay_history'] * 2) # A plan already contradicted is counterfactual whatever the gap hides. ledger = ledger_with([1, 1, 1, 1], [None] + [self.OFF] * 11) self.assertEqual(verdict(ledger, 30), 'counterfactual') def test_thermostat_disagreeing_is_counterfactual_never_an_error(self): ledger = ledger_with([1] * 4, [self.OFF] * 12, temps=lambda k: 20.0) stats = ledger.summary(['r']) self.assertEqual([stats[h]['counterfactual'] for h in (15, 30, 60)], [1, 1, 1]) self.assertEqual([stats[h]['samples'] for h in (15, 30, 60)], [0, 0, 0]) def test_forecasts_made_with_assumed_inputs_are_incomplete(self): ledger = ledger_with([0] * 4, [self.OFF] * 12, uncertain=['window: Bedroom']) self.assertEqual(verdict(ledger, 15), 'incomplete:uncertain_inputs') self.assertEqual(ledger.summary(['r'])[15]['samples'], 0) def test_counts_over_several_forecasts(self): ledger = ev.OperationalForecastLedger() ledger.observe(T0, {'r': 20.0}, None, None) for k, plan in enumerate(([0, 0], [1, 1], [0, 0])): # one forecast per poll, relay always off ledger.record(T0 + 300 * k, {'r': 20.0}, {'r': [20.0, 20.0]}, plan, 'rev1', {}) ledger.observe(T0 + 300 * (k + 1), {'r': 20.0}, 0.0, 300.0) for k in range(4, 12): ledger.observe(T0 + 300 * k, {'r': 20.0}, 0.0, 300.0) stats = ledger.summary(['r'])[15] self.assertEqual((stats['matched'], stats['counterfactual'], stats['samples'], stats['incomplete']), (2, 1, 2, 0)) restored = ev.OperationalForecastLedger(ledger.as_dict()) self.assertEqual(restored.summary(['r']), ledger.summary(['r']))