fix(teslemetry): keep scheduled-export slots out of real-tier quantisation

Follow-up to #4346. _quantise_side splits [min,max] into 3 equal-width bands;
a saving-session / scheduled-export spike in rate_export blew up the range so
the whole normal day (overnight, midday, evening peak) collapsed into a single
bottom tier. The Powerwall then saw a flat price outside the export window and
lost its normal charge/export signal.

The spike is already priced via the reserved ON_PEAK boost band, so it should
not also define the 3 real bands. build_tariff now computes the boost segments
before quantising and passes those ranges to _quantise_side, which excludes
them from the band definition (they are still assigned a real tier, clamped
into range, but that is overwritten by the boost). With no discharge window
the behaviour is byte-identical to before.

Adds a regression test: an Agile day plus a 175p saving session keeps overnight
< midday < evening-peak as distinct increasing sell tiers, with the session in
ON_PEAK above them all.

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

Author: springfall2008

apps/predbat/teslemetry.py

@@ -831,36 +831,57 @@ def current_rates(self):
         return import_rate, export_rate
 
     @staticmethod
-    def _quantise_side(rate_dict, default_pence):
+    def _quantise_side(rate_dict, default_pence, exclude_ranges=()):
         """Quantise a per-minute pence rate dict into <=3 named GBP tiers over today+tomorrow.
 
         Samples every 30 minutes for today (minutes 0-1439) and tomorrow (1440-2879), converts to
         GBP clamped at 0 and rounded to whole pence, then either maps <=3 distinct values one-to-one
         onto the real tiers (cheapest -> SUPER_OFF_PEAK) or splits [min,max] into 3 equal-width bands
         priced at each band's mean. Returns (tier_prices, today_tiers, tomorrow_tiers) where the slot
         lists only ever name tiers present in tier_prices (matched sets).
+
+        exclude_ranges is a list of (start_min, end_min) absolute-minute ranges (today 0-1439,
+        tomorrow 1440-2879) whose slots are a scheduled export - they are priced via the reserved
+        ON_PEAK boost band, not the 3 real bands, so they are kept OUT of the band definition here.
+        Otherwise a saving-session/export spike would blow up the [min,max] range and collapse the
+        whole normal day into a single tier. Excluded slots are still assigned a real tier (their
+        value clamped into the excluded-slot-free range) but that tier is later overwritten by the
+        boost, so the choice is immaterial. With no exclude_ranges this is byte-identical to before.
         """
 
         def slot_price(minute):
             """Convert a per-minute pence rate to GBP, clamped at 0 and rounded to whole pence."""
             pence = rate_dict.get(minute, default_pence)
             return round(max(0.0, pence) / 100.0, 2)
 
-        today = [slot_price(m) for m in range(0, 1440, SLOT_MINUTES)]
-        tomorrow = [slot_price(m) for m in range(1440, 2880, SLOT_MINUTES)]
-        combined = today + tomorrow
-        distinct = sorted(set(combined))
+        def is_excluded(minute):
+            """True if a sample minute falls in a scheduled-export range (priced via ON_PEAK instead)."""
+            return any(start <= minute < end for start, end in exclude_ranges)
+
+        sample_minutes = list(range(0, 2880, SLOT_MINUTES))
+        combined = [slot_price(minute) for minute in sample_minutes]
+        today = combined[:SLOTS_PER_DAY]
+        tomorrow = combined[SLOTS_PER_DAY:]
+        # Values that DEFINE the bands exclude scheduled-export slots so a spike cannot dominate.
+        band_values = [price for minute, price in zip(sample_minutes, combined) if not is_excluded(minute)]
+        if not band_values:
+            band_values = list(combined)
+        low, high = min(band_values), max(band_values)
+        distinct = sorted(set(band_values))
+
+        def clamp(value):
+            """Clamp a slot value into the (excluded-slot-free) band range before tier assignment."""
+            return min(max(value, low), high)
 
         if len(distinct) <= len(REAL_TIERS):
             value_to_tier = {value: REAL_TIERS[index] for index, value in enumerate(distinct)}
             tier_prices = {tier: value for value, tier in value_to_tier.items()}
 
             def band_of(value):
-                """Return the tier for an exact value in the small-distinct case."""
-                return value_to_tier[value]
+                """Return the tier for a value, clamped so excluded-slot outliers map to the top real tier."""
+                return value_to_tier[clamp(value)]
 
         else:
-            low, high = distinct[0], distinct[-1]
             width = (high - low) / len(REAL_TIERS)
             buckets = {index: [] for index in range(len(REAL_TIERS))}
 
@@ -870,13 +891,13 @@ def band_index(value):
                     return 0
                 return min(len(REAL_TIERS) - 1, int((value - low) / width))
 
-            for value in combined:
-                buckets[band_index(value)].append(value)
+            for price in band_values:
+                buckets[band_index(price)].append(price)
             tier_prices = {REAL_TIERS[index]: round(sum(values) / len(values), 2) for index, values in buckets.items() if values}
 
             def band_of(value):
-                """Return the tier name for the band a value falls in."""
-                return REAL_TIERS[band_index(value)]
+                """Return the tier name for the band a value falls in (clamped into the band range)."""
+                return REAL_TIERS[band_index(clamp(value))]
 
         today_tiers = [band_of(value) for value in today]
         tomorrow_tiers = [band_of(value) for value in tomorrow]
@@ -1012,17 +1033,18 @@ def _apply_boost(buy_layout, sell_layout, segments, today_dow):
                 day = (today_dow + offset) % 7
                 layout[day] = TeslemetryAPI._carve_interval(layout[day], seg_start, seg_end, BOOST_TIER)
 
-    def _rate_side(self, rate_dict, default_gbp):
+    def _rate_side(self, rate_dict, default_gbp, exclude_ranges=()):
         """Return the 4-tuple (energy_charges_side, tou_periods, tier_prices, layout) for one side.
 
         layout is None in the flat/fallback branch (no rates), signalling build_tariff to skip the boost;
-        otherwise it is the per-DOW interval layout the boost carves into.
+        otherwise it is the per-DOW interval layout the boost carves into. exclude_ranges names the
+        scheduled-export slots to keep out of the real-tier band definition (see _quantise_side).
         """
         if not rate_dict:
             flat = round(max(0.0, default_gbp), 2)
             return {"ALL": {"rates": {"ALL": flat}}}, {}, {"SUPER_OFF_PEAK": flat}, None
         today_dow = self._tesla_dow(self._local_today_weekday())
-        tier_prices, today_tiers, tomorrow_tiers = self._quantise_side(rate_dict, default_gbp * 100.0)
+        tier_prices, today_tiers, tomorrow_tiers = self._quantise_side(rate_dict, default_gbp * 100.0, exclude_ranges)
         layout = self._side_layout(today_tiers, tomorrow_tiers, today_dow)
         return (*self._render_side(layout, tier_prices), tier_prices, layout)
 
@@ -1062,15 +1084,19 @@ def build_tariff(self, discharge_window=None, now_min=None):
             now_min = self.get_minutes_now()
         import_gbp, export_gbp = self.current_rates()
         today_dow = self._tesla_dow(self._local_today_weekday())
-        buy_charges, buy_periods, buy_prices, buy_layout = self._rate_side(self._side_rates("import"), import_gbp)
-        sell_charges, sell_periods, sell_prices, sell_layout = self._rate_side(self._side_rates("export"), export_gbp)
+        # Compute the boost segments BEFORE quantising so the scheduled-export slots can be kept out of
+        # the real-tier band definition - they are priced via the reserved ON_PEAK band, and leaving a
+        # saving-session/export spike in would blow up the [min,max] range and flatten the normal day.
+        segments = self._boost_segments(discharge_window, now_min) if discharge_window is not None else []
+        exclude_ranges = [(offset * 1440 + seg_start, offset * 1440 + seg_end) for offset, seg_start, seg_end in segments]
+        buy_charges, buy_periods, buy_prices, buy_layout = self._rate_side(self._side_rates("import"), import_gbp, exclude_ranges)
+        sell_charges, sell_periods, sell_prices, sell_layout = self._rate_side(self._side_rates("export"), export_gbp, exclude_ranges)
         code = "PREDBAT"
         # buy_layout/sell_layout are None only in the flat-fallback branch (no rate data, priced via the
         # ALL field) - there are no per-day bands to carve a boost into, and that degenerate zero-rate-data
         # case is not a normal production state, so the boost is intentionally skipped there.
-        if discharge_window is not None and buy_layout is not None and sell_layout is not None:
+        if segments and buy_layout is not None and sell_layout is not None:
             boost = self._boost_price(buy_prices, sell_prices)
-            segments = self._boost_segments(discharge_window, now_min)
             self._apply_boost(buy_layout, sell_layout, segments, today_dow)
             buy_charges, buy_periods = self._render_side(buy_layout, {**buy_prices, BOOST_TIER: boost})
             sell_charges, sell_periods = self._render_side(sell_layout, {**sell_prices, BOOST_TIER: boost})

apps/predbat/tests/test_teslemetry.py

@@ -630,6 +630,61 @@ def test_teslemetry_build_tariff_consolidates_replicated_days():
             assert len(block["periods"]) <= 3, "tier {} has {} period entries, expected consolidated ranges (<=3): {}".format(tier, len(block["periods"]), block["periods"])
 
 
+def test_teslemetry_saving_session_spike_keeps_daily_shape():
+    """A scheduled-export/saving-session spike (issue #4346 follow-up) must not flatten the rest of the
+    day's rate shape when quantised into Tesla tiers.
+
+    The spike lives in the reserved ON_PEAK boost band, so it is excluded from the 3 real-tier band
+    definition. Before the fix, the outlier blew up the [min,max] range and collapsed overnight AND
+    the evening peak into one bottom tier - the Powerwall then saw a flat price outside the session
+    and lost its normal charge/export signal. Here overnight, midday and the evening peak must each
+    resolve to a distinct, monotonically increasing sell tier, while the boosted session sits in
+    ON_PEAK above them all.
+    """
+    from types import SimpleNamespace
+    from datetime import datetime
+
+    def agile(offset):
+        """An Agile-like export day: overnight 5p, midday 12p, evening peak 25p, half-hourly."""
+        day = {}
+        for slot in range(48):
+            hour = slot // 2
+            price = 5.0 if hour < 6 else (25.0 if 16 <= hour < 19 else 12.0)
+            for minute in range(slot * 30, slot * 30 + 30):
+                day[offset + minute] = price
+        return day
+
+    rate_export = {**agile(0), **agile(1440)}
+    for minute in range(17 * 60, 18 * 60 + 30):  # 17:00-18:30 saving session already folded into rate_export
+        rate_export[minute] = 175.0
+
+    api = MockTeslemetryAPI()
+    api.base = SimpleNamespace(rate_import={m: 15.0 for m in range(2880)}, rate_export=rate_export, minutes_now=0, now=datetime(2026, 7, 20, 12, 0), local_tz=None, get_arg=lambda a, d=None, **k: d)
+    # Predbat scheduled an export over the session, so build_tariff is called with that window.
+    tariff = api.build_tariff((17 * 60, 18 * 60 + 30), now_min=12 * 60)
+    sell = tariff["sell_tariff"]["energy_charges"]["AllYear"]["rates"]
+    periods = tariff["sell_tariff"]["seasons"]["AllYear"]["tou_periods"]
+    today = api._tesla_dow(api._local_today_weekday())
+
+    def sell_price_at(minute):
+        """Return the sell tier price applying on today's day at the given minute-of-day."""
+        for tier, block in periods.items():
+            for period in block["periods"]:
+                if period["fromDayOfWeek"] <= today <= period["toDayOfWeek"]:
+                    start = period["fromHour"] * 60 + period["fromMinute"]
+                    end = (period["toHour"] * 60 + period["toMinute"]) or 1440
+                    if start <= minute < end:
+                        return sell[tier]
+        return None
+
+    overnight = sell_price_at(3 * 60)  # real 5p
+    midday = sell_price_at(12 * 60)  # real 12p
+    eve_peak = sell_price_at(16 * 60)  # real 25p, outside the boosted session
+    session = sell_price_at(17 * 60 + 30)  # boosted session
+    assert overnight < midday < eve_peak, "daily shape lost: overnight={} midday={} peak={}".format(overnight, midday, eve_peak)
+    assert session > eve_peak, "boosted session {} must price above the real peak {}".format(session, eve_peak)
+
+
 def test_teslemetry_day_runs_groups_replicated_days():
     """_day_runs() itself: 6 identical days plus 1 different day collapse to 2 runs, not 7 singletons."""
     api = MockTeslemetryAPI()
@@ -1831,6 +1886,7 @@ def test_teslemetry(my_predbat=None):
     test_teslemetry_build_tariff_boost_clamps_above_high_rates()
     test_teslemetry_build_tariff_periods_partition_each_day()
     test_teslemetry_build_tariff_consolidates_replicated_days()
+    test_teslemetry_saving_session_spike_keeps_daily_shape()
     test_teslemetry_day_runs_groups_replicated_days()
     test_teslemetry_day_runs_all_identical_single_run()
     test_teslemetry_set_tariff_posts_tou_settings()