Make LoadML and temperature coverage follow forecast_hours

apps/predbat/load_ml_component.py

@@ -40,7 +40,7 @@
 
 class LoadMLComponent(ComponentBase):
     """
-    ML Load Forecaster component that predicts household load for the next 48 hours.
+    ML Load Forecaster component that predicts household load for the configured forecast hours.
 
     This component:
     - Fetches load history from configured sensor
@@ -418,6 +418,12 @@ async def _fetch_load_data(self):
                     step_energy = pv_forecast_minute.get(minute, 0.0)
                     pv_data[-(minute - self.minutes_now)] = dp4(step_energy)
 
+            # Future input coverage is independent of the available training history.
+            # minute_data bounds both directions, so retain the whole forecast even
+            # when the load or PV sensor only has a few days of history.
+            forecast_hours = max(int(self.get_arg("forecast_hours", 48)), 24)
+            future_input_days = max(days_to_fetch, (forecast_hours + 23) // 24)
+
             # Temperature predictions
             temp_entity = "sensor." + self.prefix + "_temperature"
             temperature_info = self.get_state_wrapper(temp_entity, attribute="results")
@@ -431,7 +437,7 @@ async def _fetch_load_data(self):
                 # We also get the last N days in the past to help the model learn the daily pattern
                 temperature_data, _ = minute_data(
                     data_array,
-                    days_to_fetch,
+                    future_input_days,
                     self.now_utc,
                     "state",
                     "last_updated",
@@ -454,7 +460,7 @@ async def _fetch_load_data(self):
                     data_array.append({"state": value, "last_updated": key})
                 import_rates_data, _ = minute_data(
                     data_array,
-                    days_to_fetch,
+                    future_input_days,
                     self.now_utc,
                     "state",
                     "last_updated",
@@ -472,7 +478,7 @@ async def _fetch_load_data(self):
                     data_array.append({"state": value, "last_updated": key})
                 export_rates_data, _ = minute_data(
                     data_array,
-                    days_to_fetch,
+                    future_input_days,
                     self.now_utc,
                     "state",
                     "last_updated",
@@ -688,11 +694,13 @@ def _get_predictions(self, now_utc, midnight_utc, exog_features=None):
                 import_rates=self.import_rates_data,
                 export_rates=self.export_rates_data,
                 exog_features=exog_features,
+                forecast_hours=self.get_arg("forecast_hours", 48),
             )
 
             if predictions:
                 self.current_predictions = predictions
-                self.log("ML Component: Generated {} predictions (total {:.2f} kWh over 48h)".format(len(predictions), max(predictions.values()) if predictions else 0))
+                prediction_hours = len(predictions) * PREDICT_STEP / 60.0
+                self.log("ML Component: Generated {} predictions (total {:.2f} kWh over {:g}h)".format(len(predictions), max(predictions.values()), prediction_hours))
             else:
                 self.log("ML Component: Predictor returned no predictions, returning previous dict")
 
@@ -1033,7 +1041,7 @@ async def _do_training(self, is_initial):
 
                 # Save model (fast - no lock needed)
                 if self.model_filepath:
-                    self.predictor.save(self.model_filepath)
+                    self.predictor.save(self.model_filepath, forecast_hours=self.get_arg("forecast_hours", 48))
             else:
                 self.log("Warn: ML Component: Training failed")
 

apps/predbat/load_predictor.py

@@ -15,7 +15,7 @@
 
 Implements a 3-hidden-layer feed-forward network ([512, 256, 64]) trained via
 AdamW optimiser with cosine LR decay and Huber loss. Uses autoregressive
-prediction for 48-hour load forecasts at CHUNK_MINUTES (5-min) resolution.
+prediction for configurable load forecasts at CHUNK_MINUTES (5-min) resolution.
 Inputs: historical load, PV generation, temperature, import/export rates,
 and cyclical time features (minute-of-day, day-of-week, day-of-year).
 """
@@ -32,7 +32,7 @@
 CHUNK_STEPS = CHUNK_MINUTES // STEP_MINUTES  # 5-min steps aggregated per chunk (3)
 LOOKBACK_STEPS = 24 * (60 // CHUNK_MINUTES)  # 24 hours at CHUNK_MINUTES resolution
 OUTPUT_STEPS = 1  # Single step output (autoregressive)
-PREDICT_HORIZON = 48 * (60 // CHUNK_MINUTES)  # 48 hours of predictions (96 * 30 min)
+PREDICT_HORIZON = 48 * (60 // CHUNK_MINUTES)  # Default prediction horizon and fixed 48-hour blend reference
 HIDDEN_SIZES = [512, 256, 64]  # Deeper network with more capacity
 BATCH_SIZE = 128  # Batch size
 MAX_BATCHES_PER_EPOCH = 200  # Cap on SGD batches per epoch - with importance sampling this keeps training
@@ -177,7 +177,7 @@ class LoadPredictor:
     """
     Lightweight MLP-based load predictor using NumPy only.
 
-    Predicts household electrical load for the next 48 hours using:
+    Predicts household electrical load for the configured forecast horizon using:
     - Historical load data (lookback window)
     - Cyclical time encodings (hour-of-day, day-of-week)
     - Placeholder for future exogenous features (temperature, solar)
@@ -200,6 +200,7 @@ def __init__(self, log_func=None, learning_rate=0.001, max_load_kw=23.0, weight_
         self.max_load_kw = max_load_kw
         self.weight_decay = weight_decay
         self.dropout_rate = dropout_rate
+        self.predict_horizon = PREDICT_HORIZON
 
         # Model weights (initialized on first train)
         self.weights = None
@@ -1651,9 +1652,9 @@ def train_curriculum(
         self.log("ML Predictor: Curriculum training complete, final val_mae={}".format("{:.4f} kWh".format(val_mae) if val_mae is not None else "None (all passes failed)"))
         return val_mae
 
-    def predict(self, load_minutes, now_utc, midnight_utc, pv_minutes=None, temp_minutes=None, import_rates=None, export_rates=None, exog_features=None):
+    def predict(self, load_minutes, now_utc, midnight_utc, pv_minutes=None, temp_minutes=None, import_rates=None, export_rates=None, exog_features=None, forecast_hours=48):
         """
-        Generate predictions for the next 48 hours using autoregressive approach.
+        Generate predictions for the requested forecast hours using autoregressive approach.
 
         Raw 5-min data is aggregated into CHUNK_MINUTES aligned chunks. Each
         prediction step predicts one CHUNK_MINUTES chunk, then feeds that back
@@ -1671,6 +1672,7 @@ def predict(self, load_minutes, now_utc, midnight_utc, pv_minutes=None, temp_min
             import_rates: Dict of {minute: rate_per_kwh}
             export_rates: Dict of {minute: rate_per_kwh}
             exog_features: Optional dict with future exogenous data
+            forecast_hours: Forecast duration in hours (default 48, minimum 24, matching the planner)
 
         Returns:
             Dict of {minute: cumulative_kwh} keyed at STEP_MINUTES (5-min) intervals.
@@ -1693,6 +1695,10 @@ def predict(self, load_minutes, now_utc, midnight_utc, pv_minutes=None, temp_min
             self.log("Warn: ML Predictor: No load data available for prediction")
             return {}
 
+        # Match the planner's existing whole-hour conversion and 24-hour minimum.
+        predict_horizon = max(int(forecast_hours), 24) * (60 // CHUNK_MINUTES)
+        self.predict_horizon = predict_horizon
+
         # Compute historical daily patterns at CHUNK_MINUTES resolution for blending
         # Returns 7 DOW-specific patterns (falling back to global when data is sparse)
         daily_patterns = self._compute_daily_pattern(energy_per_step, now_utc)
@@ -1738,7 +1744,7 @@ def predict(self, load_minutes, now_utc, midnight_utc, pv_minutes=None, temp_min
             pattern = daily_patterns.get(dow, {})
             max_energy_by_dow[dow] = max(pattern.values()) if pattern else max_kwh_per_chunk
 
-        # Blending: model weight decreases linearly from 1.0 to blend_floor
+        # Preserve the original blend through 48 hours, then keep blend_floor.
         blend_floor = 0.5
 
         # Seed previous value from the most recent historical chunk for the step-to-step cap
@@ -1761,9 +1767,9 @@ def predict(self, load_minutes, now_utc, midnight_utc, pv_minutes=None, temp_min
         last_export_rate = export_rate_buffer[0] if export_rate_buffer else 0.0
         exog_carried_steps = 0
 
-        self.log("ML Predictor: Starting autoregressive prediction loop for {} steps ({} hours), prev_energy value {} max_dow {}".format(PREDICT_HORIZON, PREDICT_HORIZON * CHUNK_MINUTES / 60, prev_energy_value, max_energy_by_dow))
+        self.log("ML Predictor: Starting autoregressive prediction loop for {} steps ({} hours), prev_energy value {} max_dow {}".format(predict_horizon, predict_horizon * CHUNK_MINUTES / 60, prev_energy_value, max_energy_by_dow))
 
-        for step_idx in range(PREDICT_HORIZON):
+        for step_idx in range(predict_horizon):
             # Target time: start of the step_idx-th future chunk (newer/recent edge).
             # Uses step_idx (not step_idx+1) to match training, where time features
             # represent the chunk's newer boundary (now_utc - alignment_offset -
@@ -1820,8 +1826,8 @@ def predict(self, load_minutes, now_utc, midnight_utc, pv_minutes=None, temp_min
             slot = (minute_of_day // CHUNK_MINUTES) * CHUNK_MINUTES
             hist_value = daily_patterns[day_of_week].get(slot, model_pred)
 
-            # Linear blend: 100% model at step 0, blend_floor% model at horizon
-            progress = step_idx / PREDICT_HORIZON
+            # Keep the blend independent of the configured forecast length.
+            progress = min(step_idx / PREDICT_HORIZON, 1.0)
             model_weight = 1.0 - progress * (1.0 - blend_floor)
             energy_value = model_weight * model_pred + (1.0 - model_weight) * hist_value
 
@@ -1852,7 +1858,7 @@ def predict(self, load_minutes, now_utc, midnight_utc, pv_minutes=None, temp_min
         # Only worth reporting when there was exogenous data to carry; a setup with no
         # temperature or rate sensors legitimately runs the whole rollout without any
         if exog_carried_steps and (temp_values or import_rate_values or export_rate_values):
-            self.log("ML Predictor: Forward temperature/rate forecast ran out for {} of {} rollout steps, last known values carried forward".format(exog_carried_steps, PREDICT_HORIZON * CHUNK_STEPS))
+            self.log("ML Predictor: Forward temperature/rate forecast ran out for {} of {} rollout steps, last known values carried forward".format(exog_carried_steps, predict_horizon * CHUNK_STEPS))
 
         # Convert to cumulative kWh format at STEP_MINUTES resolution.
         # Each CHUNK_MINUTES prediction is split into CHUNK_STEPS equal 5-min sub-steps
@@ -1861,7 +1867,7 @@ def predict(self, load_minutes, now_utc, midnight_utc, pv_minutes=None, temp_min
         result = {}
         cumulative = 0
 
-        for step_idx in range(PREDICT_HORIZON):
+        for step_idx in range(predict_horizon):
             energy = predictions_energy[step_idx]
             energy_per_substep = energy / CHUNK_STEPS
             for j in range(CHUNK_STEPS):
@@ -1871,24 +1877,26 @@ def predict(self, load_minutes, now_utc, midnight_utc, pv_minutes=None, temp_min
 
         return result
 
-    def save(self, filepath):
+    def save(self, filepath, forecast_hours=None):
         """
         Save model to file.
 
         Args:
             filepath: Path to save model (without extension)
+            forecast_hours: Configured forecast hours, or None to retain the last prediction horizon
         """
         if not self.model_initialized:
             self.log("Warn: ML Predictor: No model to save")
             return False
 
         try:
             # Prepare metadata
+            predict_horizon = self.predict_horizon if forecast_hours is None else max(int(forecast_hours), 24) * (60 // CHUNK_MINUTES)
             metadata = {
                 "model_version": MODEL_VERSION,
                 "lookback_steps": LOOKBACK_STEPS,
                 "output_steps": OUTPUT_STEPS,
-                "predict_horizon": PREDICT_HORIZON,
+                "predict_horizon": predict_horizon,
                 "hidden_sizes": HIDDEN_SIZES,
                 "training_timestamp": self.training_timestamp.isoformat() if self.training_timestamp else None,
                 "validation_mae": float(self.validation_mae) if self.validation_mae else None,

apps/predbat/temperature.py

@@ -17,7 +17,8 @@
 
 import aiohttp
 import asyncio
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
+from urllib.parse import urlsplit, urlunsplit, parse_qsl, urlencode
 from utils import dp1
 from component_base import ComponentBase
 from predbat_metrics import record_api_call
@@ -39,6 +40,8 @@ def initialize(self, temperature_enable, temperature_latitude, temperature_longi
         self.temperature_data = None
         self.last_updated_timestamp = None
         self.failures_total = 0
+        self.temperature_forecast_minutes = None
+        self.temperature_horizon_warned = None
 
     async def select_event(self, entity_id, value):
         pass
@@ -74,24 +77,33 @@ async def run(self, seconds, first):
                 self.update_success_timestamp()
                 return True
 
-            fetch_due = first or (seconds % (60 * 60) == 0)
+            forecast_minutes = max(int(self.get_arg("forecast_hours", 48)), 24) * 60
+            horizon_changed = self.temperature_forecast_minutes is not None and forecast_minutes != self.temperature_forecast_minutes
+            fetch_due = first or horizon_changed or (seconds % (60 * 60) == 0)
 
             if first:
                 # Restore cached data; skip API fetch if it is less than an hour old
                 await self.load_temperature_cache()
                 if self.last_updated_timestamp is not None:
                     age_minutes = (datetime.now() - self.last_updated_timestamp).total_seconds() / 60
-                    if age_minutes < 60:
+                    if age_minutes < 60 and not horizon_changed and self.cached_forecast_covers(forecast_minutes):
                         fetch_due = False
 
             if fetch_due:
+                # Remember the attempt so failures retain the normal hourly retry cadence.
+                self.temperature_forecast_minutes = forecast_minutes
                 temperature_data = await self.fetch_temperature_data()
                 if temperature_data is not None:
                     self.temperature_data = temperature_data
+                    self.temperature_forecast_minutes = forecast_minutes
                     self.last_updated_timestamp = datetime.now()
                     self.publish_temperature_sensor()
                     await self.save_temperature_cache()
+                elif self.temperature_data is not None:
+                    # Publish retained data without treating the failed fetch as fresh or successful.
+                    self.publish_temperature_sensor()
             elif self.temperature_data is not None:
+                self.temperature_forecast_minutes = forecast_minutes
                 self.update_success_timestamp()
                 self.publish_temperature_sensor()
         except Exception as e:
@@ -136,7 +148,55 @@ def build_api_url(self, latitude, longitude):
         Build the API URL with latitude and longitude placeholders replaced
         """
         url = self.temperature_url.replace("LATITUDE", str(latitude)).replace("LONGITUDE", str(longitude))
-        return url
+        parts = urlsplit(url)
+        hostname = parts.hostname or ""
+        if not (hostname == "open-meteo.com" or hostname.endswith(".open-meteo.com")) or parts.path != "/v1/forecast":
+            return url
+
+        query = parse_qsl(parts.query, keep_blank_values=True)
+        keys = {key for key, _value in query}
+        if keys & {"start_date", "end_date", "start_hour", "end_hour"}:
+            # Explicit date ranges are deliberate overrides of the standard forecast window.
+            return url
+
+        forecast_minutes = max(int(self.get_arg("forecast_hours", 48)), 24) * 60
+        # Open-Meteo starts at midnight, so include an extra calendar day.
+        requested_days = max(7, (forecast_minutes + 1439) // 1440 + 1)
+        forecast_days = min(requested_days, 16)
+        if requested_days > 16:
+            if self.temperature_horizon_warned != forecast_minutes:
+                self.log("Warn: TemperatureAPI: Requested {}h rolling forecast may exceed Open-Meteo's 16 calendar days; missing temperature steps will use the predictor's last-known-value fallback".format(forecast_minutes / 60.0))
+                self.temperature_horizon_warned = forecast_minutes
+        else:
+            self.temperature_horizon_warned = None
+
+        if requested_days <= 7 and not (keys & {"forecast_days", "forecast_hours"}):
+            return url
+        query = [(key, value) for key, value in query if key not in {"forecast_days", "forecast_hours"}]
+        query.append(("forecast_days", str(forecast_days)))
+        return urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(query), parts.fragment))
+
+    def cached_forecast_covers(self, forecast_minutes):
+        """Check available cached temperature coverage before skipping the first fetch."""
+        times = (self.temperature_data or {}).get("hourly", {}).get("time", [])
+        if not times:
+            # Retain the existing cache policy for payloads without hourly timestamps.
+            return True
+        try:
+            latest = datetime.fromisoformat(times[-1])
+            if latest.tzinfo is None:
+                offset = self.temperature_data.get("utc_offset_seconds", 0)
+                latest = latest.replace(tzinfo=timezone(timedelta(seconds=offset)))
+            now = getattr(self, "now_utc", datetime.now(timezone.utc))
+            if now.tzinfo is None:
+                now = now.replace(tzinfo=timezone.utc)
+            now = now.astimezone(timezone.utc)
+            # Do not demand coverage that the provider's calendar-day limit cannot supply.
+            provider_end = now.astimezone(latest.tzinfo).replace(hour=0, minute=0, second=0, microsecond=0) + timedelta(days=16, hours=-1)
+            required_end = min(now + timedelta(minutes=forecast_minutes), provider_end)
+            return latest >= required_end
+        except (ValueError, TypeError, OverflowError):
+            return False
 
     def convert_timezone_offset(self, utc_offset_seconds):
         """