Skip to content

Commit b0f1fb8

Browse files
mbuhansenclaude
andauthored
fix(load-ml): stop a training run that crosses midnight inflating the forecast (#4537)
The load baseline (load_minutes_now, cumulative load since local midnight) is captured during the data fetch, while minutes_now is read live from the base at publish time. Training runs for many minutes, so when a fine-tune starts before midnight and finishes after it the two disagree: minutes_now has reset to the new day but the baseline still holds yesterday's full daily total, which then gets added on top of every published forecast point. Seen in the wild as a single-sample spike in the ML load chart - a fetch at 23:50 followed by training that completed at 00:15 published load_today_h1 of ~25 kWh and load_today_h8 of ~30 kWh against an actual daily total of 25 kWh. The next cycle re-fetched and the values returned to normal. Re-fetch after training whenever the data has gone stale, which re-anchors both the baseline and the lookback window feeding the prediction to the current time. As a second line of defence, detect a baseline snapshot belonging to a previous day at publish time and re-derive it from the per-step load history instead. Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
1 parent 5fd3389 commit b0f1fb8

2 files changed

Lines changed: 227 additions & 11 deletions

File tree

‎apps/predbat/load_ml_component.py‎

Lines changed: 63 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -108,6 +108,10 @@ def initialize(self, load_ml_enable, load_ml_source=True, load_ml_max_days_histo
108108
self.data_lock = asyncio.Lock()
109109
self.last_data_fetch = None
110110
self.load_ml_calculating = False
111+
# Cumulative load since local midnight, captured at fetch time along with the timestamp
112+
# it refers to so a stale snapshot can be detected if publishing is delayed across midnight
113+
self.load_minutes_now = 0
114+
self.load_minutes_now_time = None
111115

112116
# Model state
113117
self.predictor = None
@@ -589,9 +593,27 @@ def _merge_fetch_data(self, load_data, age_days, load_minutes_now, pv_data, temp
589593
self.export_rates_data = self._merge_channel(self.export_rates_data, export_rates_data) if export_rates_data else self.export_rates_data
590594
self._recompute_age_days()
591595
self.load_minutes_now = load_minutes_now
596+
self.load_minutes_now_time = self.now_utc
592597
self.last_data_fetch = self.now_utc
593598
self.data_ready = True
594599

600+
async def _do_fetch(self):
601+
"""
602+
Shift the in-memory data to 'now', fetch fresh sensor history and merge the two.
603+
604+
Re-anchors every channel plus the load_minutes_now baseline to the current time,
605+
so anything computed afterwards (predictions, published entity) uses data that
606+
belongs to the same moment as self.minutes_now.
607+
"""
608+
async with self.data_lock:
609+
self._shift_fetch_data()
610+
load_data, age_days, load_minutes_now, pv_data, temperature_data, import_rates_data, export_rates_data = await self._fetch_load_data()
611+
if load_data:
612+
self._merge_fetch_data(load_data, age_days, load_minutes_now, pv_data, temperature_data, import_rates_data, export_rates_data)
613+
else:
614+
self.log("Warn: ML Component: Failed to fetch load data, no data was returned.")
615+
self.log("ML Component: Data fetch completed, load data age {:.1f} days, {} data points".format(self.load_data_age_days, len(self.load_data) if self.load_data else 0))
616+
595617
def get_current_prediction(self):
596618
"""
597619
Returns the current ML load predictions.
@@ -702,14 +724,7 @@ async def run(self, seconds, first):
702724

703725
if should_fetch:
704726
# Shift existing data to keep it anchored to 'minutes ago from now' before fetching new data, so that we can merge them and preserve historical depth even if the fetch returns limited history
705-
async with self.data_lock:
706-
self._shift_fetch_data()
707-
load_data, age_days, load_minutes_now, pv_data, temperature_data, import_rates_data, export_rates_data = await self._fetch_load_data()
708-
if load_data:
709-
self._merge_fetch_data(load_data, age_days, load_minutes_now, pv_data, temperature_data, import_rates_data, export_rates_data)
710-
else:
711-
self.log("Warn: ML Component: Failed to fetch load data, no data was returned.")
712-
self.log("ML Component: Data fetch completed, load data age {:.1f} days, {} data points".format(self.load_data_age_days, len(self.load_data) if self.load_data else 0))
727+
await self._do_fetch()
713728

714729
# Check if we have data
715730
if not self.data_ready:
@@ -751,6 +766,17 @@ async def run(self, seconds, first):
751766
self.log("ML Component: Doing training...")
752767
await self._do_training(is_initial)
753768

769+
# Training can run for many minutes, during which the data fetched above
770+
# goes stale - both the lookback window feeding the prediction and the
771+
# load_minutes_now baseline still refer to the pre-training time. Re-fetch
772+
# so everything is anchored to the current time again. Without this a
773+
# training run that straddles midnight publishes yesterday's daily total
774+
# as today's baseline, inflating the forecast by a whole day of load.
775+
stale_minutes = (self.now_utc - self.last_data_fetch).total_seconds() / 60.0 if self.last_data_fetch else 0
776+
if stale_minutes >= PREDICT_STEP:
777+
self.log("ML Component: Data is {:.0f} minutes stale after training, re-fetching before prediction".format(stale_minutes))
778+
await self._do_fetch()
779+
754780
# Update model validity status
755781
self._update_model_status()
756782

@@ -1007,6 +1033,31 @@ def _update_model_status(self):
10071033
self.model_valid = False
10081034
self.model_status = "fallback_" + reason
10091035

1036+
def _load_baseline_now(self):
1037+
"""
1038+
Return the cumulative load since local midnight to use as the forecast baseline.
1039+
1040+
Normally this is the snapshot captured during the last data fetch. If that snapshot
1041+
was taken on a previous day then the daily counter has since reset, and carrying it
1042+
forward would add a whole day of load on top of every forecast point, so in that case
1043+
it is re-derived from the per-step load history instead.
1044+
1045+
Returns:
1046+
Cumulative load in kWh since local midnight
1047+
"""
1048+
if self.load_minutes_now_time is None or self.load_minutes_now_time.date() == self.now_utc.date():
1049+
return self.load_minutes_now
1050+
1051+
derived_baseline = 0.0
1052+
if self.load_data:
1053+
for minute in range(0, self.minutes_now, PREDICT_STEP):
1054+
derived_baseline += self.load_data.get(minute, 0.0)
1055+
derived_baseline = dp4(derived_baseline)
1056+
self.log(
1057+
"Warn: ML Component: Load baseline of {} kWh was captured on {} which is a previous day, re-derived load so far today as {} kWh".format(dp2(self.load_minutes_now), self.load_minutes_now_time.strftime("%Y-%m-%d %H:%M"), dp2(derived_baseline))
1058+
)
1059+
return derived_baseline
1060+
10101061
def _publish_entity(self):
10111062
"""Publish the load_forecast_ml entity with current predictions."""
10121063
# Convert predictions to timestamp format for entity
@@ -1018,6 +1069,7 @@ def _publish_entity(self):
10181069
power_today_now = 0
10191070
power_today_h1 = 0
10201071
power_today_h8 = 0
1072+
load_minutes_now = self._load_baseline_now()
10211073
# Future predictions
10221074
if self.current_predictions:
10231075
prev_value = 0
@@ -1026,8 +1078,8 @@ def _publish_entity(self):
10261078
timestamp_str = timestamp.strftime(TIME_FORMAT)
10271079
# Reset at midnight
10281080
if minute > 0 and ((minute + self.minutes_now) % (24 * 60) == 0):
1029-
reset_amount = value + self.load_minutes_now
1030-
output_value = round(value - reset_amount + self.load_minutes_now, 4)
1081+
reset_amount = value + load_minutes_now
1082+
output_value = round(value - reset_amount + load_minutes_now, 4)
10311083
results[timestamp_str] = output_value
10321084
delta_value = (value - prev_value) / PREDICT_STEP * 60.0
10331085
if minute == 0:
@@ -1060,7 +1112,7 @@ def _publish_entity(self):
10601112
"sensor." + self.prefix + "_load_ml_stats",
10611113
state=round(total_kwh, 2),
10621114
attributes={
1063-
"load_today": dp2(self.load_minutes_now),
1115+
"load_today": dp2(load_minutes_now),
10641116
"load_today_h1": dp2(load_today_h1),
10651117
"load_today_h8": dp2(load_today_h8),
10661118
"load_total": dp2(total_kwh),

‎apps/predbat/tests/test_load_ml.py‎

Lines changed: 164 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,7 @@ def test_load_ml(my_predbat=None):
5959
("prediction_with_temp", _test_prediction_with_temp, "Prediction with temperature forecast data"),
6060
("component_fetch_load_data", _test_component_fetch_load_data, "LoadMLComponent _fetch_load_data method"),
6161
("component_publish_entity", _test_component_publish_entity, "LoadMLComponent _publish_entity method"),
62+
("component_stale_midnight_baseline", _test_component_stale_midnight_baseline, "LoadMLComponent baseline handling when publishing crosses midnight"),
6263
("car_subtraction_direct", _test_car_subtraction_direct, "Direct car_subtraction method with interpolation and smoothing"),
6364
("component_run_data_merge", _test_component_run_data_merge, "LoadMLComponent run() data fetch, save and merge across two runs"),
6465
("component_init_predictor_last_train_time", _test_component_init_predictor_sets_last_train_time, "LoadMLComponent _init_predictor sets last_train_time from embedded training_timestamp"),
@@ -2607,6 +2608,169 @@ def mock_dashboard_item(entity_id, state, attributes, app):
26072608
print(" All _publish_entity tests passed!")
26082609

26092610

2611+
def _test_component_stale_midnight_baseline():
2612+
"""Test that a load baseline captured on a previous day is not carried into today's forecast.
2613+
2614+
Reproduces the case where a long training run starts before local midnight and finishes
2615+
after it: load_minutes_now still holds yesterday's full daily total while minutes_now has
2616+
already reset, which used to add a whole day of load on top of every published value.
2617+
"""
2618+
import asyncio
2619+
from datetime import datetime, timezone
2620+
from load_ml_component import LoadMLComponent
2621+
from unittest.mock import MagicMock
2622+
from const import PREDICT_STEP
2623+
2624+
def run_async(coro):
2625+
"""Run a coroutine on the current or a fresh event loop."""
2626+
try:
2627+
loop = asyncio.get_event_loop()
2628+
except RuntimeError:
2629+
loop = asyncio.new_event_loop()
2630+
asyncio.set_event_loop(loop)
2631+
return loop.run_until_complete(coro)
2632+
2633+
class MockBase:
2634+
"""Minimal PredBat stand-in positioned at 00:15, just after local midnight."""
2635+
2636+
def __init__(self):
2637+
"""Set up the mock at 2026-01-02 00:15, 15 minutes into the new day."""
2638+
self.prefix = "predbat"
2639+
self.config_root = None
2640+
self.now_utc = datetime(2026, 1, 2, 0, 15, 0, tzinfo=timezone.utc)
2641+
self.midnight_utc = datetime(2026, 1, 2, 0, 0, 0, tzinfo=timezone.utc)
2642+
self.minutes_now = 15
2643+
self.local_tz = timezone.utc
2644+
self.args = {}
2645+
self.log_messages = []
2646+
self.dashboard_calls = []
2647+
2648+
def log(self, msg):
2649+
"""Record a log message."""
2650+
self.log_messages.append(msg)
2651+
2652+
def get_arg(self, key, default=None, indirect=True, combine=False, attribute=None, index=None, domain=None, can_override=True, required_unit=None):
2653+
"""Return canned config values for the keys the component reads."""
2654+
return {
2655+
"load_today": ["sensor.load_today"],
2656+
"load_power": None,
2657+
"car_charging_energy": None,
2658+
"load_scaling": 1.0,
2659+
"car_charging_energy_scale": 1.0,
2660+
}.get(key, default)
2661+
2662+
mock_base = MockBase()
2663+
component = LoadMLComponent(mock_base, load_ml_enable=True)
2664+
2665+
def mock_dashboard_item(entity_id, state, attributes, app):
2666+
"""Capture published entities."""
2667+
mock_base.dashboard_calls.append({"entity_id": entity_id, "state": state, "attributes": attributes, "app": app})
2668+
2669+
component.dashboard_item = mock_dashboard_item
2670+
2671+
# Baseline snapshot taken at 23:50 the previous day, holding that whole day's total
2672+
component.load_minutes_now = 24.9
2673+
component.load_minutes_now_time = datetime(2026, 1, 1, 23, 50, 0, tzinfo=timezone.utc)
2674+
# Per-step load history for the 15 minutes since midnight (0.05 kWh per 5 min = 0.15 kWh)
2675+
component.load_data = {0: 0.05, 5: 0.05, 10: 0.05, 15: 0.05, 20: 0.05}
2676+
2677+
component.current_predictions = {0: 0.05, 60: 0.6, 480: 5.0}
2678+
component.predictor.validation_mae = 0.5
2679+
component.predictor.get_model_age_hours = MagicMock(return_value=0.5)
2680+
component.last_train_time = mock_base.now_utc
2681+
component.load_data_age_days = 7.0
2682+
component.model_status = "active"
2683+
2684+
component._publish_entity()
2685+
2686+
attrs2 = mock_base.dashboard_calls[1]["attributes"]
2687+
2688+
# Baseline must be re-derived from the 15 minutes of load since midnight, not 24.9 kWh
2689+
assert abs(attrs2["load_today"] - 0.15) < 0.01, f"Expected load_today 0.15 re-derived since midnight, got {attrs2['load_today']}"
2690+
assert abs(attrs2["load_today_h1"] - 0.75) < 0.01, f"Expected load_today_h1 0.75 (0.6 + 0.15), got {attrs2['load_today_h1']}"
2691+
assert abs(attrs2["load_today_h8"] - 5.15) < 0.01, f"Expected load_today_h8 5.15 (5.0 + 0.15), got {attrs2['load_today_h8']}"
2692+
assert any("previous day" in msg for msg in mock_base.log_messages), "Stale baseline should be logged as a warning"
2693+
2694+
print(" ✓ Stale pre-midnight baseline re-derived from load history")
2695+
2696+
# Same-day snapshot must be used verbatim, no re-derivation
2697+
mock_base.dashboard_calls = []
2698+
component.load_minutes_now = 0.4
2699+
component.load_minutes_now_time = datetime(2026, 1, 2, 0, 10, 0, tzinfo=timezone.utc)
2700+
2701+
component._publish_entity()
2702+
2703+
attrs2 = mock_base.dashboard_calls[1]["attributes"]
2704+
assert abs(attrs2["load_today"] - 0.4) < 0.01, f"Expected load_today 0.4 from the snapshot, got {attrs2['load_today']}"
2705+
assert abs(attrs2["load_today_h1"] - 1.0) < 0.01, f"Expected load_today_h1 1.0 (0.6 + 0.4), got {attrs2['load_today_h1']}"
2706+
2707+
print(" ✓ Same-day baseline snapshot used unchanged")
2708+
2709+
# A missing snapshot timestamp (e.g. loaded from an older state) keeps the old behaviour
2710+
mock_base.dashboard_calls = []
2711+
component.load_minutes_now_time = None
2712+
2713+
component._publish_entity()
2714+
2715+
attrs2 = mock_base.dashboard_calls[1]["attributes"]
2716+
assert abs(attrs2["load_today"] - 0.4) < 0.01, f"Expected load_today 0.4 when no snapshot time is known, got {attrs2['load_today']}"
2717+
2718+
print(" ✓ Missing snapshot timestamp falls back to the stored baseline")
2719+
2720+
# Re-fetch after a long training run must re-anchor the baseline before predicting
2721+
async def run_stale_refetch():
2722+
"""Drive run() through a training cycle that leaves the fetched data stale."""
2723+
fetch_times = []
2724+
2725+
async def mock_fetch_load_data():
2726+
"""Return canned data and record when the fetch happened."""
2727+
fetch_times.append(mock_base.now_utc)
2728+
load_now = 24.9 if mock_base.now_utc.day == 1 else 0.15
2729+
return {0: 0.05, 5: 0.05}, 7.0, load_now, {}, {}, {}, {}
2730+
2731+
async def mock_do_training(is_initial):
2732+
"""Simulate a training run that takes 25 minutes and crosses midnight."""
2733+
mock_base.now_utc = datetime(2026, 1, 2, 0, 15, 0, tzinfo=timezone.utc)
2734+
mock_base.midnight_utc = datetime(2026, 1, 2, 0, 0, 0, tzinfo=timezone.utc)
2735+
mock_base.minutes_now = 15
2736+
2737+
component._fetch_load_data = mock_fetch_load_data
2738+
component._do_training = mock_do_training
2739+
component._update_model_status = MagicMock()
2740+
component.base.prediction_started = False
2741+
component.load_ml_database_days = 0
2742+
component.ml_min_days = 0
2743+
component.initial_training_done = True
2744+
component.model_valid = True
2745+
component.data_ready = True
2746+
component.last_train_time = datetime(2026, 1, 1, 21, 50, 0, tzinfo=timezone.utc)
2747+
component.predictor.predict = MagicMock(return_value={0: 0.05, 60: 0.6, 480: 5.0})
2748+
2749+
# Start the cycle at 23:50 the previous day, before training moves the clock
2750+
mock_base.now_utc = datetime(2026, 1, 1, 23, 50, 0, tzinfo=timezone.utc)
2751+
mock_base.midnight_utc = datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
2752+
mock_base.minutes_now = 1430
2753+
mock_base.dashboard_calls = []
2754+
2755+
await component.run(0, False)
2756+
2757+
assert len(fetch_times) == 2, f"Expected a second fetch after training, got {len(fetch_times)} fetches"
2758+
assert fetch_times[1].day == 2, f"Second fetch should happen after midnight, got {fetch_times[1]}"
2759+
assert component.load_minutes_now_time == mock_base.now_utc, "Baseline timestamp should be re-anchored to the post-training time"
2760+
assert abs(component.load_minutes_now - 0.15) < 0.01, f"Expected re-fetched baseline 0.15, got {component.load_minutes_now}"
2761+
2762+
attrs = mock_base.dashboard_calls[1]["attributes"]
2763+
assert abs(attrs["load_today_h8"] - 5.15) < 0.01, f"Expected load_today_h8 5.15 after re-fetch, got {attrs['load_today_h8']}"
2764+
2765+
run_async(run_stale_refetch())
2766+
2767+
print(" ✓ Stale data re-fetched after training before predicting")
2768+
2769+
assert PREDICT_STEP == 5, "Test assumes a 5 minute prediction step"
2770+
2771+
print(" All midnight baseline tests passed!")
2772+
2773+
26102774
def _test_car_subtraction_direct():
26112775
"""Test car_subtraction method directly with all improved features.
26122776

0 commit comments

Comments
 (0)