Issue/blend forecasts (#163)

* first version of blend

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* lint

* add adjust_mw to blend

* add model for updateing forecast values latest

* update blend

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* fix merge

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: peterdudfield

nowcasting_datamodel/read/blend.py

@@ -222,7 +222,7 @@ def get_blend_forecast_values_latest(
     return forecast_values
 
 
-def make_weights_df(model_names, weights, start_datetime_now=None):
+def make_weights_df(model_names, weights, start_datetime=None):
     """Makes weights to half an hour and blocks
 
     A pd data frame like
@@ -254,17 +254,20 @@ def make_weights_df(model_names, weights, start_datetime_now=None):
             },
         ]
     # get time now rounded up to 30 mins
-    if start_datetime_now is None:
-        start_datetime_now = datetime.now().replace(second=0, microsecond=0)
-        if start_datetime_now.minute >= 30:
-            start_datetime_now += timedelta(hours=1)
-            start_datetime_now = start_datetime_now.replace(minute=00)
-        else:
-            start_datetime_now = start_datetime_now.replace(minute=30)
+    start_datetime_now = datetime.now(tz=timezone.utc).replace(second=0, microsecond=0)
+    if start_datetime_now.minute >= 30:
+        start_datetime_now += timedelta(hours=1)
+        start_datetime_now = start_datetime_now.replace(minute=00)
+    else:
+        start_datetime_now = start_datetime_now.replace(minute=30)
+
+    if start_datetime is None:
+        start_datetime = start_datetime_now
 
     # make dataframe of 8 hours in 30 minutes chunks from now
     weights_all_df = []
-    for weight in weights:
+    for i in range(len(weights)):
+        weight = weights[i]
         if "start_horizon_hour" not in weight:
             start_horizon_hour = 0
         else:
@@ -283,41 +286,55 @@ def make_weights_df(model_names, weights, start_datetime_now=None):
         else:
             end_weight = weight["end_weight"]
 
+        # add the first weight to the dataframe, from  start_datetime to start_datetime_now.
+        # This could be for the two days before now
+        if i == 0:
+            weights_df = pd.DataFrame(
+                index=pd.date_range(start=start_datetime, end=start_datetime_now, freq="30min")
+            )
+            weights_df[model_names] = start_weight
+            weights_all_df.append(weights_df)
+
         logger.debug(
             f"Making weights for {start_horizon_hour} to {end_horizon_hour} "
             f"hours with weights {start_weight} to {end_weight}"
         )
 
-        start_datetime = (
+        start_datetime_one_weight = (
             start_datetime_now + timedelta(hours=start_horizon_hour) - timedelta(minutes=30)
         )
         if start_horizon_hour == 0:
-            start_datetime += timedelta(minutes=30)
+            start_datetime_one_weight += timedelta(minutes=30)
+
         end_datetime = (
             start_datetime_now + timedelta(hours=end_horizon_hour) - timedelta(minutes=30)
         )
         if end_horizon_hour == 8:
             end_datetime += timedelta(minutes=30)
 
+        logger.debug(f"Making weights from {start_datetime_one_weight} to {end_datetime}")
         weights_df = pd.DataFrame(
-            index=pd.date_range(start=start_datetime, end=end_datetime, freq="30min")
+            index=pd.date_range(start=start_datetime_one_weight, end=end_datetime, freq="30min")
         )
+
         # get rid of last timestamp
         weights_df = weights_df[:-1]
 
         assert len(model_names) == len(start_weight)
         assert len(model_names) == len(end_weight)
         for model_name, start_weight, end_weight in zip(model_names, start_weight, end_weight):
             weights_df[model_name] = start_weight + (end_weight - start_weight) * (
-                weights_df.index - start_datetime
-            ) / (end_datetime - start_datetime)
+                weights_df.index - start_datetime_one_weight
+            ) / (end_datetime - start_datetime_one_weight)
 
         weights_all_df.append(weights_df)
 
     weights_all_df = pd.concat(weights_all_df)
     weights_all_df = weights_all_df[~weights_all_df.index.duplicated(keep="first")]
 
     # only keep from now
-    weights_all_df = weights_all_df[weights_all_df.index >= start_datetime_now]
+    logger.debug(weights_all_df)
+    logger.debug(start_datetime)
+    weights_all_df = weights_all_df[weights_all_df.index >= start_datetime]
 
     return weights_all_df

nowcasting_datamodel/read/read.py

@@ -124,6 +124,7 @@ def get_latest_forecast(
     gsp_id: Optional[int] = None,
     historic: bool = False,
     start_target_time: Optional[datetime] = None,
+    model_name: Optional[str] = None,
 ) -> ForecastSQL:
     """
     Read forecasts
@@ -135,6 +136,7 @@ def get_latest_forecast(
     :param historic: Option to load historic values or not
     :param start_target_time:
         Filter: forecast values target time should be larger than this datetime
+    :param model_name: Filter: model name
 
     return: List of forecasts objects from database
     """
@@ -151,6 +153,7 @@ def get_latest_forecast(
         start_target_time=start_target_time,
         historic=historic,
         gsp_ids=gsp_ids,
+        model_name=model_name,
     )
 
     if forecasts is None:
@@ -255,6 +258,7 @@ def get_latest_forecast_for_gsps(
     preload_children: Optional[bool] = False,
     historic: bool = False,
     gsp_ids: List[int] = None,
+    model_name: Optional[int] = None,
 ):
     """
     Read forecasts
@@ -268,10 +272,13 @@ def get_latest_forecast_for_gsps(
     :param preload_children: Option to preload children. This is a speed up, if we need them.
     :param historic: Option to load historic values or not
     :param gsp_ids: Option to filter on gsps. If None, then only the lastest forecast is loaded.
+    :param model_name: Option to filter on model name
 
     :return: List of forecasts objects from database
 
     """
+    logger.debug(f"Getting latest forecast for gsps {gsp_ids} {historic=} {model_name=}")
+
     order_by_cols = []
 
     # start main query
@@ -292,6 +299,11 @@ def get_latest_forecast_for_gsps(
     # filter on historic
     query = query.filter(ForecastSQL.historic == historic)
 
+    # filter on model name
+    if model_name is not None:
+        query = query.join(MLModelSQL)
+        query = query.filter(MLModelSQL.name == model_name)
+
     # filter on target time
     if start_target_time is not None:
         query = filter_query_on_target_time(
@@ -677,7 +689,7 @@ def get_all_locations(session: Session, gsp_ids: List[int] = None) -> List[Locat
     return locations
 
 
-def get_model(session: Session, name: str, version: str) -> MLModelSQL:
+def get_model(session: Session, name: str, version: Optional[str]) -> MLModelSQL:
     """
     Get model object from name and version
 

nowcasting_datamodel/save/update.py

@@ -17,24 +17,28 @@
 from nowcasting_datamodel.read.read import (
     get_latest_forecast,
     get_latest_forecast_for_gsps,
-    get_model,
 )
 
 logger = logging.getLogger(__name__)
 
 
-def get_historic_forecast(session: Session, forecast: ForecastSQL) -> ForecastSQL:
+def get_historic_forecast(
+    session: Session, forecast: ForecastSQL, model_name: Optional[str] = None
+) -> ForecastSQL:
     """
     Get historic forecast
 
     :param session:
     :param gsp_id:
+    :param model_name: the model name to filter on
     :return:
     """
 
     gsp_id = forecast.location.gsp_id
 
-    forecast_historic = get_latest_forecast(session=session, gsp_id=gsp_id, historic=True)
+    forecast_historic = get_latest_forecast(
+        session=session, gsp_id=gsp_id, historic=True, model_name=model_name
+    )
 
     if forecast_historic is None:
         logger.debug("Could not find a historic forecast, so will make one")
@@ -44,7 +48,7 @@ def get_historic_forecast(session: Session, forecast: ForecastSQL) -> ForecastSQ
             forecast_creation_time=datetime.now(timezone.utc),
             location=forecast.location,
             input_data_last_updated=forecast.input_data_last_updated,
-            model=get_model(session=session, name="historic", version="all"),
+            model=forecast.model,
         )
         session.add(forecast_historic)
         session.commit()
@@ -79,7 +83,10 @@ def upsert(session: Session, model, rows: List[dict]):
 
 
 def update_forecast_latest(
-    forecast: ForecastSQL, session: Session, forecast_historic: Optional[ForecastSQL] = None
+    forecast: ForecastSQL,
+    session: Session,
+    forecast_historic: Optional[ForecastSQL] = None,
+    model_name: Optional[str] = None,
 ):
     """
     Update the forecast_values table
@@ -90,12 +97,15 @@ def update_forecast_latest(
     3. upsert them (update and or insert)
 
     :param forecast:
+    :param model_name: the model name to filter on
     :return:
     """
 
     # 1. get forecast object
     if forecast_historic is None:
-        forecast_historic = get_historic_forecast(session=session, forecast=forecast)
+        forecast_historic = get_historic_forecast(
+            session=session, forecast=forecast, model_name=model_name
+        )
 
     # 2. create forecast value latest
     forecast_values = []
@@ -106,6 +116,7 @@ def update_forecast_latest(
             forecast_id=forecast_historic.id,
             model_id=forecast_historic.model_id,
         )
+        logger.debug(f"{forecast_historic.model_id=}")
         forecast_values.append(forecast_value_latest.__dict__)
 
     # upsert forecast values
@@ -151,6 +162,7 @@ def update_all_forecast_latest(
     session: Session,
     update_national: Optional[bool] = True,
     update_gsp: Optional[bool] = True,
+    model_name: Optional[str] = None,
 ):
     """
     Update all latest forecasts
@@ -160,6 +172,7 @@ def update_all_forecast_latest(
     :param session: sqlalmacy session
     :param update_national: Optional (default true), to update the national forecast
     :param update_gsp: Optional (default true), to update all the GSP forecasts
+    :param model_name: Optional (default None), if not None will only update the forecasts
     """
 
     logger.debug("Getting the earliest forecast target time for the first forecast")
@@ -184,6 +197,7 @@ def update_all_forecast_latest(
         preload_children=True,
         gsp_ids=gsp_ids,
         start_target_time=start_target_time,
+        model_name=model_name,
     )
     # get all these ids, so we only have to load it once
     historic_gsp_ids = [forecast.location.gsp_id for forecast in forecasts_historic_all_gsps]
@@ -220,7 +234,10 @@ def update_all_forecast_latest(
             logger.debug(f"Found historic for GSP id {gsp_id}")
 
         update_forecast_latest(
-            forecast=forecast, session=session, forecast_historic=forecast_historic
+            forecast=forecast,
+            session=session,
+            forecast_historic=forecast_historic,
+            model_name=model_name,
         )
         session.commit()
 

tests/read/test_blend.py

@@ -16,7 +16,7 @@
 
 @freeze_time("2023-01-01 00:00:01")
 def test_make_weights_df():
-    start_datetime = datetime(2023, 1, 1, 0, 30)
+    start_datetime = datetime(2023, 1, 1, 0, 30, tzinfo=timezone.utc)
 
     weights = make_weights_df(model_names=["test_1", "test_2"], weights=None)
     assert len(weights) == 16
@@ -45,9 +45,43 @@ def test_make_weights_df():
     assert weights["test_2"][15] == 1
 
 
-#
+@freeze_time("2023-01-02 00:00:01")
+def test_make_weights_df_yesterday():
+    start_datetime = datetime(2023, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
 
+    weights = make_weights_df(
+        model_names=["test_1", "test_2"], weights=None, start_datetime=start_datetime
+    )
+    assert len(weights) == 49 + 16
+    assert "test_1" in weights.columns
+    assert "test_2" in weights.columns
 
+    assert weights.index[0].isoformat() == start_datetime.isoformat()
+    assert (
+        weights.index[15].isoformat()
+        == (start_datetime + timedelta(hours=7, minutes=30)).isoformat()
+    )
+
+    assert weights["test_1"][0] == 1
+    assert weights["test_2"][0] == 0
+
+    assert weights["test_1"][49] == 1
+    assert weights["test_2"][49] == 0
+
+    assert weights["test_1"][49 + 3] == 1
+    assert weights["test_2"][49 + 3] == 0
+
+    assert weights["test_1"][49 + 7] == 0.5
+    assert weights["test_2"][49 + 7] == 0.5
+
+    assert weights["test_1"][49 + 11] == 0
+    assert weights["test_2"][49 + 11] == 1
+
+    assert weights["test_1"][49 + 15] == 0
+    assert weights["test_2"][49 + 15] == 1
+
+
+@freeze_time("2023-01-01 00:00:01")
 def test_get_blend_forecast_values_latest_one_model(db_session):
 
     model = get_model(session=db_session, name="test_1", version="0.0.1")
@@ -86,6 +120,7 @@ def test_get_blend_forecast_values_latest_one_model(db_session):
     )
 
 
+@freeze_time("2023-01-01 00:00:01")
 def test_get_blend_forecast_values_latest_two_model_read_one(db_session):
 
     model_1 = get_model(session=db_session, name="test_1", version="0.0.1")
@@ -126,6 +161,7 @@ def test_get_blend_forecast_values_latest_two_model_read_one(db_session):
     )
 
 
+@freeze_time("2023-01-01 00:00:01")
 def test_get_blend_forecast_values_latest_two_model_read_two(db_session):
 
     model_1 = get_model(session=db_session, name="test_1", version="0.0.1")
@@ -142,7 +178,7 @@ def test_get_blend_forecast_values_latest_two_model_read_two(db_session):
             power = 2
             adjust = 100
 
-        forecast_horizon_minutes = [0, 30, 7 * 30, 15 * 30]
+        forecast_horizon_minutes = [0, 30, 8 * 30, 15 * 30]
         f1[0].forecast_values_latest = [
             ForecastValueLatestSQL(
                 gsp_id=1,