fix(plan): score plan selection on the plan as optimised, not as clipped

.cspell/custom-dictionary-workspace.txt

@@ -354,6 +354,7 @@ powerline
 Powerwall
 ppdetails
 ppkwh
+preclip
 pred
 predai
 predbat

apps/predbat/plan.py

@@ -999,6 +999,25 @@ def keep_window_change_if_improved(self, baseline, candidate, typ, window_n, sna
         self.plan_window_restore(typ, window_n, snapshot)
         return baseline
 
+    def plan_scoring_pair(self, plan_new, plan_prev, preclip_new, preclip_prev):
+        """Return the (new, previous) plans that selection should be scored on.
+
+        Plans are (charge_limit, charge_window, export_window, export_limits) tuples.
+
+        Clipping sets the charge/export percentage actually shown on the plan and sent to the inverter, so it
+        has to stay. It is a no-op in the expected case - the mid-case cost is unchanged - but it moves the
+        metric through the PV10 branch by an amount that depends on plan shape. Scoring the clipped plans
+        therefore decides between them partly on a difference clipping invented rather than one the plans
+        really have, and by then the slot is usually hours away and will be re-planned many times before it
+        executes.
+
+        Both sides fall back together when either snapshot is missing (the first recompute after a restart):
+        scoring an un-taxed new plan against a taxed incumbent would favour the new plan on the tax alone.
+        """
+        if preclip_new is not None and preclip_prev is not None:
+            return preclip_new, preclip_prev
+        return plan_new, plan_prev
+
     def calculate_plan(self, recompute=True, debug_mode=False, publish=True):
         """
         Calculate the new plan (best)
@@ -1047,12 +1066,14 @@ def calculate_plan(self, recompute=True, debug_mode=False, publish=True):
                 charge_window_best_prev = copy.deepcopy(self.charge_window_best)
                 export_window_best_prev = copy.deepcopy(self.export_window_best)
                 export_limits_best_prev = copy.deepcopy(self.export_limits_best)
+                preclip_prev = self.plan_preclip
                 self.log("Recompute is saving previous plan...")
             else:
                 charge_limit_best_prev = None
                 charge_window_best_prev = None
                 export_window_best_prev = None
                 export_limits_best_prev = None
+                preclip_prev = None
                 self.log("Recompute, previous plan is invalid...")
 
             self.plan_valid = False  # In case of crash, plan is now invalid
@@ -1179,6 +1200,9 @@ def calculate_plan(self, recompute=True, debug_mode=False, publish=True):
             # Remove charge windows that overlap with export windows
             self.charge_limit_best, self.charge_window_best = remove_intersecting_windows(self.charge_limit_best, self.charge_window_best, self.export_limits_best, self.export_window_best)
 
+            # Snapshot the plan as optimised, before clipping adjusts the percentages for execution
+            preclip_new = (copy.deepcopy(self.charge_limit_best), copy.deepcopy(self.charge_window_best), copy.deepcopy(self.export_window_best), copy.deepcopy(self.export_limits_best))
+
             # Filter out any unused export windows
             if self.calculate_best_export and self.export_window_best:
                 # Filter out the windows we disabled
@@ -1263,27 +1287,38 @@ def calculate_plan(self, recompute=True, debug_mode=False, publish=True):
 
             # Plan comparison
             if charge_window_best_prev is not None and not debug_mode:
-                metric, battery_value, cost, metric_keep, battery_cycle, final_carbon_g, import_kwh, export_kwh = self.run_prediction_metric(
-                    self.charge_limit_best, self.charge_window_best, self.export_window_best, self.export_limits_best, end_record=self.end_record
+                # Score the plans as optimised rather than as clipped - see plan_scoring_pair()
+                score_new, score_prev = self.plan_scoring_pair(
+                    (self.charge_limit_best, self.charge_window_best, self.export_window_best, self.export_limits_best),
+                    (charge_limit_best_prev, charge_window_best_prev, export_window_best_prev, export_limits_best_prev),
+                    preclip_new,
+                    preclip_prev,
                 )
+                metric, battery_value, cost, metric_keep, battery_cycle, final_carbon_g, import_kwh, export_kwh = self.run_prediction_metric(score_new[0], score_new[1], score_new[2], score_new[3], end_record=self.end_record)
                 metric_prev, battery_value_prev, cost_prev, metric_keep_prev, battery_cycle_prev, final_carbon_g_prev, import_kwh_prev, export_kwh_prev = self.run_prediction_metric(
-                    charge_limit_best_prev, charge_window_best_prev, export_window_best_prev, export_limits_best_prev, end_record=self.end_record
+                    score_prev[0], score_prev[1], score_prev[2], score_prev[3], end_record=self.end_record
                 )
 
                 self.log("Previous plan best metric is {} (cost {}) and new plan best metric is {} (cost {})".format(dp2(metric_prev), dp2(cost_prev), dp2(metric), dp2(cost)))
-                fragmentation_prev = self.plan_fragmentation(charge_window_best_prev, charge_limit_best_prev, export_window_best_prev, export_limits_best_prev)
-                fragmentation_new = self.plan_fragmentation(self.charge_window_best, self.charge_limit_best, self.export_window_best, self.export_limits_best)
+                fragmentation_prev = self.plan_fragmentation(score_prev[1], score_prev[0], score_prev[2], score_prev[3])
+                fragmentation_new = self.plan_fragmentation(score_new[1], score_new[0], score_new[2], score_new[3])
                 if not self.should_replace_plan(metric_prev, metric, fragmentation_prev, fragmentation_new):
                     self.log("New plan metric is not significantly better (metric_min_improvement_plan {}) than previous plan, using previous plan".format(self.metric_min_improvement_plan))
                     self.charge_window_best = copy.deepcopy(charge_window_best_prev)
                     self.charge_limit_best = copy.deepcopy(charge_limit_best_prev)
                     self.export_window_best = copy.deepcopy(export_window_best_prev)
                     self.export_limits_best = copy.deepcopy(export_limits_best_prev)
+                    # Keeping the incumbent keeps its pre-clip snapshot too, so the next cycle still compares
+                    # like for like
+                    preclip_new = preclip_prev
                 elif (metric_prev - metric) >= self.metric_min_improvement_plan:
                     self.log("New plan metric is significantly better from previous plan, using new plan")
                 else:
                     self.log("New plan is a cost-neutral improvement but less fragmented ({} vs {} segments), using new plan".format(fragmentation_new, fragmentation_prev))
 
+            # Carry the pre-clip snapshot of whichever plan we kept into the next cycle
+            self.plan_preclip = preclip_new
+
             # Plan is now valid
             self.log("Plan valid is now true after recompute was {}".format(self.plan_valid))
             if not self.update_pending:

apps/predbat/predbat.py

@@ -340,6 +340,7 @@ def reset(self):
         self.previous_status = None
         self.had_errors = False
         self.plan_valid = False
+        self.plan_preclip = None
         self.plan_last_updated = None
         self.plan_last_updated_minutes = 0
         self.plugin_system = None
@@ -684,6 +685,7 @@ def save_plan(self):
             "charge_limit_best": self.charge_limit_best,
             "export_window_best": self.export_window_best,
             "export_limits_best": self.export_limits_best,
+            "plan_preclip": self.plan_preclip,
             "plan_last_updated": self.plan_last_updated.isoformat() if self.plan_last_updated else None,
             "plan_last_updated_minutes": self.plan_last_updated_minutes,
         }
@@ -735,6 +737,10 @@ def load_plan(self):
         self.charge_limit_best = plan_data.get("charge_limit_best", [])
         self.export_window_best = plan_data.get("export_window_best", [])
         self.export_limits_best = plan_data.get("export_limits_best", [])
+        # The pre-clip snapshot plan selection scores against. Older saves predate it, and it is only ever a
+        # four part plan, so anything else is discarded and the comparison falls back to the clipped plans.
+        preclip = plan_data.get("plan_preclip")
+        self.plan_preclip = tuple(preclip) if isinstance(preclip, (list, tuple)) and len(preclip) == 4 else None
         self.plan_last_updated = saved_dt
         self.plan_last_updated_minutes = plan_data.get("plan_last_updated_minutes", 0)
         self.plan_valid = True

apps/predbat/tests/test_plan_persistence.py

@@ -62,6 +62,11 @@ def test_plan_persistence(my_predbat):
         my_predbat.plan_last_updated_minutes = saved_minutes
         my_predbat.plan_valid = True
 
+        # The pre-clip snapshot is what plan selection scores against next cycle, so it has to survive a
+        # restart too - without it the first recompute back compares a clipped incumbent and falls back
+        preclip = ([9.0], charge_windows, export_windows, [0.0])
+        my_predbat.plan_preclip = preclip
+
         # 1. save_plan() must not raise and must write something loadable
         print("  Test 1: save_plan() round-trip")
         my_predbat.save_plan()
@@ -74,6 +79,7 @@ def test_plan_persistence(my_predbat):
         my_predbat.plan_last_updated = None
         my_predbat.plan_last_updated_minutes = 0
         my_predbat.plan_valid = False
+        my_predbat.plan_preclip = None
 
         my_predbat.load_plan()
 
@@ -95,6 +101,12 @@ def test_plan_persistence(my_predbat):
         if my_predbat.plan_last_updated_minutes != saved_minutes:
             print("  FAILED: plan_last_updated_minutes mismatch: {}".format(my_predbat.plan_last_updated_minutes))
             failed += 1
+        if my_predbat.plan_preclip is None:
+            print("  FAILED: plan_preclip was not restored")
+            failed += 1
+        elif [list(x) for x in my_predbat.plan_preclip] != [list(x) for x in preclip]:
+            print("  FAILED: plan_preclip mismatch: {}".format(my_predbat.plan_preclip))
+            failed += 1
 
         # 2. load_plan() with empty storage leaves plan_valid False
         print("  Test 2: load_plan() with no saved plan")

apps/predbat/tests/test_plan_preclip.py

@@ -0,0 +1,75 @@
+# -----------------------------------------------------------------------------
+# 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
+"""Tests that plan selection scores the plan as optimised rather than as clipped.
+
+clip_export_slots and clip_charge_slots set the percentage shown on the plan and sent to the inverter. They
+are a no-op in the expected case but move the metric through the PV10 branch, so calculate_plan keeps a
+snapshot of the plan taken before clipping and compares on that instead - see plan_scoring_pair().
+"""
+from tests.test_infra import reset_inverter
+from tests.test_random_scenarios import load_scenarios, run_scenario
+
+
+def run_plan_preclip_tests(my_predbat):
+    """Run the pre-clip plan selection tests. Returns True on failure."""
+    print("**** Running plan pre-clip selection tests ****")
+    failed = False
+
+    # read_debug_yaml reconfigures the instance wholesale, so work on a throwaway one rather than leaving the
+    # shared fixture holding this debug case for every test that runs after
+    from unit_test import create_predbat
+
+    my_predbat = create_predbat()
+
+    reset_inverter(my_predbat)
+    my_predbat.read_debug_yaml("cases/predbat_debug_agile1.yaml")
+    my_predbat.config_root = "./"
+    my_predbat.save_restore_dir = "./"
+    my_predbat.load_user_config()
+
+    # Reuse a benchmark scenario rather than generating one: these are known to plan in a couple of seconds
+    # Scenario 10 is one where clipping measurably taxes the metric, so the comparison below has teeth: on an
+    # untaxed scenario pre-clip and clipped score the same and the assertion would pass either way
+    scenario = [s for s in load_scenarios("cases/random_scenarios.yaml") if s["id"] == 10][0]
+
+    # First plan establishes an incumbent; the second recompute is the one that runs the comparison
+    run_scenario(my_predbat, scenario)
+    if my_predbat.plan_preclip is None:
+        print("ERROR: plan_preclip was not captured by the first recompute")
+        return True
+
+    run_scenario(my_predbat, scenario)
+
+    if my_predbat.plan_preclip is None:
+        print("ERROR: plan_preclip was not carried through the second recompute")
+        return True
+    if len(my_predbat.plan_preclip) != 4:
+        print("ERROR: plan_preclip should be a four part plan, got {}".format(len(my_predbat.plan_preclip)))
+        return True
+
+    # Clipping never improves a plan, so the snapshot selection scores must be no worse than the plan that
+    # actually executes. Taking the snapshot after clipping instead would make these equal at best and hand
+    # the taxed metric to should_replace_plan.
+    preclip = my_predbat.plan_preclip
+    metric_preclip = my_predbat.run_prediction_metric(preclip[0], preclip[1], preclip[2], preclip[3], end_record=my_predbat.end_record)[0]
+    metric_final = my_predbat.run_prediction_metric(my_predbat.charge_limit_best, my_predbat.charge_window_best, my_predbat.export_window_best, my_predbat.export_limits_best, end_record=my_predbat.end_record)[0]
+
+    if metric_final - metric_preclip < 0.01:
+        print("ERROR: clipping did not tax this scenario ({} vs {}), so the comparison proves nothing".format(metric_preclip, metric_final))
+        failed = True
+    elif metric_preclip > metric_final + 0.0001:
+        print("ERROR: pre-clip plan scored {} which is worse than the clipped plan {} - snapshot is not pre-clip".format(metric_preclip, metric_final))
+        failed = True
+    else:
+        print("pre-clip metric {} vs clipped {} (clipping tax {})".format(round(metric_preclip, 4), round(metric_final, 4), round(metric_final - metric_preclip, 4)))
+
+    if not failed:
+        print("**** Plan pre-clip selection tests passed ****")
+    return failed

apps/predbat/tests/test_plan_tiebreak.py

@@ -79,10 +79,36 @@ def _tiebreak_decision_tests(my_predbat, failures):
     _check(my_predbat.should_replace_plan(-200.0, -200.0, 4, 2) is True, "cost-neutral cleaner plan adopts new", failures)
 
 
+def _scoring_pair_tests(my_predbat, failures):
+    """plan_scoring_pair: selection scores the pre-clip plans, and never mixes a clipped side with a pre-clip one.
+
+    Clipping sets the percentage actually sent to the inverter. It is a no-op in the expected case but moves the
+    metric through the PV10 branch by an amount that depends on plan shape, so comparing post-clip decides
+    between plans on a difference clipping invented rather than one the plans really have.
+    """
+    clipped_new = ([1.0], [{"start": 0, "end": 30}], [{"start": 30, "end": 60}], [100.0])
+    clipped_prev = ([2.0], [{"start": 0, "end": 30}], [{"start": 30, "end": 60}], [100.0])
+    preclip_new = ([3.0], [{"start": 0, "end": 30}], [{"start": 30, "end": 60}], [0.0])
+    preclip_prev = ([4.0], [{"start": 0, "end": 30}], [{"start": 30, "end": 60}], [0.0])
+
+    # Both pre-clip snapshots available: score the plans as optimised
+    pair = my_predbat.plan_scoring_pair(clipped_new, clipped_prev, preclip_new, preclip_prev)
+    _check(pair == (preclip_new, preclip_prev), "with both snapshots the pre-clip plans are scored", failures)
+
+    # No incumbent snapshot (first recompute after a restart): fall back to clipped on BOTH sides, never a mix,
+    # otherwise the new plan is scored untaxed against a taxed incumbent and wins on the difference
+    pair = my_predbat.plan_scoring_pair(clipped_new, clipped_prev, preclip_new, None)
+    _check(pair == (clipped_new, clipped_prev), "without an incumbent snapshot both sides fall back to clipped", failures)
+
+    pair = my_predbat.plan_scoring_pair(clipped_new, clipped_prev, None, preclip_prev)
+    _check(pair == (clipped_new, clipped_prev), "without a new snapshot both sides fall back to clipped", failures)
+
+
 def run_plan_tiebreak_tests(my_predbat):
     """Run the plan fragmentation tie-break tests. Returns True on failure."""
     print("**** Running plan tie-break tests ****")
     failures = []
     _fragmentation_tests(my_predbat, failures)
     _tiebreak_decision_tests(my_predbat, failures)
+    _scoring_pair_tests(my_predbat, failures)
     return len(failures) > 0

apps/predbat/unit_test.py

@@ -46,6 +46,7 @@
 from tests.test_optimise_levels import run_optimise_levels_tests
 from tests.test_trim_export import run_trim_export_tests
 from tests.test_plan_tiebreak import run_plan_tiebreak_tests
+from tests.test_plan_preclip import run_plan_preclip_tests
 from tests.test_export_commitment import run_export_commitment_tests
 from tests.test_energydataservice import run_energydataservice_tests
 from tests.test_iboost import run_iboost_smart_tests
@@ -454,6 +455,7 @@ def main():
         ("optimise_levels", run_optimise_levels_tests, "Optimise levels tests", False),
         ("trim_export", run_trim_export_tests, "Export trim ordering (buffer from cheapest slot) tests", False),
         ("plan_tiebreak", run_plan_tiebreak_tests, "Plan fragmentation near-tie tie-break tests", False),
+        ("plan_preclip", run_plan_preclip_tests, "Plan selection scores the pre-clip plan", True),
         ("export_commitment", run_export_commitment_tests, "Forced-export commitment / anti-flapping tests", False),
         ("load_ml", test_load_ml, "ML Load Forecaster tests (MLP, training, persistence, validation)", True),
         # ("optimise_windows", run_optimise_all_windows_tests, "Optimise all windows tests", True),