fix(ml): resolve NaN training loss on startup and preserve database zero entries

.cspell/custom-dictionary-workspace.txt

@@ -327,6 +327,8 @@ mtok
 mult
 myenergi
 mypy
+nanmean
+nanstd
 nattribute
 ncalls
 nearr

apps/predbat/load_ml_component.py

@@ -679,7 +679,16 @@ def _get_predictions(self, now_utc, midnight_utc, exog_features=None):
             self.log("ML Component: Generating predictions load data age {:.1f} days, {} data points".format(self.load_data_age_days, len(self.load_data) if self.load_data else 0))
             if 0:
                 self._log_prediction_input_table(now_utc)
-            predictions = self.predictor.predict(self.load_data, now_utc, midnight_utc, pv_minutes=self.pv_data, temp_minutes=self.temperature_data, import_rates=self.import_rates_data, export_rates=self.export_rates_data, exog_features=exog_features)
+            predictions = self.predictor.predict(
+                self.load_data,
+                now_utc,
+                midnight_utc,
+                pv_minutes=self.pv_data,
+                temp_minutes=self.temperature_data,
+                import_rates=self.import_rates_data,
+                export_rates=self.export_rates_data,
+                exog_features=exog_features,
+            )
 
             if predictions:
                 self.current_predictions = predictions
@@ -823,14 +832,20 @@ async def save_database_history(self):
         max_steps = self.load_ml_database_days * 24 * 60 // PREDICT_STEP
 
         def dict_to_array(data_dict):
-            arr = np.zeros(max_steps, dtype=np.float32)
+            """Convert sparse {minute: value} dictionary into a fixed-size numpy array."""
+            arr = np.full(max_steps, np.nan, dtype=np.float32)
             if data_dict:
                 for minute, value in data_dict.items():
                     # Only persist historical data (non-negative integer keys)
                     if isinstance(minute, int) and minute >= 0:
                         idx = minute // PREDICT_STEP
-                        if 0 <= idx < max_steps:
-                            arr[idx] = float(value)
+                        if 0 <= idx < max_steps and value is not None:
+                            try:
+                                val = float(value)
+                                if np.isfinite(val):
+                                    arr[idx] = val
+                            except (ValueError, TypeError):
+                                pass
             return arr
 
         total_steps = len(self.load_data) if self.load_data else 0
@@ -842,6 +857,7 @@ def dict_to_array(data_dict):
                 "step_minutes": PREDICT_STEP,
                 "n_steps": total_steps,
                 "age_days": float(self.load_data_age_days),
+                "nan_sentinel": True,
             }
         )
 
@@ -872,48 +888,52 @@ async def load_database_history(self):
 
         try:
             # np.load is blocking disk IO - run in executor so the event loop stays alive
-            data = np.load(self.database_filepath, allow_pickle=False)
-            metadata = json.loads(str(data["metadata_json"]))
-
-            version = metadata.get("version", 0)
-            if version != DATABASE_VERSION:
-                self.log("Warn: ML Component: Database version mismatch (saved={}, current={}), discarding".format(version, DATABASE_VERSION))
-                return
-
-            step_minutes = metadata.get("step_minutes", PREDICT_STEP)
-            if step_minutes != PREDICT_STEP:
-                self.log("Warn: ML Component: Database step size mismatch (saved={}min, current={}min), discarding".format(step_minutes, PREDICT_STEP))
-                return
-
-            saved_utc = datetime.fromisoformat(metadata["saved_utc"])
-            age_days = float(metadata.get("age_days", 0))
-
-            max_steps = self.load_ml_database_days * 24 * 60 // PREDICT_STEP
-
-            def array_to_dict(arr):
-                """Reconstruct a sparse {minute: value} dict with keys as stored"""
-                result = {}
-                for i in range(len(arr)):
-                    val = float(arr[i])
-                    if val != 0.0:
-                        result[i * PREDICT_STEP] = val
-                return result
-
-            self.load_data = array_to_dict(data["load"])
-            self.pv_data = array_to_dict(data["pv"])
-            self.temperature_data = array_to_dict(data["temp"])
-            self.import_rates_data = array_to_dict(data["import_rate"])
-            self.export_rates_data = array_to_dict(data["export_rate"])
-            self.load_data_age_days = age_days
-            # Restore last_data_fetch to the save time so that run()'s _shift() will
-            # compute the correct elapsed time and apply the shift exactly once.
-            self.last_data_fetch = saved_utc
-
-            if self.load_data:
-                self.data_ready = True
-
-            elapsed_minutes = (self.now_utc - saved_utc).total_seconds() / 60.0
-            self.log("ML Component: Loaded database history: {} load points, {:.1f}h elapsed since save, age {:.1f} days (shift will be applied on next fetch)".format(len(self.load_data), elapsed_minutes / 60.0, self.load_data_age_days))
+            with np.load(self.database_filepath, allow_pickle=False) as data:
+                metadata = json.loads(str(data["metadata_json"]))
+
+                version = metadata.get("version", 0)
+                if version != DATABASE_VERSION:
+                    self.log("Warn: ML Component: Database version mismatch (saved={}, current={}), discarding".format(version, DATABASE_VERSION))
+                    return
+
+                step_minutes = metadata.get("step_minutes", PREDICT_STEP)
+                if step_minutes != PREDICT_STEP:
+                    self.log("Warn: ML Component: Database step size mismatch (saved={}min, current={}min), discarding".format(step_minutes, PREDICT_STEP))
+                    return
+
+                saved_utc = datetime.fromisoformat(metadata["saved_utc"])
+                age_days = float(metadata.get("age_days", 0))
+                use_nan_sentinel = metadata.get("nan_sentinel", False)
+
+                def array_to_dict(arr):
+                    """Reconstruct a sparse {minute: value} dict with keys as stored"""
+                    result = {}
+                    nan_format = use_nan_sentinel or np.isnan(arr).any()
+                    for i in range(len(arr)):
+                        val = float(arr[i])
+                        if nan_format:
+                            if not np.isnan(val) and np.isfinite(val):
+                                result[i * PREDICT_STEP] = val
+                        else:
+                            if val != 0.0:
+                                result[i * PREDICT_STEP] = val
+                    return result
+
+                self.load_data = array_to_dict(data["load"])
+                self.pv_data = array_to_dict(data["pv"])
+                self.temperature_data = array_to_dict(data["temp"])
+                self.import_rates_data = array_to_dict(data["import_rate"])
+                self.export_rates_data = array_to_dict(data["export_rate"])
+                self.load_data_age_days = age_days
+                # Restore last_data_fetch to the save time so that run()'s _shift() will
+                # compute the correct elapsed time and apply the shift exactly once.
+                self.last_data_fetch = saved_utc
+
+                if self.load_data:
+                    self.data_ready = True
+
+                elapsed_minutes = (self.now_utc - saved_utc).total_seconds() / 60.0
+                self.log("ML Component: Loaded database history: {} load points, {:.1f}h elapsed since save, age {:.1f} days (shift will be applied on next fetch)".format(len(self.load_data), elapsed_minutes / 60.0, self.load_data_age_days))
 
         except Exception as e:
             self.log("Warn: ML Component: Failed to load database history: {} - {}".format(e, traceback.format_exc()))
@@ -978,22 +998,22 @@ async def _do_training(self, is_initial):
                     curriculum_step_days=5,
                     max_intermediate_passes=8,
                 )
-            # Even if initial was done we need to do one fine tuned curriculum pass too.
-            val_mae = self.predictor.train_curriculum(
-                load_data_snap,
-                now_utc_snap,
-                pv_minutes=pv_data_snap,
-                temp_minutes=temp_data_snap,
-                import_rates=import_rates_snap,
-                export_rates=export_rates_snap,
-                epochs=epochs,
-                time_decay_days=time_decay,
-                validation_holdout_hours=holdout_hours,
-                patience=patience,
-                curriculum_window_days=window_days,
-                curriculum_step_days=step_days,
-                max_intermediate_passes=max_intermediate_passes,
-            )
+            else:
+                val_mae = self.predictor.train_curriculum(
+                    load_data_snap,
+                    now_utc_snap,
+                    pv_minutes=pv_data_snap,
+                    temp_minutes=temp_data_snap,
+                    import_rates=import_rates_snap,
+                    export_rates=export_rates_snap,
+                    epochs=epochs,
+                    time_decay_days=time_decay,
+                    validation_holdout_hours=holdout_hours,
+                    patience=patience,
+                    curriculum_window_days=window_days,
+                    curriculum_step_days=step_days,
+                    max_intermediate_passes=max_intermediate_passes,
+                )
 
             if val_mae is not None:
                 self.last_train_time = datetime.now(timezone.utc)
@@ -1056,7 +1076,7 @@ def _load_baseline_now(self):
                 derived_baseline += self.load_data.get(minute, 0.0)
         derived_baseline = dp4(derived_baseline)
         self.log(
-            "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))
+            "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))
         )
         return derived_baseline
 

apps/predbat/load_predictor.py

@@ -107,6 +107,7 @@ def __array__(self, dtype=None, copy=None):
         if self.mean is not None and self.std is not None:
             out -= self.mean
             out /= self.std
+        np.nan_to_num(out, copy=False, nan=0.0, posinf=0.0, neginf=0.0)
         return out.astype(dtype) if dtype is not None and dtype != np.float32 else out
 
     def _assemble(self, target_chunks, out):
@@ -142,6 +143,7 @@ def __getitem__(self, indices):
         if self.mean is not None and self.std is not None:
             batch -= self.mean
             batch /= self.std
+        np.nan_to_num(batch, copy=False, nan=0.0, posinf=0.0, neginf=0.0)
         return batch[0] if scalar else batch
 
 
@@ -361,6 +363,10 @@ def _backward(self, y_true, activations, pre_activations, sample_weights=None, d
                 if dropout_masks is not None and (i - 1) < len(dropout_masks) and dropout_masks[i - 1] is not None:
                     delta = delta * dropout_masks[i - 1]
 
+        # Clip gradients to prevent exploding gradients and NaN weight updates
+        weight_grads = [np.clip(g, -10.0, 10.0) for g in weight_grads]
+        bias_grads = [np.clip(g, -10.0, 10.0) for g in bias_grads]
+
         return weight_grads, bias_grads
 
     def _adam_update(self, weight_grads, bias_grads, beta1=0.9, beta2=0.999, epsilon=1e-8, lr=None):
@@ -405,6 +411,10 @@ def _adam_update(self, weight_grads, bias_grads, beta1=0.9, beta2=0.999, epsilon
             # Update biases (no weight decay on biases)
             self.biases[i] -= effective_lr * m_hat / (np.sqrt(v_hat) + epsilon)
 
+            # Ensure weights and biases remain finite
+            self.weights[i] = np.nan_to_num(self.weights[i], copy=False, nan=0.0, posinf=0.0, neginf=0.0)
+            self.biases[i] = np.nan_to_num(self.biases[i], copy=False, nan=0.0, posinf=0.0, neginf=0.0)
+
     def _create_time_features(self, minute_of_day, day_of_week, day_of_year=1):
         """
         Create cyclical time features.
@@ -533,12 +543,16 @@ def _chunk_energy_to_aligned(self, energy_per_step, now_utc):
             for j in range(CHUNK_STEPS):
                 m = minute + j * STEP_MINUTES
                 if m in energy_per_step:
-                    total += energy_per_step[m]
+                    val = energy_per_step[m]
+                    if val is None or np.isnan(val) or np.isinf(val):
+                        valid = False
+                        break
+                    total += float(val)
                 else:
                     valid = False
                     break
-            if valid:
-                chunked[chunk_idx] = total
+            if valid and np.isfinite(total):
+                chunked[chunk_idx] = float(total)
             chunk_idx += 1
             minute += CHUNK_MINUTES
 
@@ -572,9 +586,13 @@ def _chunk_instantaneous_to_aligned(self, values_per_step, alignment_offset):
             for j in range(CHUNK_STEPS):
                 m = minute + j * STEP_MINUTES
                 if m in values_per_step:
-                    vals.append(values_per_step[m])
+                    val = values_per_step[m]
+                    if val is not None and not np.isnan(val) and not np.isinf(val):
+                        vals.append(float(val))
             if vals:
-                chunked[chunk_idx] = float(np.mean(vals))
+                mean_val = float(np.mean(vals))
+                if np.isfinite(mean_val):
+                    chunked[chunk_idx] = mean_val
             chunk_idx += 1
             minute += CHUNK_MINUTES
 
@@ -909,7 +927,7 @@ def _feature_mean_std(X, block=512):
         total = np.zeros(X.shape[1], dtype=np.float64)
         for start in range(0, rows, block):
             total += X[start : start + block].sum(axis=0, dtype=np.float64)
-        mean = total / rows
+        mean = np.nan_to_num(total / rows, nan=0.0, posinf=0.0, neginf=0.0)
 
         squares = np.zeros(X.shape[1], dtype=np.float64)
         for start in range(0, rows, block):
@@ -918,7 +936,7 @@ def _feature_mean_std(X, block=512):
             # Square into the deviation buffer rather than allocating a second one of the
             # same size; the values are identical either way
             squares += np.square(deviation, out=deviation).sum(axis=0)
-        std = np.sqrt(squares / rows)
+        std = np.nan_to_num(np.sqrt(squares / rows), nan=1.0, posinf=1.0, neginf=1.0)
 
         return mean.astype(np.float32), std.astype(np.float32)
 
@@ -936,14 +954,14 @@ def _fit_windowed_normalisation(self, dataset, ema_alpha=0.0):
         total = np.zeros(TOTAL_FEATURES, dtype=np.float64)
         for block in dataset.blocks():
             total += block.sum(axis=0, dtype=np.float64)
-        mean = total / rows
+        mean = np.nan_to_num(total / rows, nan=0.0, posinf=0.0, neginf=0.0)
 
         squares = np.zeros(TOTAL_FEATURES, dtype=np.float64)
         for block in dataset.blocks():
             deviation = block.astype(np.float64)
             deviation -= mean
             squares += np.square(deviation, out=deviation).sum(axis=0)
-        std = np.sqrt(squares / rows)
+        std = np.nan_to_num(np.sqrt(squares / rows), nan=1.0, posinf=1.0, neginf=1.0)
 
         new_mean = mean.astype(np.float32)
         new_std = np.maximum(std.astype(np.float32), self._get_min_std_array(TOTAL_FEATURES))
@@ -1054,9 +1072,11 @@ def _normalize_features(self, X, fit=False, ema_alpha=0.0, in_place=False):
         if in_place:
             X -= self.feature_mean
             X /= self.feature_std
+            np.nan_to_num(X, copy=False, nan=0.0, posinf=0.0, neginf=0.0)
             return X
 
-        return (X - self.feature_mean) / self.feature_std
+        out = (X - self.feature_mean) / self.feature_std
+        return np.nan_to_num(out, copy=False, nan=0.0, posinf=0.0, neginf=0.0)
 
     def _normalize_targets(self, y, fit=False):
         """
@@ -1070,14 +1090,17 @@ def _normalize_targets(self, y, fit=False):
             Normalized target array
         """
         if fit:
-            self.target_mean = np.mean(y)
-            self.target_std = np.std(y)
+            m = np.nanmean(y)
+            s = np.nanstd(y)
+            self.target_mean = float(m) if np.isfinite(m) else 0.0
+            self.target_std = float(s) if np.isfinite(s) else 1.0
             self.target_std = max(self.target_std, 1e-8)
 
         if self.target_mean is None or self.target_std is None:
             return y
 
-        return (y - self.target_mean) / self.target_std
+        out = (y - self.target_mean) / self.target_std
+        return np.nan_to_num(out, copy=False, nan=0.0, posinf=0.0, neginf=0.0)
 
     def _denormalize_predictions(self, y_pred):
         """
@@ -1364,8 +1387,9 @@ def train(
         self.validation_mae = best_val_loss
         self.validation_bias = float(best_val_bias)
         self.epochs_trained += epochs
-
-        self.log("ML Predictor: Training complete, final val_mae={:.4f} kWh val_bias={:+.4f} kWh ({:+.1f}%)".format(best_val_loss, float(best_val_bias), 100.0 * float(best_val_bias) / (float(np.mean(y_val)) if float(np.mean(y_val)) > 1e-8 else 1e-8)))
+        mean_y = float(np.mean(y_val)) if float(np.mean(y_val)) > 1e-8 else 1e-8
+        pct_bias = 100.0 * float(best_val_bias) / mean_y
+        self.log("ML Predictor: Training complete, final val_mae={:.4f} kWh val_bias={:+.4f} kWh ({:+.1f}%)".format(best_val_loss, float(best_val_bias), pct_bias))
 
         # Autoregressive diagnostic: run a full AR rollout over the holdout period
         # to expose compounding error (teacher-forced val_mae won't show this)
@@ -1516,7 +1540,7 @@ def train_curriculum(
             )
 
         total_passes = len(window_sizes) + 1  # intermediate passes + final full pass
-        self.log("ML Predictor: Curriculum training - {} passes, window {:.1f}→{:.1f} days + final full pass ({:.1f} days)".format(total_passes, window_sizes[0] / day_minutes, window_sizes[-1] / day_minutes, max_minute / day_minutes))
+        self.log("ML Predictor: Curriculum training - {} passes, window {:.1f}->{:.1f} days + final full pass ({:.1f} days)".format(total_passes, window_sizes[0] / day_minutes, window_sizes[-1] / day_minutes, max_minute / day_minutes))
 
         val_mae = None
         for pass_idx, window in enumerate(window_sizes):
@@ -1542,7 +1566,7 @@ def train_curriculum(
                 temp_minutes=temp_slice,
                 import_rates=import_slice,
                 export_rates=export_slice,
-                is_initial=(pass_idx == 0),
+                is_initial=(pass_idx == 0 and not self.model_initialized),
                 epochs=epochs,
                 time_decay_days=time_decay_days,
                 patience=patience,
@@ -1566,7 +1590,7 @@ def train_curriculum(
             temp_minutes=temp_minutes,
             import_rates=import_rates,
             export_rates=export_rates,
-            is_initial=False,
+            is_initial=(not self.model_initialized),
             epochs=epochs,
             time_decay_days=time_decay_days,
             patience=patience,