perf(plan): cache export-to-charge window collisions in the optimiser (1.5x faster planning) (#4507)

* perf(plan): cache export-to-charge window collisions in the optimiser

optimise_charge_limit_price_threads asks hit_charge_window which charge window
each export window collides with, once per export window per candidate, and
hit_charge_window is a linear scan of the whole charge window list. On a
benchmark scenario that was 1,741,593 scans of a ~200 entry list for 221
distinct answers - 27% of the plan's runtime, and the largest single cost after
the intersect fix in #4505.

The collision is purely geometric, and this function only ever turns windows on
and off: it never moves a window's start or end. So the answer cannot change
for the life of the call and is memoised in a local dict keyed by export window.
Only the collision index is cached - the charge limit derived from it depends on
charge_mods/best_limits_reset and still varies per trial. Scoping the cache to
the call means there is nothing to invalidate.

Confirmed by instrumentation before making the change: 1.74M calls, 221 distinct
queries, and zero cases of a repeated query returning a different answer later.

Random benchmark: mean optimise time 4.302s -> 2.896s (1.5x), worst scenario
25.6s -> 18.1s, with plan metric and cost identical on all 20 scenarios.

Adds tests for hit_charge_window's contract - overlap, boundary-touch, and that
the first matching window is returned, since the cache stores a specific index -
plus a test asserting the invariant the cache rests on: run the optimiser and
check the window bounds it was given come back untouched. If a future change
starts moving windows, that test fails rather than the cache quietly returning
stale collisions.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* test(random): refresh the random benchmark reference for the collision cache

cases/random_results.json is what run_random compares against, and it carries
the recorded runtime of each scenario as well as its metric and cost. No seed
differs on metric or cost - the cache does not change any plan - but the stored
runtimes were recorded before it, so run_random reported a permanent ~1.4s of
phantom improvement on every run.

Regenerated: total recorded runtime 86.0s -> 60.2s, metric and cost unchanged
on all 20 scenarios.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* [pre-commit.ci lite] apply automatic fixes

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: pre-commit-ci-lite[bot] <117423508+pre-commit-ci-lite[bot]@users.noreply.github.com>

Author: 3 people

apps/predbat/plan.py

@@ -392,6 +392,15 @@ def optimise_charge_limit_price_threads(
             reset_contribution = scenario_hash_entry(1, window_n, best_export_limits_reset[window_n])
             export_hash_delta[window_n] = {True: scenario_hash_entry(1, window_n, 99.0) - reset_contribution, False: scenario_hash_entry(1, window_n, min_freeze_percent) - reset_contribution}
 
+        # Which charge window an export window collides with is a purely geometric question, and this
+        # function only ever turns windows on and off - it never moves a window's start or end. So the
+        # answer is fixed for the life of the call and is memoised here, keyed by export window. Without
+        # it the trial loop below re-scans the whole charge window list for every export window of every
+        # candidate: on a benchmark scenario that was 1.74 million linear scans of a ~200 entry list,
+        # for 221 distinct answers. Only the collision itself is cached - the limit that follows from it
+        # depends on charge_mods/best_limits_reset and changes from trial to trial.
+        hit_charge_cache = {}
+
         # Start loop of trials
         for loop_price in all_prices:
             if best_level_score is not None:
@@ -454,7 +463,10 @@ def optimise_charge_limit_price_threads(
                                 # Remove export hitting charge windows if this is disabled
                                 if not self.calculate_export_oncharge:
                                     for window_n in all_d[:]:
-                                        hit_charge = self.hit_charge_window(self.charge_window_best, export_window[window_n]["start"], export_window[window_n]["end"])
+                                        hit_charge = hit_charge_cache.get(window_n)
+                                        if hit_charge is None:
+                                            hit_charge = self.hit_charge_window(self.charge_window_best, export_window[window_n]["start"], export_window[window_n]["end"])
+                                            hit_charge_cache[window_n] = hit_charge
                                         if hit_charge >= 0:
                                             if hit_charge in charge_mods:
                                                 hit_charge_limit = self.reserve if charge_mods[hit_charge] else self.soc_max

apps/predbat/tests/test_hit_charge_cache.py

@@ -0,0 +1,168 @@
+# -----------------------------------------------------------------------------
+# 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 export-to-charge window collision cache in optimise_charge_limit_price_threads.
+
+The cache is only sound because window geometry is immutable for the life of that call - the
+optimiser turns windows on and off but never moves a start or end. These tests pin both halves:
+hit_charge_window's own contract, and the invariant that lets its result be reused.
+"""
+from prediction import Prediction
+from tests.test_infra import reset_inverter, reset_rates
+
+
+def run_hit_charge_cache_tests(my_predbat):
+    """Run the hit_charge_window and collision-cache invariant tests"""
+    failed = False
+    failed |= test_hit_charge_window_finds_overlap(my_predbat)
+    failed |= test_hit_charge_window_boundaries_do_not_overlap(my_predbat)
+    failed |= test_hit_charge_window_returns_first_match(my_predbat)
+    failed |= test_optimise_threads_does_not_move_window_bounds(my_predbat)
+    return failed
+
+
+def make_window(start, end, average=10.0):
+    """Build a minimal window dict"""
+    return {"start": start, "end": end, "average": average}
+
+
+def test_hit_charge_window_finds_overlap(my_predbat):
+    """hit_charge_window returns the index of the charge window an interval overlaps, else -1"""
+    print("**** test_hit_charge_window_finds_overlap ****")
+    failed = False
+    reset_inverter(my_predbat)
+
+    windows = [make_window(0, 30), make_window(60, 90), make_window(120, 180)]
+
+    for start, end, expect in ((0, 30, 0), (10, 20, 0), (25, 65, 0), (70, 80, 1), (150, 200, 2), (30, 60, -1), (200, 300, -1)):
+        got = my_predbat.hit_charge_window(windows, start, end)
+        if got != expect:
+            print("ERROR: hit_charge_window({}, {}) expected {} but got {}".format(start, end, expect, got))
+            failed = True
+
+    if not failed:
+        print("PASS")
+    return failed
+
+
+def test_hit_charge_window_boundaries_do_not_overlap(my_predbat):
+    """An interval that merely touches a charge window at a boundary is not a collision"""
+    print("**** test_hit_charge_window_boundaries_do_not_overlap ****")
+    failed = False
+    reset_inverter(my_predbat)
+
+    windows = [make_window(60, 90)]
+
+    if my_predbat.hit_charge_window(windows, 30, 60) != -1:
+        print("ERROR: interval ending exactly at the window start reported a collision")
+        failed = True
+    if my_predbat.hit_charge_window(windows, 90, 120) != -1:
+        print("ERROR: interval starting exactly at the window end reported a collision")
+        failed = True
+    if my_predbat.hit_charge_window(windows, 59, 61) != 0:
+        print("ERROR: interval straddling the window start did not report a collision")
+        failed = True
+
+    if not failed:
+        print("PASS")
+    return failed
+
+
+def test_hit_charge_window_returns_first_match(my_predbat):
+    """When several charge windows overlap the interval, the lowest index is returned - the cache
+    stores this answer, so it has to be deterministic rather than any-match"""
+    print("**** test_hit_charge_window_returns_first_match ****")
+    failed = False
+    reset_inverter(my_predbat)
+
+    windows = [make_window(0, 30), make_window(30, 60), make_window(60, 90)]
+    got = my_predbat.hit_charge_window(windows, 10, 80)
+    if got != 0:
+        print("ERROR: Expected the first overlapping window (0) but got {}".format(got))
+        failed = True
+
+    if not failed:
+        print("PASS")
+    return failed
+
+
+def test_optimise_threads_does_not_move_window_bounds(my_predbat):
+    """The collision cache assumes optimise_charge_limit_price_threads never moves a window's start
+    or end - it only enables and disables windows. Run the optimiser and assert the geometry it was
+    given comes back untouched; if a future change starts moving windows, this fails rather than the
+    cache silently returning stale collisions."""
+    print("**** test_optimise_threads_does_not_move_window_bounds ****")
+    failed = False
+    reset_inverter(my_predbat)
+    reset_rates(my_predbat, 10.0, 5.0)
+
+    # Set the metric config this path reads rather than depending on whatever earlier tests left
+    # behind, and restore it afterwards so this test neither depends on nor leaks ambient state
+    saved = {name: getattr(my_predbat, name) for name in ("metric_battery_value_export_scaling", "metric_battery_value_scaling", "pv_metric10_weight", "pv_metric90_weight")}
+    my_predbat.metric_battery_value_export_scaling = 1.0
+    my_predbat.metric_battery_value_scaling = 1.0
+    my_predbat.pv_metric10_weight = 0.0
+    my_predbat.pv_metric90_weight = 0.0
+
+    # The optimiser simulates, so it needs a prediction with flat PV/load data behind it
+    pv_step = {}
+    load_step = {}
+    for minute in range(0, my_predbat.forecast_minutes, 5):
+        pv_step[minute] = 0.0
+        load_step[minute] = 0.1
+    my_predbat.pv_forecast_minute_step = pv_step
+    my_predbat.load_minutes_step = load_step
+    my_predbat.prediction = Prediction(my_predbat, pv_step, pv_step, load_step, load_step)
+    my_predbat.args["threads"] = 0
+
+    my_predbat.calculate_export_oncharge = False
+    charge_window = [make_window(my_predbat.minutes_now + 60 * n, my_predbat.minutes_now + 60 * n + 30, 10.0) for n in range(4)]
+    export_window = [make_window(my_predbat.minutes_now + 60 * n + 30, my_predbat.minutes_now + 60 * n + 60, 15.0) for n in range(4)]
+    charge_limit = [my_predbat.soc_max] * len(charge_window)
+    export_limits = [100.0] * len(export_window)
+
+    my_predbat.charge_window_best = charge_window
+    my_predbat.charge_limit_best = charge_limit
+    my_predbat.export_window_best = export_window
+    my_predbat.export_limits_best = export_limits
+
+    before_charge = [(w["start"], w["end"]) for w in charge_window]
+    before_export = [(w["start"], w["end"]) for w in export_window]
+
+    window_sorted, window_index, price_set, price_links = my_predbat.sort_window_by_price_combined(charge_window, export_window)
+    try:
+        my_predbat.optimise_charge_limit_price_threads(
+            price_set,
+            price_links,
+            window_index,
+            len(charge_window),
+            len(export_window),
+            charge_limit,
+            charge_window,
+            export_window,
+            export_limits,
+            end_record=my_predbat.end_record,
+            quiet=True,
+        )
+    finally:
+        for name, value in saved.items():
+            setattr(my_predbat, name, value)
+
+    after_charge = [(w["start"], w["end"]) for w in my_predbat.charge_window_best]
+    after_export = [(w["start"], w["end"]) for w in my_predbat.export_window_best]
+    if after_charge != before_charge:
+        print("ERROR: charge window bounds moved during optimisation: {} -> {}".format(before_charge, after_charge))
+        failed = True
+    if after_export != before_export:
+        print("ERROR: export window bounds moved during optimisation: {} -> {}".format(before_export, after_export))
+        failed = True
+
+    if not failed:
+        print("PASS")
+    return failed

apps/predbat/unit_test.py

@@ -96,6 +96,7 @@
     test_web_annual_validation_error_preserves_input,
 )
 from tests.test_window import run_window_sort_tests, run_intersect_window_tests
+from tests.test_hit_charge_cache import run_hit_charge_cache_tests
 from tests.test_find_charge_rate import test_find_charge_rate, test_find_charge_rate_pv_overlap, test_find_charge_rate_string_temperature, test_find_charge_rate_string_charge_curve
 from tests.test_manual_api import run_test_manual_api
 from tests.test_manual_soc import run_test_manual_soc
@@ -402,6 +403,7 @@ def main():
         ("iboost_smart", run_iboost_smart_tests, "iBoost smart tests", False),
         ("car_charging_smart", run_car_charging_smart_tests, "Car charging smart tests", False),
         ("intersect_window", run_intersect_window_tests, "Intersect window tests", False),
+        ("hit_charge_cache", run_hit_charge_cache_tests, "Hit charge window cache tests", False),
         ("inverter_multi", run_inverter_multi_tests, "Inverter multi tests", False),
         ("octopus_free", test_octopus_free, "Octopus free electricity tests", False),
         ("battery_curve_keys", run_battery_curve_keys_tests, "Battery curve keys tests", False),