fix(prediction): collision-safe simulation cache key + clip windows in the C++ kernel (#4508)

* fix(prediction): make the simulation cache key collision-safe, clip windows in the kernel

Two changes to the per-simulation hot path, one of which turns out to be a
correctness fix rather than the optimisation it was meant to be.

The prediction cache key XORed a hash of every window bound together:

    window_hash ^= hash(window["start"]) ^ hash(window["end"])

XOR is commutative and self-cancelling, and hash(n) == n for the small
integers involved, so two windows sharing bounds cancel out entirely, window
order is invisible, and start/end values fold into each other. Different
scenarios therefore share a key and the cache hands back a result simulated
for a different plan. Instrumenting the benchmark, that is not theoretical:
186 collisions on one scenario and 828 on another, each one a wrong result fed
back into the optimiser. The key is now a single tuple hash, which is
order-sensitive and does not cancel. It is also cheaper, because the
per-element hashing happens in C rather than a Python loop, and it is only
computed when the cache is actually in play - a saving run always simulates,
so the key was previously built and discarded.

This changes plans: 3 of 20 benchmark scenarios move, 2 better and 1 worse.
That spread is what a cache no longer returning the wrong scenario's answer
looks like - the movement is the bug being removed, not a regression.

Separately, remove_intersecting_windows is now done inside the kernel
(clip_intersecting_charge_windows in prediction_kernel.cpp) instead of in
Python before every kernel call, where it cost more than the simulation it
preceded. The Python engine keeps its own copy, so the two must agree; both
parity revisions are bumped and all platform binaries rebuilt, since a stale
binary is refused and would silently drop users back to the Python engine.

The clipping port is behaviour-neutral, verified separately from the cache
fix: with the new key in place, adding the port leaves all 20 benchmark
scenarios bit-identical on metric and cost.

Parity is pinned by a new sweep. The existing random sweep places a few
windows at random, so an export window landing strictly inside a charge window
- the split case, and the subtlest part of clipping - almost never arises. The
new sweep positions export windows relative to charge windows on purpose
(covering, overlapping either end, inside, touching exactly at a boundary,
leaving a sub-5-minute remnant) and fails if it stops generating splits. 250
layouts, 33 of which split a charge window, all matching.

Benchmark: worst scenario 25.6s -> 18.2s, total optimise time across the 20
scenarios 87.1s -> 74.3s for the clipping port alone.

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

* chore: update prediction kernel binaries for all platforms

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

Author: 3 people

apps/predbat/prediction.py

@@ -19,6 +19,8 @@
 
 from datetime import timedelta
 from const import PREDICT_STEP, PV_SCENARIO_PV10, PV_SCENARIO_PV90, RUN_EVERY, TIME_FORMAT
+from operator import itemgetter
+
 from utils import remove_intersecting_windows, get_charge_rate_curve_cached, get_discharge_rate_curve_cached, find_charge_rate, calc_percent_limit, in_iboost_slot, in_car_slot, charge_curve_to_tuple
 from prediction_kernel import create_kernel_context, kernel_supported, run_prediction_kernel
 
@@ -92,6 +94,11 @@ def get_total_inverted(battery_draw, pv_dc, pv_ac, inverter_loss, inverter_hybri
     return total_inverted
 
 
+# Pulls (start, end) from a window dict; used to build the prediction cache key without a Python
+# level loop over every window
+window_bounds = itemgetter("start", "end")
+
+
 class Prediction:
     """
     Class to hold prediction input and output data and the run function
@@ -409,24 +416,34 @@ def run_prediction(self, charge_limit, charge_window, export_window, export_limi
         KERNEL_PARITY_REVISION (prediction_kernel.py) and PK_PARITY_REVISION (prediction_kernel.cpp)
         must both be bumped, and the kernel_parity test must pass (cd coverage && ./run_all --test kernel_parity).
         """
-        window_hash = 0
-        for window in charge_window:
-            window_hash ^= hash(window["start"]) ^ hash(window["end"])
-        for window in export_window:
-            window_hash ^= hash(window["start"]) ^ hash(window["end"])
-
-        sim_hash = hash(tuple(charge_limit)) ^ window_hash ^ hash(tuple(export_limits)) ^ hash(pv_scenario) ^ hash(end_record) ^ hash(step)
-
-        if not save and cache and sim_hash in self.prediction_cache:
-            # Return cached result
-            return self.prediction_cache[sim_hash]
+        # The cache key is only wanted when the cache is actually in play - a saving run always
+        # simulates - so it is not computed otherwise. Building it as one tuple hash keeps the
+        # per-window hashing in C rather than looping in Python, which matters because this runs on
+        # every simulation with a few hundred windows.
+        sim_hash = None
+        if cache and not save:
+            sim_hash = hash(
+                (
+                    tuple(charge_limit),
+                    tuple(map(window_bounds, charge_window)),
+                    tuple(export_limits),
+                    tuple(map(window_bounds, export_window)),
+                    pv_scenario,
+                    end_record,
+                    step,
+                )
+            )
+            cached_result = self.prediction_cache.get(sim_hash)
+            if cached_result is not None:
+                # Return cached result
+                return cached_result
 
         # Try the C++ prediction kernel first; unsupported scenarios fall through to the Python engine.
         # The kernel understands all three pv_scenario values (see PkScenario.pv_scenario, ABI 3).
         if kernel_supported(self, save, step):
             kernel_result = run_prediction_kernel(self, charge_limit, charge_window, export_window, export_limits, pv_scenario, end_record, step, cache)
             if kernel_result is not None:
-                if not save and cache:
+                if sim_hash is not None:
                     # Store in cache without the SoC/car data to save memory, mirroring the Python engine
                     self.prediction_cache[sim_hash] = kernel_result[:11] + ([], []) + kernel_result[13:]
                 return kernel_result
@@ -1301,7 +1318,7 @@ def run_prediction(self, charge_limit, charge_window, export_window, export_limi
             self.import_kwh_time = import_kwh_time
             self.export_kwh_time = export_kwh_time
 
-        if not save and cache:
+        if sim_hash is not None:
             self.prediction_cache[sim_hash] = (
                 round(final_metric, 4),
                 round(final_import_kwh_battery, 4),

apps/predbat/prediction_kernel.cpp

@@ -27,7 +27,7 @@
 #include <vector>
 
 #define PK_ABI_VERSION 3
-#define PK_PARITY_REVISION 4
+#define PK_PARITY_REVISION 5
 #define PK_MAX_CARS 4
 #define PK_RUN_EVERY 5 // const.py RUN_EVERY
 
@@ -234,6 +234,92 @@ inline double rate_curve(double soc_key, double rate_setting, double rate_max, d
 // Build per-step window membership, mirroring Prediction.find_charge_window_optimised():
 // dict keyed by absolute minute stepping 5 from each window start, last window wins,
 // looked up at minute_absolute = minutes_now + k*5 (so misaligned windows never match).
+// Mirrors remove_intersecting_windows() in utils.py, which the Python engine applies before
+// simulating. Charge windows that collide with an enabled export window are trimmed, and a window
+// with an export landing inside it is split in two. Two rules are easy to get wrong and are pinned
+// by tests: a window that was never clipped survives whatever its length, while a remnant clipping
+// itself created is dropped below 5 minutes; and windows that merely touch at a boundary overlap
+// arithmetically but remove nothing, so they must not count as clipped.
+//
+// PARITY: any change here must be mirrored in utils.remove_intersecting_windows and vice versa.
+static void clip_intersecting_charge_windows(std::vector<int32_t> &out_start, std::vector<int32_t> &out_end, std::vector<double> &out_limit, int32_t n_charge, const int32_t *charge_start, const int32_t *charge_end, const double *charge_limit, int32_t n_export, const int32_t *export_start,
+                                             const int32_t *export_end, const double *export_limits)
+{
+    // Enabled export windows only - the sole candidates for clipping anything - in start order
+    std::vector<std::pair<int32_t, int32_t>> export_active;
+    export_active.reserve(n_export);
+    for (int32_t n = 0; n < n_export; n++) {
+        if (export_limits[n] < 100.0) {
+            export_active.emplace_back(export_start[n], export_end[n]);
+        }
+    }
+    std::sort(export_active.begin(), export_active.end());
+
+    out_start.clear();
+    out_end.clear();
+    out_limit.clear();
+    out_start.reserve(n_charge);
+    out_end.reserve(n_charge);
+    out_limit.reserve(n_charge);
+
+    if (export_active.empty()) {
+        for (int32_t n = 0; n < n_charge; n++) {
+            out_start.push_back(charge_start[n]);
+            out_end.push_back(charge_end[n]);
+            out_limit.push_back(charge_limit[n]);
+        }
+        return;
+    }
+
+    for (int32_t n = 0; n < n_charge; n++) {
+        int32_t start = charge_start[n];
+        int32_t end = charge_end[n];
+        const double limit = charge_limit[n];
+
+        if (!(limit > 0.0)) {
+            // A disabled charge window can never be clipped
+            out_start.push_back(start);
+            out_end.push_back(end);
+            out_limit.push_back(limit);
+            continue;
+        }
+
+        bool clipped = false;
+        for (const auto &dw : export_active) {
+            const int32_t dstart = dw.first;
+            const int32_t dend = dw.second;
+            if ((dstart < end) && (dend >= start)) {
+                if (dstart <= start) {
+                    if (start != dend) {
+                        start = dend;
+                        clipped = true;
+                    }
+                } else if (dend >= end) {
+                    if (end != dstart) {
+                        end = dstart;
+                        clipped = true;
+                    }
+                } else {
+                    // Two segments - emit the head now, carry on clipping the tail
+                    if ((dstart - start) >= 5) {
+                        out_start.push_back(start);
+                        out_end.push_back(dstart);
+                        out_limit.push_back(limit);
+                    }
+                    start = dend;
+                    clipped = true;
+                }
+            }
+        }
+
+        if (!clipped || ((end - start) >= 5)) {
+            out_start.push_back(start);
+            out_end.push_back(end);
+            out_limit.push_back(limit);
+        }
+    }
+}
+
 void build_window_membership(std::vector<int32_t> &member, int32_t n_windows, const int32_t *starts, const int32_t *ends, const double *limits, bool is_export, int32_t minutes_now, int32_t n_steps)
 {
     member.assign(n_steps, -1);
@@ -360,7 +446,15 @@ int32_t pk_run(int64_t handle, const PkScenario *s, PkResult *out)
 
     // Window membership - prediction.py:494-495 / find_charge_window_optimised
     std::vector<int32_t> charge_window_optimised, export_window_optimised;
-    build_window_membership(charge_window_optimised, s->n_charge, s->charge_start, s->charge_end, s->charge_limit, false, c->minutes_now, n_steps);
+
+    // The caller hands over the raw charge windows; clipping them against the export windows used to
+    // be done in Python on every simulation, which cost more than the simulation itself
+    std::vector<int32_t> clipped_start, clipped_end;
+    std::vector<double> clipped_limit;
+    clip_intersecting_charge_windows(clipped_start, clipped_end, clipped_limit, s->n_charge, s->charge_start, s->charge_end, s->charge_limit, s->n_export, s->export_start, s->export_end, s->export_limits);
+    const int32_t n_charge_clipped = static_cast<int32_t>(clipped_start.size());
+
+    build_window_membership(charge_window_optimised, n_charge_clipped, clipped_start.data(), clipped_end.data(), clipped_limit.data(), false, c->minutes_now, n_steps);
     build_window_membership(export_window_optimised, s->n_export, s->export_start, s->export_end, s->export_limits, true, c->minutes_now, n_steps);
 
     // Initial state - prediction.py:435-490
@@ -461,7 +555,7 @@ int32_t pk_run(int64_t handle, const PkScenario *s, PkResult *out)
         // Find charge limit - prediction.py:609-620
         double charge_limit_n = 0;
         if (charge_window_active) {
-            charge_limit_n = s->charge_limit[charge_window_n];
+            charge_limit_n = clipped_limit[charge_window_n];
             if (c->set_charge_freeze && (calc_percent_limit(charge_limit_n, soc_max) == reserve_percent)) {
                 // Charge freeze via reserve
                 charge_limit_n = std::max(soc, reserve);

apps/predbat/prediction_kernel.py

@@ -27,11 +27,11 @@
 import weakref
 
 from const import PREDICT_STEP
-from utils import remove_intersecting_windows, get_curve_value, find_battery_temperature_cap, in_car_slot, in_iboost_slot
+from utils import get_curve_value, find_battery_temperature_cap, in_car_slot, in_iboost_slot
 
 # Expected ABI/parity revisions of the shared library (see prediction_kernel.cpp)
 KERNEL_ABI_VERSION = 3
-KERNEL_PARITY_REVISION = 4
+KERNEL_PARITY_REVISION = 5
 
 # Maximum number of cars supported by the kernel (PK_MAX_CARS in prediction_kernel.cpp)
 KERNEL_MAX_CARS = 4
@@ -466,8 +466,9 @@ def run_prediction_kernel(pred, charge_limit, charge_window, export_window, expo
     if not lib:
         return None
 
-    # Remove intersecting windows, mirroring the Python engine - prediction.py:492-493
-    charge_limit, charge_window = remove_intersecting_windows(charge_limit, charge_window, export_limits, export_window)
+    # The kernel clips intersecting windows itself (clip_intersecting_charge_windows in
+    # prediction_kernel.cpp), mirroring what the Python engine does before simulating. Doing it in
+    # Python here cost more per simulation than the simulation, so the raw windows are handed over.
 
     n_steps = pred.forecast_minutes // PREDICT_STEP
     scenario = PkScenario()

apps/predbat/prediction_kernel_lib_aarch64.so

@@ -29,6 +29,7 @@
 from const import PV_SCENARIO_NOMINAL, PV_SCENARIO_PV10, PV_SCENARIO_PV90
 from prediction import Prediction
 from prediction_kernel import create_kernel_context, run_prediction_kernel, load_kernel
+from utils import remove_intersecting_windows
 from tests.test_infra import reset_inverter, reset_rates
 from tests.test_model import run_model_tests
 
@@ -687,6 +688,119 @@ def run_random_sweep_tests(my_predbat, count=150):
     return failed
 
 
+def make_intersecting_windows(rng, minutes_now, forecast_minutes):
+    """Build a charge/export layout that deliberately exercises window clipping.
+
+    The generic sweep places a handful of windows at random, so an export window landing strictly
+    inside a charge window - the split case, and the subtlest part of the clipping - almost never
+    comes up. Here export windows are positioned relative to the charge windows on purpose: covering
+    them entirely, overlapping either end, sitting inside them, and touching exactly at a boundary.
+    Short windows and short remnants are included because the 5 minute minimum only applies to
+    remnants clipping itself created, never to a window that was left alone.
+    """
+    charge_window = []
+    minute = minutes_now
+    for _ in range(rng.randint(1, 5)):
+        length = rng.choice([5, 10, 30, 60, 120, 240])
+        end = min(minute + length, minutes_now + forecast_minutes)
+        if end <= minute:
+            break
+        charge_window.append({"start": minute, "end": end, "average": round(rng.uniform(0, 40), 2)})
+        minute = end + rng.choice([0, 0, 5, 30])
+        if minute >= minutes_now + forecast_minutes:
+            break
+
+    export_window = []
+    for window in charge_window:
+        mode = rng.choice(["inside", "inside", "overlap_start", "overlap_end", "cover", "touch_start", "touch_end", "clear", "tiny_remnant"])
+        start, end = window["start"], window["end"]
+        span = end - start
+        if mode == "inside" and span >= 20:
+            dstart = start + rng.randrange(5, max(span - 10, 6), 5)
+            dend = min(dstart + rng.choice([5, 10, 30]), end - 1)
+            if dend <= dstart:
+                continue
+        elif mode == "overlap_start":
+            dstart, dend = max(start - rng.choice([5, 30]), minutes_now), start + max(span // 3, 5)
+        elif mode == "overlap_end":
+            dstart, dend = end - max(span // 3, 5), end + rng.choice([5, 30])
+        elif mode == "cover":
+            dstart, dend = start, end
+        elif mode == "touch_start":
+            dstart, dend = max(start - 30, minutes_now), start
+        elif mode == "touch_end":
+            dstart, dend = end, end + 30
+        elif mode == "tiny_remnant" and span >= 10:
+            # Leave only a couple of minutes at the end, below the 5 minute minimum
+            dstart, dend = start, end - rng.choice([1, 2, 3])
+        else:
+            dstart, dend = end + 60, end + 90
+        dstart = max(min(dstart, minutes_now + forecast_minutes), minutes_now)
+        dend = max(min(dend, minutes_now + forecast_minutes), dstart)
+        if dend > dstart:
+            export_window.append({"start": dstart, "end": dend, "average": round(rng.uniform(0, 40), 2)})
+
+    export_window.sort(key=lambda w: w["start"])
+    return charge_window, export_window
+
+
+def run_clipping_parity_tests(my_predbat, count=250):
+    """Compare both engines on layouts built to exercise window clipping, returns True on failure.
+
+    The kernel clips intersecting charge windows itself (clip_intersecting_charge_windows in
+    prediction_kernel.cpp) rather than having Python do it first, so this is the sweep that pins
+    those two implementations together.
+    """
+    failed = False
+    split_layouts = 0
+    for seed in range(count):
+        rng = random.Random(500000 + seed)
+        reset_inverter(my_predbat)
+        reset_rates(my_predbat, 10.0, 5.0)
+        my_predbat.battery_rate_max_export = my_predbat.battery_rate_max_discharge
+        apply_random_scenario(my_predbat, rng)
+        pv_step, pv10_step, load_step, load10_step = make_step_data(my_predbat, rng=rng)
+
+        charge_window, export_window = make_intersecting_windows(rng, my_predbat.minutes_now, my_predbat.forecast_minutes)
+        charge_limit = [rng.choice([0.0, my_predbat.reserve, my_predbat.soc_max, round(rng.uniform(0, my_predbat.soc_max), 2)]) for _ in charge_window]
+        export_limits = [rng.choice([100.0, 99.0, 0.0, round(rng.uniform(0, 100), 1)]) for _ in export_window]
+        end_record = rng.choice([my_predbat.forecast_minutes, my_predbat.forecast_minutes - 30])
+
+        # Count the layouts that actually split a charge window, so a generator that stopped
+        # producing them would show up rather than silently weakening this sweep
+        clipped_limits, clipped_windows = remove_intersecting_windows(charge_limit, charge_window, export_limits, export_window)
+        if len(clipped_windows) > len(charge_window):
+            split_layouts += 1
+
+        for pv_scenario in (PV_SCENARIO_NOMINAL, PV_SCENARIO_PV10):
+            failed |= dual_run(
+                "clip_{}_s{}".format(seed, pv_scenario),
+                my_predbat,
+                pv_step,
+                pv10_step,
+                load_step,
+                load10_step,
+                charge_limit,
+                charge_window,
+                export_window,
+                export_limits,
+                pv_scenario,
+                end_record,
+            )
+            if failed:
+                print("Clipping sweep failed at seed {} scenario {}".format(seed, pv_scenario))
+                print("   charge {} limits {}".format([(w["start"], w["end"]) for w in charge_window], charge_limit))
+                print("   export {} limits {}".format([(w["start"], w["end"]) for w in export_window], export_limits))
+                break
+        if failed:
+            break
+    print("Clipping sweep ran {} layouts, {} of which split a charge window".format(count, split_layouts))
+    if split_layouts == 0:
+        print("ERROR: clipping sweep generated no window splits - the sweep is not exercising the split path")
+        failed = True
+    return failed
+
+
 def kernel_available():
     """Ensure the kernel library is built and loaded, returns (available, required_failure)"""
     if not ensure_kernel_built():
@@ -729,6 +843,8 @@ def run_kernel_parity_tests(my_predbat):
         failed = run_edge_case_tests(my_predbat)
         if not failed:
             failed |= run_random_sweep_tests(my_predbat)
+        if not failed:
+            failed |= run_clipping_parity_tests(my_predbat)
     finally:
         restore_scenario_state(my_predbat, state)