fix(pv90): clamp the synthesised p90 against the array ceiling, not against capped_data

The previous commit's clamp used capped_data, which is
min(ceiling_slot, max(observed_slot, raw_value)) - not the array's physical
ceiling. It collapses to the raw forecast itself whenever the forecast sits
below the ceiling, and capped_p50 then equals capped_data, so the ratio came
out at exactly 1.0 and erased best_day_scaling altogether.

That is worse than the bug it was meant to fix. The create_pv10 path is reached
only by Forecast.solar and Open-Meteo - exactly the users with no real Solcast
p90, and the ones the docs now tell that calibration provides the 90% data.
With calibration not enabled (fewer than 3 valid history days, so a fresh
install or a system that was down) every adjustment is forced to 1.0 and the
p90 would have equalled the p50 exactly, every slot, every day: the scenario
still simulated, still costing planning time, measuring nothing on the axis it
exists for, silently.

Both the planner series and the published pv_estimate90 now clamp against
ceiling_slot, the physical ceiling. The two expressions are the same clamp
algebraically, so they still agree slot for slot as the comment above them
requires. The division is guarded against a zero capped_p50.

test_pv_calibration_capped_data_clamp gains a second scenario where the ceiling
does NOT bind (max_kwh 20.0 against the same 3 kW forecast) and asserts the p90
carries its full 1.3x upside while staying under the array limit. Scenario A
alone could not pin this: with the ceiling binding, a p90 collapsed onto the p50
is indistinguishable from a correctly clamped one. Verified in three states -
no clamp fails the over-ceiling assertion, the capped_data clamp fails the new
non-binding assertion, and the ceiling_slot clamp passes both.

Weight-0 plan identity gate re-confirmed byte-identical.

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

Author: springfall2008

apps/predbat/solcast.py

@@ -1137,22 +1137,33 @@ def pv_calibration(self, pv_forecast_minute, pv_forecast_minute10, pv_forecast_m
             capped_data = min(ceiling_slot, max(observed_slot, raw_value))
             # Derive all three published series from the capped P50 so they agree with the
             # planner series built below from the (also capped) pv_forecast_minute_adjusted.
-            # pv_estimate10 needs no min(..., capped_data): worst_day_scaling is clamped to at
-            # most 1.0 above, so capped_p50 * worst_day_scaling <= capped_p50 <= capped_data
-            # always holds. pv_estimate90 keeps the clamp because best_day_scaling can exceed
-            # 1.0, so the optimistic case can genuinely exceed the physical ceiling.
+            # pv_estimate10 needs no clamp at all: worst_day_scaling is clamped to at most 1.0
+            # above, so capped_p50 * worst_day_scaling <= capped_p50 <= capped_data always holds.
+            # pv_estimate90 keeps a clamp because best_day_scaling can exceed 1.0 (up to 2.0, and
+            # 1.3 by default when calibration is off), so the optimistic case can genuinely exceed
+            # what the array can physically produce.
+            #
+            # That clamp is against ceiling_slot, the physical ceiling, NOT against capped_data.
+            # capped_data is min(ceiling_slot, max(observed_slot, raw_value)), so it collapses to the
+            # raw forecast itself whenever the raw forecast sits below the ceiling - and capped_p50
+            # then equals capped_data, making the clamp bite at exactly 1.0 and erasing the upside
+            # entirely. On a system whose forecast is comfortably inside its array limit (the normal
+            # case) that would leave pv90 identical to nominal, silently: the scenario would still be
+            # simulated and still cost planning time while measuring nothing on the axis it exists
+            # for. Only the array limit may cap the upside case.
             capped_p50 = min(pv_value, capped_data)
             pv_estimateCL[minute] = dp4(capped_p50)
             pv_estimate10[minute] = dp4(capped_p50 * worst_day_scaling)
-            pv_estimate90[minute] = dp4(min(capped_p50 * best_day_scaling, capped_data))
+            pv_estimate90[minute] = dp4(min(capped_p50 * best_day_scaling, ceiling_slot))
 
             # 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
+            # disagree exactly where the comment above says they must agree. ceiling_slot 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
+            # holds this slot's p90 total at min(capped_p50 * best_day_scaling, ceiling_slot) instead
+            # and let the block below scale every minute of the slot by it. An empty slot has no
+            # ratio to take, and needs no clamp either - scaling zero by anything stays zero.
+            slot_best_scaling[minute] = min(best_day_scaling, ceiling_slot / 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.

apps/predbat/tests/test_solcast.py

@@ -3314,19 +3314,27 @@ def mock_get_history(entity_id, days, required=False, _h0=h0_ha_history):
 
 def test_pv_calibration_capped_data_clamp(my_predbat):
     """
-    Test the per-slot cap in pv_calibration.
+    Test the per-slot cap in pv_calibration, and the array-ceiling clamp on the synthesised p90.
 
-    Setup: observed peak is 1 kW, the raw forecast is 3 kW, and max_kwh (declared array
-    capacity) is 2 kW. The h0 forecast history is empty so calibration is disabled and the
-    adjusted values equal the raw forecast.
+    Setup: observed peak is 1 kW and the raw forecast is 3 kW. The h0 forecast history is empty so
+    calibration is disabled, the adjusted values equal the raw forecast, and best_day_scaling takes
+    its fixed 1.3 default.
 
-    cap = min(ceiling, max(observed_slot, raw_slot))
-        = min(max(1.2 * 2.0, 1.0), max(1.0, 3.0))
-        = min(2.4, 3.0)
-        = 2.4 kW  ->  2.4 * plan_interval / 60 per slot
+    Scenario A - max_kwh (declared array capacity) is 2 kW, so the ceiling BINDS:
+
+        cap = min(ceiling, max(observed_slot, raw_slot))
+            = min(max(1.2 * 2.0, 1.0), max(1.0, 3.0))
+            = min(2.4, 3.0)
+            = 2.4 kW  ->  2.4 * plan_interval / 60 per slot
 
     The 1.2 * max_kwh ceiling is what binds here. Note the cap is ABOVE the observed peak
-    of 1 kW: a dull week must not suppress a sunny day's forecast.
+    of 1 kW: a dull week must not suppress a sunny day's forecast. The p90 must be held AT that
+    ceiling rather than 1.3x above it.
+
+    Scenario B - max_kwh is 20 kW, so the ceiling does NOT bind and the p90 must carry its full
+    1.3x upside. This is the scenario that distinguishes a clamp against the physical ceiling from
+    a clamp against capped_data, which silently erases the upside on any system whose forecast sits
+    inside its array limit - i.e. very nearly all of them.
     """
     print("  - test_pv_calibration_capped_data_clamp")
     failed = False
@@ -3360,27 +3368,45 @@ def mock_minute_data_import_export(max_days_previous, now_utc, key, scale=1.0, r
         base.minute_data_import_export = mock_minute_data_import_export
         solar.get_history_wrapper = lambda entity_id, days, required=False: []
 
-        # Forecast: 3 kW constant (above historical and above max_kwh)
-        total_minutes = 4 * 24 * 60
-        pv_forecast_minute = {m: 3.0 / 60 for m in range(total_minutes)}  # kWh per minute
-        pv_forecast_minute10 = {m: 2.0 / 60 for m in range(total_minutes)}
-
         # Build forecast data entries — one per plan_interval over 1 day
         from datetime import timedelta
         import pytz
 
         midnight = base.midnight_utc.replace(tzinfo=pytz.utc)
-        pv_forecast_data = []
-        for slot in range(0, 24 * 60, plan_interval):
-            ts = midnight + timedelta(minutes=slot)
-            pv_forecast_data.append({"period_start": ts.strftime("%Y-%m-%dT%H:%M:%S+0000"), "pv_estimate": 3.0 * plan_interval / 60})
 
+        def _calibrate(max_kwh):
+            """Run pv_calibration against a 3 kW constant forecast for the given declared array capacity.
+
+            Every input is rebuilt per call: pv_calibration writes into the p10/p90 dicts it is handed
+            and annotates the forecast data entries in place, so reusing them across scenarios would
+            leak the first scenario's results into the second.
+            """
+            total_minutes = 4 * 24 * 60
+            pv_forecast_minute = {m: 3.0 / 60 for m in range(total_minutes)}  # kWh per minute
+            pv_forecast_minute10 = {m: 2.0 / 60 for m in range(total_minutes)}
+            pv_forecast_data = []
+            for slot in range(0, 24 * 60, plan_interval):
+                ts = midnight + timedelta(minutes=slot)
+                pv_forecast_data.append({"period_start": ts.strftime("%Y-%m-%dT%H:%M:%S+0000"), "pv_estimate": 3.0 * plan_interval / 60})
+            # create_pv10=True so the synthesised p10/p90 planner series are built too - they are the series
+            # every Open-Meteo and Forecast.solar user's planner actually consumes, and the p90 is scaled by
+            # best_day_scaling (1.3 here with calibration disabled, up to 2.0 with it on), so it is the one
+            # series that can be pushed above the array limit after the p50 has been capped.
+            return solar.pv_calibration(pv_forecast_minute, pv_forecast_minute10, {}, pv_forecast_data, create_pv10=True, divide_by=1.0, max_kwh=max_kwh, forecast_days=solar.forecast_days)
+
+        def _max_slot_total(series):
+            """Return the largest per-plan-interval total in a per-minute kWh series."""
+            if not series:
+                return 0.0
+            slot_totals = {}
+            for minute, value in series.items():
+                slot_start = int(minute / plan_interval) * plan_interval
+                slot_totals[slot_start] = slot_totals.get(slot_start, 0.0) + value
+            return max(slot_totals.values())
+
+        # ---- Scenario A: the array ceiling BINDS (max_kwh 2.0 against a 3 kW forecast) ----
         max_kwh = 2.0  # panel peak output cap in kW
-        # create_pv10=True so the synthesised p10/p90 planner series are built too - they are the series
-        # every Open-Meteo and Forecast.solar user's planner actually consumes, and the p90 is scaled by
-        # best_day_scaling (1.3 here with calibration disabled, up to 2.0 with it on), so it is the one
-        # series that can be pushed back above the ceiling after the p50 has been capped.
-        adj_minute, adj_minute10, adj_minute90, adj_data = solar.pv_calibration(pv_forecast_minute, pv_forecast_minute10, {}, pv_forecast_data, create_pv10=True, divide_by=1.0, max_kwh=max_kwh, forecast_days=solar.forecast_days)
+        adj_minute, adj_minute10, adj_minute90, adj_data = _calibrate(max_kwh)
 
         expected_cap = 1.2 * max_kwh / 60 * plan_interval  # the 1.2 * max_kwh ceiling binds here
 
@@ -3400,21 +3426,10 @@ def mock_minute_data_import_export(max_days_previous, now_utc, key, scale=1.0, r
             print("ERROR: pv_estimateCL {} is below the expected cap {} - the raw forecast floor is not being applied".format(got_max, expected_cap))
             failed = True
 
-        # The synthesised p90 planner series must respect the SAME ceiling the p50 was capped to.
-        # best_day_scaling has no floor at 1.0 (it defaults to 1.3 with calibration disabled and can
-        # reach 2.0 with it on), so an unclamped p90 lands at 1.3x the array's physical ceiling and
-        # the planner prices a spill that cannot happen - and it disagrees with the published
+        # The synthesised p90 planner series must respect the array ceiling. best_day_scaling has no
+        # floor at 1.0, so an unclamped p90 lands at 1.3x the array's physical ceiling here and the
+        # planner prices a spill that cannot happen - and it disagrees with the published
         # pv_estimate90 for the very same slot, which is clamped.
-        def _max_slot_total(series):
-            """Return the largest per-plan-interval total in a per-minute kWh series."""
-            if not series:
-                return 0.0
-            slot_totals = {}
-            for minute, value in series.items():
-                slot_start = int(minute / plan_interval) * plan_interval
-                slot_totals[slot_start] = slot_totals.get(slot_start, 0.0) + value
-            return max(slot_totals.values())
-
         max_slot_p50 = _max_slot_total(adj_minute)
         max_slot_p90 = _max_slot_total(adj_minute90)
         if max_slot_p90 > expected_cap * 1.01:
@@ -3424,6 +3439,30 @@ def _max_slot_total(series):
             print("ERROR: pv_forecast_minute90 peaks at {} kWh/slot, below the p50 peak of {} kWh/slot - the clamp must hold the upside case AT the ceiling, not below it".format(round(max_slot_p90, 4), round(max_slot_p50, 4)))
             failed = True
 
+        # ---- Scenario B: the array ceiling does NOT bind (max_kwh 20.0 against the same 3 kW forecast) ----
+        # Scenario A alone cannot pin the clamp: there every slot sits AT the ceiling, so a p90 that has
+        # been collapsed onto the p50 is indistinguishable from a correctly clamped one. Here the ceiling
+        # is 8x above the forecast and must not bite at all - the p90 has to carry the full
+        # best_day_scaling upside. Clamping against capped_data instead of the physical ceiling fails
+        # this: capped_data collapses to the raw forecast whenever the forecast is below the ceiling, so
+        # the ratio comes out at exactly 1.0 and the upside is erased on every normally-sized system.
+        max_kwh_open = 20.0
+        _, _, adj_minute90_open, _ = _calibrate(max_kwh_open)
+        ceiling_open = 1.2 * max_kwh_open / 60 * plan_interval
+        best_day_scaling = 1.3  # calibration is disabled (no h0 history), so this is the fixed default
+        max_slot_p90_open = _max_slot_total(adj_minute90_open)
+        expected_p90_open = 3.0 / 60 * plan_interval * best_day_scaling
+        if max_slot_p90_open < expected_p90_open * 0.99:
+            print(
+                "ERROR: with the array ceiling {} kWh/slot well above the forecast, pv_forecast_minute90 peaks at only {} kWh/slot - expected {} ({}x the {} kWh/slot forecast). The upside has been clamped away by something other than the array limit".format(
+                    round(ceiling_open, 4), round(max_slot_p90_open, 4), round(expected_p90_open, 4), best_day_scaling, round(3.0 / 60 * plan_interval, 4)
+                )
+            )
+            failed = True
+        if max_slot_p90_open > ceiling_open * 1.01:
+            print("ERROR: pv_forecast_minute90 peaks at {} kWh/slot, above the array ceiling of {} kWh/slot".format(round(max_slot_p90_open, 4), round(ceiling_open, 4)))
+            failed = True
+
     finally:
         test_api.cleanup()