fix(prediction): stop predict_pv_power scaling with plan_interval_minutes

apps/predbat/prediction.py

@@ -1230,8 +1230,11 @@ def run_prediction(self, charge_limit, charge_window, export_window, export_limi
                 predict_state[stamp] = "g" + grid_state + "b" + battery_state
                 predict_battery_power[stamp] = round(battery_draw * (60 / step), 3)
                 predict_battery_cycle[stamp] = round(battery_cycle, 3)
-                # Use plan_interval_minutes instead of hardcoded 30 for scaling
-                predict_pv_power[stamp] = round((pv_forecast_minute_step[minute] + pv_forecast_minute_step.get(minute + step, 0)) * (self.plan_interval_minutes / step), 3)
+                # Two consecutive `step`-sized energy chunks cover 2*step minutes; convert to an
+                # instantaneous kW reading with 60/(2*step) - a constant derived from `step` (the
+                # simulation's fixed PREDICT_STEP), not plan_interval_minutes, which is unrelated
+                # to how many raw steps are being summed here.
+                predict_pv_power[stamp] = round((pv_forecast_minute_step[minute] + pv_forecast_minute_step.get(minute + step, 0)) * (60 / (2 * step)), 3)
                 predict_grid_power[stamp] = round(diff * (60 / step), 3)
                 predict_load_power[stamp] = round(load_yesterday * (60 / step), 3)
                 if carbon_enable:

apps/predbat/tests/test_predict_pv_power.py

@@ -0,0 +1,102 @@
+# -----------------------------------------------------------------------------
+# Predbat Home Battery System
+# Copyright Trefor Southwell 2026 - All Rights Reserved
+# This application maybe used for personal use only and not for commercial use
+# -----------------------------------------------------------------------------
+# fmt off
+# pylint: disable=consider-using-f-string
+# pylint: disable=line-too-long
+# pylint: disable=attribute-defined-outside-init
+
+from prediction import Prediction
+from tests.test_infra import reset_inverter
+
+
+def _run_flat_pv(my_predbat, plan_interval_minutes, flat_pv_kw):
+    """
+    Run a real run_prediction(save="best") with a known, flat PV forecast and return the
+    resulting Prediction object so predict_pv_power can be inspected.
+    """
+    my_predbat.plan_interval_minutes = plan_interval_minutes
+    reset_inverter(my_predbat)
+    my_predbat.forecast_minutes = 24 * 60
+    my_predbat.end_record = my_predbat.forecast_minutes
+
+    per_minute_kwh = flat_pv_kw / 60.0
+    horizon = my_predbat.forecast_minutes + my_predbat.minutes_now + 10
+    my_predbat.pv_forecast_minute = {m: per_minute_kwh for m in range(0, horizon)}
+    my_predbat.pv_forecast_minute10 = dict(my_predbat.pv_forecast_minute)
+    my_predbat.load_minutes_step = {m: 0.0 for m in range(0, horizon)}
+    my_predbat.load_minutes_step10 = dict(my_predbat.load_minutes_step)
+
+    pv_step = {m: per_minute_kwh * 5 for m in range(0, horizon, 5)}
+    load_step = {m: 0.0 for m in range(0, horizon, 5)}
+    my_predbat.prediction = Prediction(my_predbat, pv_step, pv_step, load_step, load_step)
+
+    my_predbat.run_prediction([], [], [], [], False, end_record=my_predbat.end_record, save="best")
+    return my_predbat.prediction
+
+
+def test_predict_pv_power_matches_input_regardless_of_plan_interval(my_predbat):
+    """
+    Regression test for issue #4361: predict_pv_power (which feeds pv_power_best, the source of
+    the "PV Power (Predicted)" trace on the LoadMLPower chart) must report the same instantaneous
+    kW as the underlying flat PV forecast, regardless of plan_interval_minutes.
+
+    Root cause: the formula converts two summed PREDICT_STEP (5-min) energy chunks into an
+    instantaneous kW reading via a 60/(2*step) constant - PR #2865 (2025-11-07) accidentally
+    replaced that constant with self.plan_interval_minutes, which is unrelated to how many raw
+    5-minute steps are being summed. That made the reported PV power scale with
+    plan_interval_minutes/30 - correct only for the 30-minute default, and exactly half for
+    anyone on 15-minute plan intervals (as in #4361), a third for 10-minute, etc.
+    """
+    print("**** Running predict_pv_power plan-interval-independence test ****")
+    failed = False
+
+    # my_predbat is a shared fixture reused by every test module in the suite - save everything
+    # this test mutates and restore it afterwards so later tests (e.g. model_kernel, which assumes
+    # the 30-minute default) aren't left running against a stale plan_interval_minutes.
+    original_plan_interval_minutes = my_predbat.plan_interval_minutes
+    original_forecast_minutes = my_predbat.forecast_minutes
+    original_end_record = my_predbat.end_record
+    original_pv_forecast_minute = getattr(my_predbat, "pv_forecast_minute", None)
+    original_pv_forecast_minute10 = getattr(my_predbat, "pv_forecast_minute10", None)
+    original_load_minutes_step = getattr(my_predbat, "load_minutes_step", None)
+    original_load_minutes_step10 = getattr(my_predbat, "load_minutes_step10", None)
+    original_prediction = getattr(my_predbat, "prediction", None)
+
+    try:
+        flat_pv_kw = 4.0
+        for plan_interval_minutes in [30, 15, 10]:
+            pred = _run_flat_pv(my_predbat, plan_interval_minutes, flat_pv_kw)
+            values = list(pred.predict_pv_power.values())
+            # Skip the first/last couple of samples in case of edge alignment effects, check a mid-horizon one
+            sample = values[len(values) // 2]
+            if abs(sample - flat_pv_kw) > 0.01:
+                print("ERROR: plan_interval_minutes={} flat PV input={}kW but predict_pv_power sample={}kW".format(plan_interval_minutes, flat_pv_kw, sample))
+                failed = True
+            else:
+                print("  plan_interval_minutes={}: predict_pv_power={}kW (expected {}kW) - OK".format(plan_interval_minutes, sample, flat_pv_kw))
+    finally:
+        my_predbat.plan_interval_minutes = original_plan_interval_minutes
+        my_predbat.forecast_minutes = original_forecast_minutes
+        my_predbat.end_record = original_end_record
+        my_predbat.pv_forecast_minute = original_pv_forecast_minute
+        my_predbat.pv_forecast_minute10 = original_pv_forecast_minute10
+        my_predbat.load_minutes_step = original_load_minutes_step
+        my_predbat.load_minutes_step10 = original_load_minutes_step10
+        my_predbat.prediction = original_prediction
+
+    if failed:
+        print("Test: predict_pv_power_matches_input_regardless_of_plan_interval FAILED")
+    else:
+        print("Test: predict_pv_power_matches_input_regardless_of_plan_interval PASSED")
+
+    return failed
+
+
+def run_predict_pv_power_tests(my_predbat):
+    """Run all predict_pv_power regression tests."""
+    failed = False
+    failed |= test_predict_pv_power_matches_input_regardless_of_plan_interval(my_predbat)
+    return failed

apps/predbat/unit_test.py

@@ -19,6 +19,7 @@
 from tests.test_compute_metric import run_compute_metric_tests
 from tests.test_perf import run_perf_test
 from tests.test_model import run_model_tests
+from tests.test_predict_pv_power import run_predict_pv_power_tests
 from tests.test_kernel_parity import run_kernel_parity_tests, run_model_kernel_tests
 from tests.test_execute import run_execute_tests
 from tests.test_octopus_slots import run_load_octopus_slots_tests
@@ -225,6 +226,7 @@ def main():
         ("secrets", run_secrets_tests, "Secrets loading tests", False),
         ("perf", run_perf_test, "Performance tests", False),
         ("model", run_model_tests, "Model tests", False),
+        ("predict_pv_power", run_predict_pv_power_tests, "predict_pv_power plan-interval scaling tests", False),
         ("model_kernel", run_model_kernel_tests, "Model tests run with the C++ prediction kernel enabled", False),
         ("kernel_parity", run_kernel_parity_tests, "C++ prediction kernel vs Python engine parity tests", False),
         ("inverter", run_inverter_tests, "Inverter tests", False),