feat: multi-peak priceCurveAlgo + Peak.dynamic.minEndOfPeakSoe (v2.1.1)

CLAUDE.md

@@ -138,6 +138,8 @@ pandas, plotly, pulp, pendulum, sqlalchemy, psycopg2-binary, marshmallow, pyyaml
 
 | Date | PR | Branch | Summary |
 |------|-----|--------|---------|
+| 2026-05-08 | TBD | feature/support-axle-flex | New `Peak.dynamic.minEndOfPeakSoe` parameter reserves SoE for post-peak niv-chase. Used in time-to-empty calc as `dischargeable_soe = soe − min_end_of_peak_soe`, creating slack so the dynamic HOLD-on-LONG branch can actually fire instead of always falling through to forced full discharge. Default 0 (legacy behaviour). +5 unit tests (v2.1.1) |
+| 2026-05-08 | TBD | feature/support-axle-flex | Multi-peak support in priceCurveAlgo: new `peaks: [...]` list form alongside legacy `peak: ...` (mutually exclusive). Enables dispatch into multiple price-elevated windows per day. Backward-compat verified — existing single-peak fixture summary unchanged within tolerance (v2.1.0) |
 | 2026-04-30 | TBD | feature/profile-filter-and-nameplate | Opt-in profile anomaly filter (`maxEnergyPerIntervalKwh`) + display-only `nameplateKwp` metadata field on Profile (v2.0.4) |
 | 2026-01-12 | - | main | License changed from AGPL-3.0 to MIT |
 | 2026-01-09 | #59 | bugfix/nan-threshold-aggregation | Fix NaN propagation in rate averages and cost totals (v2.0.1) |

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "skypro"
-version = "2.0.5"
+version = "2.1.1"
 description = "Skyprospector by Cepro"
 authors = ["damonrand <damon@cepro.energy>"]
 license = "MIT"

src/skypro/commands/simulator/algorithms/price_curve/algo.py

@@ -127,8 +127,9 @@ def _get_power_at_time(self, t: datetime, time_step: timedelta) -> Tuple[float,
         niv_config = get_relevant_niv_config(self._algo_config.niv_chase_periods, t).niv
         system_state = get_system_state(self._df, t, niv_config.volume_cutoff_for_prediction)
         # If we are in a pre-defined 'peak period' then we probably just want to discharge fully to benefit from the DUoS red band:
+        peaks = self._algo_config.resolve_peaks()
         peak_power = get_peak_power(
-            peak_config=self._algo_config.peak,
+            peaks=peaks,
             t=t,
             time_step=time_step,
             soe=self._df.loc[t, "soe"],
@@ -147,7 +148,7 @@ def _get_power_at_time(self, t: datetime, time_step: timedelta) -> Tuple[float,
                 time_step=time_step,
                 soe=self._df.loc[t, "soe"],
                 charge_efficiency=self._bess_config.charge_efficiency,
-                peak_config=self._algo_config.peak,
+                peaks=peaks,
                 is_long=system_state == SystemState.LONG
             )
 

src/skypro/commands/simulator/algorithms/price_curve/peak.py

@@ -1,5 +1,5 @@
 from datetime import datetime, timedelta
-from typing import Tuple, Optional
+from typing import List, Tuple, Optional
 
 import numpy as np
 import pandas as pd
@@ -13,8 +13,43 @@
 REF_DATETIME = TIMEZONE.localize(datetime(year=2000, month=1, day=1))
 
 
+def _find_active_peak(peaks: List[Peak], t: datetime) -> Optional[Peak]:
+    """Return the peak whose period contains `t`, or None.
+
+    Raises ValueError if more than one peak's period claims `t` (overlapping peaks are
+    a configuration error — discharge logic for a single timestep can only be driven by one peak).
+    """
+    matches = [p for p in peaks if p.period.contains(t)]
+    if len(matches) > 1:
+        raise ValueError(
+            f"Multiple peaks active at {t}: {[str(p.period) for p in matches]}. "
+            "Peak periods must not overlap."
+        )
+    return matches[0] if matches else None
+
+
+def _find_next_peak(peaks: List[Peak], t: datetime) -> Optional[Peak]:
+    """Return the next peak that starts strictly after `t` on the same calendar day (in `t`'s tz)
+    and applies to that day-class. None if no peak qualifies.
+
+    The approach machinery (force/encourage charging up to a target SoE before the peak) only makes
+    sense for an upcoming peak; once the peak starts the dispatch logic in `get_peak_power` takes over.
+    """
+    upcoming = []
+    for p in peaks:
+        if not p.period.days.is_on_day(t):
+            continue
+        start = p.period.period.start_absolute(t.date())
+        if start > t:
+            upcoming.append((start, p))
+    if not upcoming:
+        return None
+    upcoming.sort(key=lambda pair: pair[0])
+    return upcoming[0][1]
+
+
 def get_peak_power(
-        peak_config: Optional[Peak],
+        peaks: List[Peak],
         t: datetime,
         time_step: timedelta | pd.Timedelta,
         soe: float,
@@ -23,16 +58,14 @@ def get_peak_power(
         system_state: SystemState
 ) -> Optional[float]:
     """
-    Returns the power to deliver during the peak period, or None if the peak is not active/relevant at this time.
+    Returns the power to deliver during a peak period, or None if no peak is active at this time.
     """
     # There is a strange bug with pd.TimeDelta where it doesn't behave correctly when differencing - convert to
     # a `timedelta` type
     time_step = timedelta(seconds=time_step.total_seconds())
 
-    if not peak_config:
-        return None
-
-    if not peak_config.period.contains(t):
+    peak_config = _find_active_peak(peaks, t)
+    if peak_config is None:
         return None
 
     if not peak_config.dynamic:
@@ -43,12 +76,18 @@ def get_peak_power(
     # then we can get an improvement on the above 'dumb' method by choosing when we discharge into the peak.
     peak_end = peak_config.period.period.end_absolute(t)
 
+    # `min_end_of_peak_soe` reserves SoE for post-peak niv-chase. It's subtracted
+    # from `soe` for the time-to-empty calc, which is what creates slack — without
+    # it, time_to_empty often consumes the full peak window so HOLD-on-LONG never
+    # gets a chance to fire.
+    dischargeable_soe = max(0.0, soe - peak_config.dynamic.min_end_of_peak_soe)
+
     # This is an approximation because we may be limited by grid constraints which depend on the load and solar levels
     # which, in turn, may change throughout the peak.
     if bess_max_power_discharge <= 0:
         assumed_time_to_empty_battery = timedelta(minutes=0)
     else:
-        assumed_time_to_empty_battery = timedelta(hours=(soe / bess_max_power_discharge))
+        assumed_time_to_empty_battery = timedelta(hours=(dischargeable_soe / bess_max_power_discharge))
 
     # We want to ensure that we empty the battery completely by the end of the peak period, and there is a
     # point into the peak where we must discharge at the max power to ensure that.
@@ -90,7 +129,7 @@ def get_peak_power(
         reserve_duration = peak_end - empty_time_without_reserve
         reserve_energy = microgrid_residual_power * (reserve_duration.total_seconds() / 3600)
 
-        if reserve_energy > soe:
+        if reserve_energy > dischargeable_soe:
             # The assumptions around how much we needed to reserve were wrong, and so we are going to run out of energy.
             # Just do our best to service the residual load at this point:
             return -microgrid_residual_power
@@ -101,7 +140,7 @@ def get_peak_power(
         if duration_before_reserve.total_seconds() <= 0:
             # TODO: this should return microgrid_residual_power?!
             return -bess_max_power_discharge
-        energy_before_reserve = soe - reserve_energy
+        energy_before_reserve = dischargeable_soe - reserve_energy
         return -energy_before_reserve / (duration_before_reserve.total_seconds() / 3600)
 
 
@@ -110,7 +149,7 @@ def get_peak_approach_energies(
         time_step: timedelta,
         soe: float,
         charge_efficiency: float,
-        peak_config: Peak,
+        peaks: List[Peak],
         is_long: bool,
 ) -> Tuple[float, float]:
     """
@@ -122,24 +161,32 @@ def get_peak_approach_energies(
     The 'force' peak approach will get the battery to a target SoE by charging, even if the system is short,
     AND the current SoE is below a threshold which is defined by the timings in the configuration.
 
+    With multiple peaks configured, the next-upcoming peak today is the one whose approach drives charging.
+    Once that peak starts, dispatch in `get_peak_power` takes over; immediately after it ends, the following
+    peak (if any later today) becomes the next-upcoming and its approach kicks in.
+
     FOr a more detailed description of this mechanism, see the docstring on the `Approach` configuration class in simulator/config/config.py
 
     :param t: the time now
     :param time_step: the size of the simulation time step
     :param soe: the current battery soe
     :param charge_efficiency:
-    :param peak_config:
+    :param peaks: list of peak configurations (single-peak schemas pass a one-element list)
     :param is_long: indicates if the system is long or short - the encourage curve is only used when the system is long
     :return:
     """
     # TODO: this approach won't work if the approach curve crosses over a midnight boundary
 
-    if not peak_config or not peak_config.period or (peak_config.approach.to_soe == 0 and peak_config.approach.encourage_to_soe == 0):
+    if not peaks:
         return 0.0, 0.0
 
     t = t.astimezone(TIMEZONE)
 
-    if not peak_config.period.days.is_on_day(t):
+    peak_config = _find_next_peak(peaks, t)
+    if peak_config is None:
+        return 0.0, 0.0
+
+    if peak_config.approach.to_soe == 0 and peak_config.approach.encourage_to_soe == 0:
         return 0.0, 0.0
 
     peak_start = peak_config.period.period.start_absolute(t.date())

src/skypro/commands/simulator/config/config.py

@@ -186,6 +186,14 @@ class Approach:
 class PeakDynamic:
     prioritise_residual_load: bool = field_with_opts(key="prioritiseResidualLoad")
 
+    # Reserves SoE that the peak's dispatch logic will NOT discharge. Post-peak
+    # niv-chase handles the residual based on imbalance signals. Setting this >0
+    # creates "slack" in the time-to-empty calculation so the dynamic HOLD-on-LONG
+    # branch can actually fire instead of always falling through to forced full
+    # discharge. See peak.py — uses `dischargeable_soe = soe - min_end_of_peak_soe`
+    # in the time-to-empty calc, opening up flexibility within the peak window.
+    min_end_of_peak_soe: float = field_with_opts(key="minEndOfPeakSoe", default=0.0)
+
 
 @dataclass
 class Peak:
@@ -233,10 +241,27 @@ class Microgrid:
 class PriceCurveAlgo:
     """
     Configures the price curve algorithm.
+
+    A scenario may declare a single `peak` (legacy) or multiple `peaks` (e.g. morning + evening
+    Axle/imbalance trading windows), but not both. Use `resolve_peaks()` to get a list view that
+    works for either form, including the no-peak case.
     """
     microgrid: Optional[Microgrid] = field_with_opts(key="microgrid")
     peak: Optional[Peak] = field_with_opts(key="peak")
     niv_chase_periods: List[NivPeriod] = field_with_opts(key="nivChasePeriods")
+    peaks: Optional[List[Peak]] = field_with_opts(key="peaks", default=None)
+
+    def __post_init__(self):
+        if self.peak is not None and self.peaks is not None:
+            raise ValueError("Specify either 'peak' (single) or 'peaks' (list), not both.")
+
+    def resolve_peaks(self) -> List[Peak]:
+        """Return the configured peaks as a list, regardless of whether 'peak' or 'peaks' was set."""
+        if self.peaks is not None:
+            return self.peaks
+        if self.peak is not None:
+            return [self.peak]
+        return []
 
 
 @dataclass

src/tests/integration/fixtures/simulation/config.yaml

@@ -142,3 +142,79 @@ simulations:
       live: *rates
       final: *rates
     output: *output
+
+  # Multi-peak schema: two trading windows per weekday (morning + evening). Exercises
+  # the `peaks: [...]` list form. The single-peak scenario above continues to use the
+  # legacy `peak: ...` form so backward-compat is checked at every test run.
+  integrationTestPriceCurveMultiPeak:
+    timeFrame: *timeframe
+    site: *site
+    strategy:
+      priceCurveAlgo:
+        peaks:
+          - period:
+              days: "weekdays:Europe/London"
+              start: "07:00:00:Europe/London"
+              end: "08:30:00:Europe/London"
+            approach:
+              toSoe: 600
+              assumedChargePower: 200
+              forceChargeDurationFactor: 1
+              encourageChargeDurationFactor: 0
+              chargeCushionMins: 20
+          - period:
+              days: "weekdays:Europe/London"
+              start: "17:00:00:Europe/London"
+              end: "19:00:00:Europe/London"
+            approach:
+              toSoe: 1050
+              assumedChargePower: 200
+              forceChargeDurationFactor: 1
+              encourageChargeDurationFactor: 0
+              chargeCushionMins: 20
+        nivChasePeriods:
+        - period:
+            days: "weekdays:Europe/London"
+            start: "00:00:00:Europe/London"
+            end: "23:59:59:Europe/London"
+          niv:
+            chargeCurve: [
+              {x: -Infinity, y: 1280},
+              {x: 4, y: 1280},
+              {x: 6, y: 1000},
+              {x: 12, y: 0}
+            ]
+            dischargeCurve: [
+              {x: 8, y: 1280},
+              {x: 18, y: 1000},
+              {x: 25, y: 1000},
+              {x: 35, y: 0},
+              {x: Infinity, y: 0}
+            ]
+            curveShiftLong: 6
+            curveShiftShort: 2
+            volumeCutoffForPrediction: 150000
+        - period:
+            days: "weekends:Europe/London"
+            start: "00:00:00:Europe/London"
+            end: "23:59:59:Europe/London"
+          niv:
+            chargeCurve: [
+              {x: -Infinity, y: 1280},
+              {x: -2, y: 1280},
+              {x: 6, y: 400},
+              {x: 12, y: 0}
+            ]
+            dischargeCurve: [
+              {x: 8, y: 1280},
+              {x: 14, y: 400},
+              {x: 35, y: 100},
+              {x: Infinity, y: 0}
+            ]
+            curveShiftLong: 3
+            curveShiftShort: 3
+            volumeCutoffForPrediction: 150000
+    rates:
+      live: *rates
+      final: *rates
+    output: *output

src/tests/integration/test_integration_simulator.py

@@ -66,6 +66,19 @@ class SubTest:
                     "mfCost:all.supplierFee": [300 * num_days_simulated],
                 })
             ),
+            SubTest(
+                msg="integrationTestPriceCurveMultiPeak",
+                sim_name="integrationTestPriceCurveMultiPeak",
+                expected_summary_df=pd.DataFrame.from_dict({
+                    "c:solarToGrid": [499.97],
+                    "c:gridToLoad": [27809.65],
+                    "c:solarToLoad": [5022.52],
+                    "c:battToLoad": [4895.05],
+                    "c:battToGrid": [33884.23],
+                    "c:solarToBatt": [681.58],
+                    "c:gridToBatt": [44453.60],
+                })
+            ),
             SubTest(
                 msg="integrationTestPerfectHindsightLP",
                 sim_name="integrationTestPerfectHindsightLP",