fix(pv90): keep the p90 series in step with its p50, clamp the synthesised p90 to the array ceiling, document the weight

Fixes the five findings from the final whole-branch review. pv_metric90_weight
stays at its default of 0.0, so the feature still ships inert; every fix here
protects the expert who opts in.

1. pv_forecast_minute90 could go permanently stale. The old guard only fired on
   an EMPTY p90, so a caller that reassigns pv_forecast_minute directly - most
   importantly annual.py, which reuses ONE PredBat instance across every sampled
   day of a year - pinned every later day's "upside" to the first day's solar,
   turning pv90 into a severe downside case. annual.py now assigns the p90 on
   every sampled day, and plan.py's guard (now refresh_pv_forecast_minute90())
   re-derives the p90 whenever it fails to cover the plan horizon or has been
   left behind by a p50 that moved without it. A real p90 that moves with its
   p50 is never touched. Production is unaffected: fetch.py always builds all
   three series together over one shared minute range.

2. The calibration-synthesised p90 escaped the array-ceiling clamp its published
   sibling keeps. best_day_scaling has no floor at 1.0 (1.3 by default with
   calibration off, up to 2.0 with it on), so every Open-Meteo and Forecast.solar
   user's planner p90 could exceed what the panels can physically produce - and
   disagreed with the clamped pv_estimate90 for the same slot. It is now scaled
   per slot by min(best_day_scaling, capped_data / capped_p50), mirroring the
   published series exactly.

3. pv_metric90_weight was undocumented, leaving load_scaling90's text pointing at
   a setting no doc mentioned. Documented in customisation.md alongside its
   siblings, and the two apps-yaml.md passages updated.

4. Three of the four weight-0 skip gates were untested - the charge_min_max,
   export and levels gates could all be deleted with the suite green, imposing
   ~50% extra simulation cost on every user at the default weight. The weight-0
   test now patches all four launch functions; each gate was sabotaged in turn
   and confirmed to fail.

5. The random kernel parity sweep had swapped scenario coverage rather than
   adding it (~75/~75 nominal/pv10 became 57/58/35). It now loops all three
   scenarios per seed - 150/150/150 - leaving every previously generated
   configuration unchanged, for 0.62s -> 1.25s.

Weight-0 plan identity re-confirmed: both debug cases produce byte-identical
output to before these changes.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: springfall2008

apps/predbat/annual.py

@@ -1231,6 +1231,12 @@ def _run_scenarios(predbat, config, weather, tariff, load_source, day, midnight_
     predbat.soc_kw = START_SOC_KWH
     predbat.pv_forecast_minute = forecast_pv
     predbat.pv_forecast_minute10 = p10_pv
+    # Open-Meteo gives this model a forecast and a monthly P10 ratio but no P90 series, so the
+    # pv90 (upside) scenario runs on the same PV as nominal - its upside comes from load_scaling90
+    # alone. This must be assigned explicitly on every sampled day: one PredBat instance is reused
+    # for the whole year (see AnnualRun), so leaving it to calculate_plan()'s fallback would pin
+    # every later day's "upside" to the first sampled day's solar profile.
+    predbat.pv_forecast_minute90 = dict(forecast_pv)
     predbat.calculate_plan(recompute=True, debug_mode=False, publish=False)
 
     # Swap in the actuals before costing. There is no forecast/actual split for load (only PV

apps/predbat/fetch.py

@@ -731,6 +731,9 @@ def fetch_sensor_data(self, save=True):
         self.pv_forecast_minute = {}
         self.pv_forecast_minute10 = {}
         self.pv_forecast_minute90 = {}
+        # See Plan.refresh_pv_forecast_minute90(): both series are re-fetched together below, so no
+        # earlier pair of signatures may be held against them
+        self.pv_forecast_minute90_signatures = None
         self.load_scaling_dynamic = {}
         self.carbon_intensity = {}
         self.carbon_history = {}

apps/predbat/plan.py

@@ -1053,6 +1053,68 @@ def plan_scoring_pair(self, plan_new, plan_prev, preclip_new, preclip_prev):
             return preclip_new, preclip_prev
         return plan_new, plan_prev
 
+    @staticmethod
+    def pv_series_signature(series):
+        """Return a cheap content and coverage signature for a per-minute PV series.
+
+        The tuple is (minute count, total kWh, first minute, last minute) - enough to notice that a
+        series has been swapped or rewritten between plan runs, and to tell how much of the plan
+        horizon it spans. It is never used as a value in its own right, so a content collision costs
+        nothing more than a redundant (or skipped) refresh of the fallback p90 copy.
+        """
+        if not series:
+            return (0, 0.0, None, None)
+        return (len(series), round(sum(series.values()), 6), min(series), max(series))
+
+    def refresh_pv_forecast_minute90(self):
+        """Keep the pv90 (upside) PV forecast series in step with the p50 series it sits beside.
+
+        ``fetch.py`` always refreshes ``pv_forecast_minute90`` alongside ``pv_forecast_minute``, falling
+        back to a copy of the p50 when no forecast90 data is published, so in production the pair is
+        always consistent. But callers that assign ``pv_forecast_minute`` directly - ``annual.py``'s
+        year-long sweep, which reuses ONE PredBat instance across every sampled day, replayed debug
+        dumps with no forecast90 sensor, and unit tests sharing a fixture - can leave the p90 empty or,
+        far worse, holding a series belonging to a completely different p50.
+
+        A stale p90 is not a harmless approximation: it silently turns pv90, which exists to be the
+        UPSIDE case, into a severe downside one. January's p50 held against July's makes the "upside"
+        scenario carry a quarter of nominal PV - exactly the inversion already ruled out for
+        ``load_scaling90``, arriving by another route. Emptiness is therefore not a sufficient trigger.
+
+        Two independent tests must both pass for the p90 in hand to be used:
+
+        1. Coverage. The p90 must span the part of the plan horizon that the p50 spans. Missing minutes
+           read back as zero from ``step_data_history``, so a p90 that stops short of the horizon makes
+           pv90 a zero-PV downside case over the rest of it. In production this always holds - fetch.py
+           builds all three series over one shared minute range - so this only ever fires on a p90 that
+           came from somewhere else.
+        2. Not left behind. If the p50 changed since the previous call while the p90 did not, the p90
+           cannot belong to the p50 in hand. A p90 that moved with its p50 - the normal case for a real
+           fetched forecast90 - is kept untouched however far it diverges, because that divergence is
+           the entire point of having a real p90.
+
+        Failing either test re-derives the p90 from the p50. That costs only the accuracy of the upside
+        case (pv90 collapses to nominal PV, an inert scenario), whereas using a mismatched one silently
+        inverts what the feature means.
+        """
+        p50_signature = self.pv_series_signature(self.pv_forecast_minute)
+        p90_signature = self.pv_series_signature(self.pv_forecast_minute90)
+        previous = self.pv_forecast_minute90_signatures
+
+        if not self.pv_forecast_minute:
+            # Nothing to plan against - the only consistent p90 is an equally empty one
+            covers_horizon = not self.pv_forecast_minute90
+        else:
+            first_needed = max(p50_signature[2], self.minutes_now)
+            last_needed = min(p50_signature[3], self.minutes_now + self.forecast_minutes)
+            covers_horizon = bool(self.pv_forecast_minute90) and p90_signature[2] <= first_needed and p90_signature[3] >= last_needed
+        left_behind = previous is not None and previous[0] != p50_signature and previous[1] == p90_signature
+
+        if not covers_horizon or left_behind:
+            self.pv_forecast_minute90 = dict(self.pv_forecast_minute)
+            p90_signature = p50_signature
+        self.pv_forecast_minute90_signatures = (p50_signature, p90_signature)
+
     def calculate_plan(self, recompute=True, debug_mode=False, publish=True):
         """
         Calculate the new plan (best)
@@ -1190,9 +1252,7 @@ def calculate_plan(self, recompute=True, debug_mode=False, publish=True):
         )
         pv_forecast_minute_step = self.step_data_history(self.pv_forecast_minute, self.minutes_now, forward=True, cloud_factor=self.metric_cloud_coverage)
         pv_forecast_minute10_step = self.step_data_history(self.pv_forecast_minute10, self.minutes_now, forward=True, cloud_factor=min(self.metric_cloud_coverage + 0.2, 1.0) if self.metric_cloud_coverage else None, flip=True)
-        # Guard against a missing p90 series (older debug dumps replayed without a forecast90 fetch)
-        if not self.pv_forecast_minute90:
-            self.pv_forecast_minute90 = dict(self.pv_forecast_minute)
+        self.refresh_pv_forecast_minute90()
         pv_forecast_minute90_step = self.step_data_history(self.pv_forecast_minute90, self.minutes_now, forward=True, cloud_factor=self.metric_cloud_coverage)
 
         # Save step data for debug

apps/predbat/predbat.py

@@ -551,6 +551,9 @@ def reset(self):
         self.pv_forecast_minute = {}
         self.pv_forecast_minute10 = {}
         self.pv_forecast_minute90 = {}
+        # (p50, p90) content signatures from the previous plan run, used to spot a p90 that has been
+        # left behind by a p50 reassigned underneath it - see Plan.refresh_pv_forecast_minute90()
+        self.pv_forecast_minute90_signatures = None
         self.load_scaling_dynamic = {}
         self.carbon_intensity = {}
         self.carbon_history = {}

apps/predbat/solcast.py

@@ -1122,6 +1122,9 @@ def pv_calibration(self, pv_forecast_minute, pv_forecast_minute10, pv_forecast_m
         capped_slots = 0
         raw_exceeds_ceiling_slots = 0
         raw_exceeds_ceiling_peak = 0
+        # Per-slot best_day_scaling for the planner's p90 series, held back so the create_pv10 block
+        # below can apply the same ceiling this loop applies to the published pv_estimate90.
+        slot_best_scaling = {}
         for minute in range(0, max(pv_forecast_minute.keys()) + 1, self.plan_interval_minutes):
             pv_value = 0
             raw_value = 0
@@ -1143,6 +1146,14 @@ def pv_calibration(self, pv_forecast_minute, pv_forecast_minute10, pv_forecast_m
             pv_estimate10[minute] = dp4(capped_p50 * worst_day_scaling)
             pv_estimate90[minute] = dp4(min(capped_p50 * best_day_scaling, capped_data))
 
+            # The planner's p90 series (built in the create_pv10 block below from the capped
+            # per-minute data) must land on the same ceiling as pv_estimate90 above, or the two
+            # disagree exactly where the comment above says they must agree. capped_data is kWh per
+            # plan interval, so the clamp cannot be applied per minute; record the scaling that
+            # holds this slot's p90 total at min(capped_p50 * best_day_scaling, capped_data) instead
+            # and let the block below scale every minute of the slot by it.
+            slot_best_scaling[minute] = min(best_day_scaling, capped_data / capped_p50) if capped_p50 > 0 else best_day_scaling
+
             # Apply the same cap to the per-minute data the planner consumes. Scale rather than
             # clamp per minute: capped_data is kWh per plan interval, not per minute.
             if pv_value > capped_data and pv_value > 0:
@@ -1212,13 +1223,27 @@ def pv_calibration(self, pv_forecast_minute, pv_forecast_minute10, pv_forecast_m
 
         # Creation of PV10 data using worst day scaling factor
         if create_pv10:
+            capped_best_slots = 0
             for minute in range(0, max(pv_forecast_minute_adjusted.keys()) + 1):
                 pv_value = pv_forecast_minute_adjusted.get(minute, 0)
-                # Use the worst day scaling factor to create pv_estimate10
+                # Use the worst day scaling factor to create pv_estimate10. No ceiling clamp is
+                # needed: worst_day_scaling is capped at 1.0 above, so this can only scale down.
                 pv_forecast_minute10[minute] = dp4(pv_value * worst_day_scaling)
-                # Use the best day scaling factor to create pv_estimate90
-                pv_forecast_minute90[minute] = dp4(pv_value * best_day_scaling)
-            self.log("SolarAPI: PV Calibration: Created pv_estimate10/pv_estimate90 data using worst day scaling factor {} and best day scaling factor {}".format(dp2(worst_day_scaling), dp2(best_day_scaling)))
+                # Use the best day scaling factor to create pv_estimate90, clamped to this slot's
+                # array ceiling. best_day_scaling has no floor at 1.0 and reaches 2.0 (1.3 when
+                # calibration is disabled), so without the clamp the planner's upside case would
+                # predict more solar than the panels can physically produce - and would disagree
+                # with the published pv_estimate90 for the very same slot, which is clamped.
+                slot_start = int(minute / self.plan_interval_minutes) * self.plan_interval_minutes
+                best_scaling_slot = slot_best_scaling.get(slot_start, best_day_scaling)
+                if best_scaling_slot < best_day_scaling:
+                    capped_best_slots += 1
+                pv_forecast_minute90[minute] = dp4(pv_value * best_scaling_slot)
+            self.log(
+                "SolarAPI: PV Calibration: Created pv_estimate10/pv_estimate90 data using worst day scaling factor {} and best day scaling factor {} ({} minutes held at the array ceiling)".format(
+                    dp2(worst_day_scaling), dp2(best_day_scaling), capped_best_slots
+                )
+            )
 
         # Do we use calibrated or raw data?
         if self.get_arg("metric_pv_calibration_enable", default=True):

apps/predbat/tests/test_kernel_parity.py

@@ -641,8 +641,15 @@ def run_edge_case_tests(my_predbat):
 
 
 def run_random_sweep_tests(my_predbat, count=150):
-    """Seeded random configuration sweep comparing both engines, returns True on failure"""
+    """Seeded random configuration sweep comparing both engines, returns True on failure.
+
+    Every seed is run against all three PV scenarios. Drawing one scenario per seed instead would
+    trade away coverage of nominal and pv10 - the two scenarios every user runs at the default
+    pv_metric90_weight of 0 - to buy coverage of pv90; running all three is strictly additive and
+    the sweep is fast enough to absorb the 3x.
+    """
     failed = False
+    scenario_counts = {PV_SCENARIO_NOMINAL: 0, PV_SCENARIO_PV10: 0, PV_SCENARIO_PV90: 0}
     for seed in range(count):
         rng = random.Random(seed)
         reset_inverter(my_predbat)
@@ -662,12 +669,21 @@ def run_random_sweep_tests(my_predbat, count=150):
         export_window = make_windows(rng, my_predbat.minutes_now, my_predbat.forecast_minutes, rng.randint(0, 3), align=rng.choice([5, 5, 30]))
         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, rng.randrange(0, my_predbat.forecast_minutes, 5)])
-        pv_scenario = rng.choice([PV_SCENARIO_NOMINAL, PV_SCENARIO_PV10, PV_SCENARIO_PV90])
 
-        failed |= dual_run("random_{}".format(seed), my_predbat, pv_step, pv10_step, load_step, load10_step, charge_limit, charge_window, export_window, export_limits, pv_scenario, end_record, pv90_step=pv90_step, load90_step=load90_step)
+        # No scenario is drawn from rng here: the draw that used to sit at this position was the last
+        # use of rng in the loop body, so looping the scenarios instead leaves every previously
+        # generated configuration (windows, limits, end_record, step data) bit-for-bit unchanged.
+        for pv_scenario in (PV_SCENARIO_NOMINAL, PV_SCENARIO_PV10, PV_SCENARIO_PV90):
+            scenario_counts[pv_scenario] += 1
+            failed |= dual_run(
+                "random_{}_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, pv90_step=pv90_step, load90_step=load90_step
+            )
+            if failed:
+                print("Random sweep failed at seed {} scenario {}".format(seed, pv_scenario))
+                break
         if failed:
-            print("Random sweep failed at seed {}".format(seed))
             break
+    print("Random sweep ran {} configurations: nominal {}, pv10 {}, pv90 {}".format(sum(scenario_counts.values()), scenario_counts[PV_SCENARIO_NOMINAL], scenario_counts[PV_SCENARIO_PV10], scenario_counts[PV_SCENARIO_PV90]))
     return failed