Resolve potential data cut off in training data consolidation

Author: dominiquekleeven

src/service_ml_forecast/clients/openremote/models.py

@@ -59,7 +59,7 @@ class AssetDatapoint(BaseModel):
     """Data point of an asset attribute.
 
     Args:
-        x: The timestamp of the data point.
+        x: The timestamp of the data point in milliseconds.
         y: The value of the data point.
     """
 

src/service_ml_forecast/services/model_scheduler.py

@@ -15,6 +15,7 @@
 #
 # SPDX-License-Identifier: AGPL-3.0-or-later
 
+import datetime
 import logging
 import time
 
@@ -203,10 +204,16 @@ def _model_training_job(config: ModelConfig, data_service: OpenRemoteService) ->
     # Save the model
     provider.save_model(model)
 
+    # Log the first and last datapoint datetimes of the target attribute
+    target_first_datapoint_datetime = datetime.datetime.fromtimestamp(training_dataset.target.datapoints[0].x / 1000)
+    target_last_datapoint_datetime = datetime.datetime.fromtimestamp(training_dataset.target.datapoints[-1].x / 1000)
+
     end_time = time.perf_counter()
     logger.info(
         f"Training job for {config.id} completed - duration: {end_time - start_time}s. "
-        f"Type: {config.type}, Training Interval: {config.training_interval}"
+        f"Type: {config.type}, Training Interval: {config.training_interval}, "
+        f"Target first datapoint datetime: {target_first_datapoint_datetime}, "
+        f"Target last datapoint datetime: {target_last_datapoint_datetime}"
     )
 
 
@@ -248,8 +255,14 @@ def _model_forecast_job(config: ModelConfig, data_service: OpenRemoteService) ->
         return
 
     end_time = time.perf_counter()
+
+    # Log the first and last datapoint datetimes of the forecast
+    first_datapoint_datetime = datetime.datetime.fromtimestamp(forecast.datapoints[0].x / 1000)
+    last_datapoint_datetime = datetime.datetime.fromtimestamp(forecast.datapoints[-1].x / 1000)
+
     logger.info(
         f"Forecasting job for {config.id} completed - duration: {end_time - start_time}s. "
         f"Wrote {len(forecast.datapoints)} datapoints. "
+        f"First datapoint datetime: {first_datapoint_datetime}, Last datapoint datetime: {last_datapoint_datetime}"
         f"Asset ID: {config.target.asset_id}, Attribute: {config.target.attribute_name}"
     )

src/service_ml_forecast/services/openremote_service.py

@@ -152,7 +152,9 @@ def __get_historical_datapoints(
             logger.info(
                 f"Chunking datapoint retrieval into {months_diff} monthly chunks for {asset_id} {attribute_name}"
             )
-            for i in range(months_diff):
+            
+            # Continue until we've processed a chunk that ends at or after to_timestamp
+            while current_from < to_timestamp:
                 # Calculate the end timestamp for this chunk (1 month from current_from)
                 current_to = TimeUtil.add_months_to_timestamp(current_from, 1)
 
@@ -166,7 +168,7 @@ def __get_historical_datapoints(
                 if chunk_datapoints is None:
                     logger.error(
                         f"Failed to retrieve historical datapoints for {asset_id} {attribute_name} "
-                        f"for chunk {i + 1} of {months_diff}"
+                        f"for chunk ending at {current_to}"
                     )
                     return None
 
@@ -175,10 +177,6 @@ def __get_historical_datapoints(
                 # Move to the next chunk
                 current_from = current_to
 
-                # Break if we've reached the end
-                if current_from >= to_timestamp:
-                    break
-
             return all_datapoints
 
     def get_forecast_dataset(self, config: ModelConfig) -> ForecastDataSet | None: