Break near-tie plan comparison toward the less fragmented plan

apps/predbat/plan.py

@@ -907,6 +907,50 @@ def in_charge_window(self, charge_window, minute_abs):
             window_n += 1
         return -1
 
+    def plan_fragmentation(self, charge_window, charge_limit, export_window, export_limits):
+        """Count the contiguous active (charge/export) segments in a plan.
+
+        A slot is active if it discharges the battery (export limit < 99, i.e. not freeze/off) or charges it
+        (charge target above the reserve floor). Time-adjacent active slots of the same mode form one segment; a
+        time gap or a change of mode (charge<->export) starts a new segment. A cleaner plan has fewer segments,
+        so this is used as a tie-break to prefer a single export block over a fragmented staircase when the cost
+        is otherwise equal.
+        """
+        intervals = []
+        for window, limit in zip(export_window, export_limits):
+            if limit < 99:
+                intervals.append((window["start"], window["end"], "export"))
+        for window, limit in zip(charge_window, charge_limit):
+            if limit > self.reserve:
+                intervals.append((window["start"], window["end"], "charge"))
+
+        intervals.sort(key=lambda item: item[0])
+
+        segments = 0
+        prev_end = None
+        prev_mode = None
+        for start, end, mode in intervals:
+            if prev_end is None or start > prev_end or mode != prev_mode:
+                segments += 1
+            prev_end = end if prev_end is None else max(prev_end, end)
+            prev_mode = mode
+        return segments
+
+    def should_replace_plan(self, metric_prev, metric_new, fragmentation_prev, fragmentation_new):
+        """Decide whether to adopt the freshly optimised plan over the incumbent.
+
+        The new plan is adopted when it is better by at least metric_min_improvement_plan (the existing
+        anti-jitter behaviour). On a near-tie within that band it is also adopted when it is no worse on cost
+        and strictly less fragmented, so a cleaner single export block can replace a locked-in split schedule
+        without churning the plan for tiny cost changes. Lower metric is better, so improvement is prev - new.
+        """
+        improvement = metric_prev - metric_new
+        if improvement >= self.metric_min_improvement_plan:
+            return True
+        if improvement >= 0 and fragmentation_new < fragmentation_prev:
+            return True
+        return False
+
     def calculate_plan(self, recompute=True, debug_mode=False, publish=True):
         """
         Calculate the new plan (best)
@@ -1179,14 +1223,18 @@ def calculate_plan(self, recompute=True, debug_mode=False, publish=True):
                 )
 
                 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)))
-                if (metric_prev - metric) < self.metric_min_improvement_plan:
+                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)
+                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)
-                else:
+                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))
 
             # Plan is now valid
             self.log("Plan valid is now true after recompute was {}".format(self.plan_valid))

apps/predbat/tests/test_plan_tiebreak.py

@@ -0,0 +1,88 @@
+# -----------------------------------------------------------------------------
+# 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 for the near-tie plan fragmentation tie-break used in calculate_plan().
+
+When a freshly optimised plan is not better than the incumbent by metric_min_improvement_plan,
+Predbat keeps the incumbent. Under flat export rates that locks in a fragmented (split) export
+schedule even though the fresh plan is a cleaner single block. The tie-break lets the cleaner
+plan win a near-tie provided it is no worse on cost.
+"""
+
+
+def _check(cond, message, failures):
+    """Record a failure message if the condition is not met."""
+    if not cond:
+        print("ERROR: {}".format(message))
+        failures.append(message)
+
+
+def _fragmentation_tests(my_predbat, failures):
+    """plan_fragmentation counts contiguous active (charge/export) segments split by gaps/mode changes."""
+    my_predbat.reserve = 0.4
+
+    # Three adjacent 30-minute export windows all discharging = one contiguous block = 1 segment.
+    ew = [{"start": 0, "end": 30}, {"start": 30, "end": 60}, {"start": 60, "end": 90}]
+    _check(my_predbat.plan_fragmentation([], [], ew, [0.0, 0.0, 0.0]) == 1, "contiguous export block should be 1 segment", failures)
+
+    # Same span but the middle slot is off (100) -> a time gap -> two separate export runs = 2 segments.
+    _check(my_predbat.plan_fragmentation([], [], ew, [0.0, 100.0, 0.0]) == 2, "export split by an idle slot should be 2 segments", failures)
+
+    # Staircase targets (all discharging, all adjacent) is still one block - differing depth is not a split.
+    _check(my_predbat.plan_fragmentation([], [], ew, [28.0, 7.0, 6.0]) == 1, "adjacent discharge windows are one segment regardless of depth", failures)
+
+    # Freeze (99) and off (100) are not battery export, so they are inactive -> 0 segments.
+    _check(my_predbat.plan_fragmentation([], [], ew, [99.0, 100.0, 99.0]) == 0, "freeze/off export windows are inactive", failures)
+
+    # A charge window (target above reserve) adjacent to an export window is a mode change = 2 segments.
+    cw = [{"start": 0, "end": 30}]
+    ew2 = [{"start": 30, "end": 60}]
+    _check(my_predbat.plan_fragmentation(cw, [5.0], ew2, [0.0]) == 2, "adjacent charge then export is a mode change = 2 segments", failures)
+
+    # A charge window at/below reserve is not charging -> inactive.
+    _check(my_predbat.plan_fragmentation([{"start": 0, "end": 30}], [0.4], [], []) == 0, "charge target at reserve is inactive", failures)
+
+    # The real scenario: split evening (3 runs with gaps) must be more fragmented than the merged block (1 run).
+    split_ew = [{"start": 1250, "end": 1350}, {"start": 1370, "end": 1410}, {"start": 1425, "end": 1440}]
+    merged_ew = [{"start": 1250, "end": 1410}]
+    frag_split = my_predbat.plan_fragmentation([], [], split_ew, [28.0, 7.0, 6.0])
+    frag_merged = my_predbat.plan_fragmentation([], [], merged_ew, [0.0])
+    _check(frag_split > frag_merged, "split evening ({}) should be more fragmented than merged ({})".format(frag_split, frag_merged), failures)
+
+
+def _tiebreak_decision_tests(my_predbat, failures):
+    """should_replace_plan: adopt new on a clear win, or on a near-tie if no worse and strictly cleaner."""
+    my_predbat.metric_min_improvement_plan = 2.0
+
+    # Clearly better (gap >= threshold) always adopts the new plan, regardless of fragmentation.
+    _check(my_predbat.should_replace_plan(-100.0, -103.0, 1, 5) is True, "clear improvement adopts new plan", failures)
+
+    # Real near-tie from the trefor2 case: prev split -224.78, new merged -226.57 (1.79p better), cleaner.
+    _check(my_predbat.should_replace_plan(-224.78, -226.57, 5, 1) is True, "near-tie cleaner+cheaper plan adopts new", failures)
+
+    # Near-tie but equally fragmented -> keep previous (preserves anti-jitter hysteresis).
+    _check(my_predbat.should_replace_plan(-224.78, -226.57, 3, 3) is False, "near-tie equal fragmentation keeps previous", failures)
+
+    # Near-tie, new is cleaner but MORE expensive (gap < 0) -> keep previous (never raise cost).
+    _check(my_predbat.should_replace_plan(-226.57, -224.78, 5, 1) is False, "cleaner but costlier plan is rejected", failures)
+
+    # Near-tie, new is cheaper but MORE fragmented -> keep previous.
+    _check(my_predbat.should_replace_plan(-224.78, -225.0, 1, 4) is False, "cheaper but more fragmented plan is rejected", failures)
+
+    # Exactly cost-neutral (gap == 0) but cleaner -> adopt new.
+    _check(my_predbat.should_replace_plan(-200.0, -200.0, 4, 2) is True, "cost-neutral cleaner plan adopts new", 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)
+    return len(failures) > 0

apps/predbat/unit_test.py

@@ -43,6 +43,7 @@
 from tests.test_component_health_status import test_component_health_status
 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_export_commitment import run_export_commitment_tests
 from tests.test_energydataservice import run_energydataservice_tests
 from tests.test_iboost import run_iboost_smart_tests
@@ -386,6 +387,7 @@ def main():
         ("gateway", run_gateway_tests, "GatewayMQTT component tests (protobuf, plan serialization, commands, telemetry)", False),
         ("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),
         ("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),