diff --git a/CLAUDE.md b/CLAUDE.md index 270da4f..35a169a 100644 --- a/CLAUDE.md +++ b/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) | diff --git a/pyproject.toml b/pyproject.toml index a6143a4..e96c793 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "skypro" -version = "2.0.5" +version = "2.1.1" description = "Skyprospector by Cepro" authors = ["damonrand "] license = "MIT" diff --git a/src/skypro/commands/simulator/algorithms/price_curve/algo.py b/src/skypro/commands/simulator/algorithms/price_curve/algo.py index 7ac821a..89f817f 100644 --- a/src/skypro/commands/simulator/algorithms/price_curve/algo.py +++ b/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 ) diff --git a/src/skypro/commands/simulator/algorithms/price_curve/peak.py b/src/skypro/commands/simulator/algorithms/price_curve/peak.py index 21f0474..1f40451 100644 --- a/src/skypro/commands/simulator/algorithms/price_curve/peak.py +++ b/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()) diff --git a/src/skypro/commands/simulator/config/config.py b/src/skypro/commands/simulator/config/config.py index f047c90..213725f 100644 --- a/src/skypro/commands/simulator/config/config.py +++ b/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 diff --git a/src/tests/integration/fixtures/simulation/config.yaml b/src/tests/integration/fixtures/simulation/config.yaml index c5d1f6b..946d96a 100644 --- a/src/tests/integration/fixtures/simulation/config.yaml +++ b/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 diff --git a/src/tests/integration/test_integration_simulator.py b/src/tests/integration/test_integration_simulator.py index befddcc..b0c0e63 100644 --- a/src/tests/integration/test_integration_simulator.py +++ b/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", diff --git a/src/tests/unit/skypro/commands/simulator/test_peak.py b/src/tests/unit/skypro/commands/simulator/test_peak.py new file mode 100644 index 0000000..9e99009 --- /dev/null +++ b/src/tests/unit/skypro/commands/simulator/test_peak.py @@ -0,0 +1,216 @@ +import unittest +from datetime import datetime, time, timedelta + +import pytz + +from skypro.commands.simulator.algorithms.price_curve.peak import ( + _find_active_peak, + _find_next_peak, + get_peak_power, +) +from skypro.commands.simulator.algorithms.price_curve.system_state import SystemState +from skypro.commands.simulator.config.config import Approach, Peak, PeakDynamic +from skypro.common.timeutils.clock_time_period import ClockTimePeriod +from skypro.common.timeutils.dayed_period import DayedPeriod +from skypro.common.timeutils.days import Days + +TZ = pytz.timezone("Europe/London") + + +def _make_peak( + start_h: int, + end_h: int, + days_name: str = "weekdays", + dynamic: "PeakDynamic | None" = None, +) -> Peak: + return Peak( + period=DayedPeriod( + days=Days(name=days_name, tz_str="Europe/London"), + period=ClockTimePeriod(start=time(start_h), end=time(end_h), tz_str="Europe/London"), + ), + approach=Approach( + to_soe=100, + encourage_to_soe=None, + assumed_charge_power=200, + encourage_charge_duration_factor=0, + force_charge_duration_factor=1, + charge_cushion=timedelta(minutes=0), + ), + dynamic=dynamic, + ) + + +class TestFindActivePeak(unittest.TestCase): + + def test_empty_list(self): + t = TZ.localize(datetime(2026, 1, 5, 12, 0)) # Mon + self.assertIsNone(_find_active_peak([], t)) + + def test_single_peak_inside(self): + peak = _make_peak(17, 19) + t = TZ.localize(datetime(2026, 1, 5, 17, 30)) # Mon 17:30 + self.assertIs(_find_active_peak([peak], t), peak) + + def test_single_peak_outside(self): + peak = _make_peak(17, 19) + t = TZ.localize(datetime(2026, 1, 5, 8, 0)) # Mon 08:00, before peak + self.assertIsNone(_find_active_peak([peak], t)) + + def test_two_peaks_in_first(self): + morning = _make_peak(7, 9) + evening = _make_peak(17, 19) + t = TZ.localize(datetime(2026, 1, 5, 8, 0)) + self.assertIs(_find_active_peak([morning, evening], t), morning) + + def test_two_peaks_in_second(self): + morning = _make_peak(7, 9) + evening = _make_peak(17, 19) + t = TZ.localize(datetime(2026, 1, 5, 17, 30)) + self.assertIs(_find_active_peak([morning, evening], t), evening) + + def test_two_peaks_between(self): + morning = _make_peak(7, 9) + evening = _make_peak(17, 19) + t = TZ.localize(datetime(2026, 1, 5, 12, 0)) + self.assertIsNone(_find_active_peak([morning, evening], t)) + + def test_overlapping_peaks_raises(self): + a = _make_peak(7, 12) + b = _make_peak(10, 14) + t = TZ.localize(datetime(2026, 1, 5, 11, 0)) + with self.assertRaises(ValueError): + _find_active_peak([a, b], t) + + def test_day_class_filter_weekend(self): + # Peak only on weekdays — Saturday should not match + peak = _make_peak(17, 19, days_name="weekdays") + t = TZ.localize(datetime(2026, 1, 3, 17, 30)) # Sat 17:30 + self.assertIsNone(_find_active_peak([peak], t)) + + +class TestFindNextPeak(unittest.TestCase): + + def test_empty_list(self): + t = TZ.localize(datetime(2026, 1, 5, 6, 0)) + self.assertIsNone(_find_next_peak([], t)) + + def test_before_single_peak(self): + peak = _make_peak(17, 19) + t = TZ.localize(datetime(2026, 1, 5, 8, 0)) + self.assertIs(_find_next_peak([peak], t), peak) + + def test_after_single_peak_today(self): + # No peak left today + peak = _make_peak(17, 19) + t = TZ.localize(datetime(2026, 1, 5, 20, 0)) + self.assertIsNone(_find_next_peak([peak], t)) + + def test_picks_earliest_upcoming(self): + morning = _make_peak(7, 9) + evening = _make_peak(17, 19) + # At 06:00 both are upcoming — should pick the morning one (earliest start) + t = TZ.localize(datetime(2026, 1, 5, 6, 0)) + self.assertIs(_find_next_peak([morning, evening], t), morning) + + def test_after_morning_returns_evening(self): + morning = _make_peak(7, 9) + evening = _make_peak(17, 19) + # After morning peak ends, evening is the next-upcoming + t = TZ.localize(datetime(2026, 1, 5, 10, 0)) + self.assertIs(_find_next_peak([morning, evening], t), evening) + + def test_during_morning_returns_evening(self): + # While morning peak is active, the next-upcoming is the evening peak + # (the active peak has start <= t, so it's excluded by the `> t` filter). + morning = _make_peak(7, 9) + evening = _make_peak(17, 19) + t = TZ.localize(datetime(2026, 1, 5, 8, 0)) + self.assertIs(_find_next_peak([morning, evening], t), evening) + + def test_weekend_with_weekday_only_peaks(self): + morning = _make_peak(7, 9, days_name="weekdays") + evening = _make_peak(17, 19, days_name="weekdays") + t = TZ.localize(datetime(2026, 1, 3, 6, 0)) # Saturday + self.assertIsNone(_find_next_peak([morning, evening], t)) + + +class TestGetPeakPowerMinEndOfPeakSoe(unittest.TestCase): + """ + The min_end_of_peak_soe knob reserves SoE for post-peak niv-chase. The key + behavioural change: time_to_empty is computed against `soe - min_end_of_peak_soe` + instead of `soe`, which creates slack in the peak window so the dynamic + HOLD-on-LONG branch can actually fire. + """ + + def _common(self): + return dict( + time_step=timedelta(minutes=30), + bess_max_power_discharge=400.0, # kW; at 400 kW empties 800 kWh in 2h + microgrid_residual_power=0.0, + ) + + def test_default_zero_means_legacy_behavior(self): + # Legacy: with soe=800, max_discharge=400, peak window 2h: time_to_empty=2h, + # latest_time_before_max = peak_end - 2h = peak_start. So at any t inside peak, + # t >= peak_start = (latest - time_step) → forced full discharge. + peak = _make_peak( + 17, 19, dynamic=PeakDynamic(prioritise_residual_load=False, min_end_of_peak_soe=0.0) + ) + t = TZ.localize(datetime(2026, 4, 7, 17, 30)) # Tue 17:30 — mid-peak + power = get_peak_power( + peaks=[peak], t=t, soe=800.0, system_state=SystemState.LONG, **self._common() + ) + # min_end=0 + LONG → no slack → forced full discharge despite LONG + self.assertEqual(power, -400.0) + + def test_nonzero_min_end_creates_slack_and_holds_on_long(self): + # Reserve 400 kWh post-peak: dischargeable_soe = 800-400 = 400. + # time_to_empty = 400/400 = 1h → latest_time = 19:00-1h = 18:00. + # At t=17:30 we are early enough → flexibility branch → LONG → HOLD (0.0). + peak = _make_peak( + 17, 19, dynamic=PeakDynamic(prioritise_residual_load=False, min_end_of_peak_soe=400.0) + ) + t = TZ.localize(datetime(2026, 4, 7, 17, 30)) + power = get_peak_power( + peaks=[peak], t=t, soe=800.0, system_state=SystemState.LONG, **self._common() + ) + self.assertEqual(power, 0.0) + + def test_nonzero_min_end_short_system_full_discharge(self): + # Same setup as above (reserve 400) but system is SHORT → full discharge. + peak = _make_peak( + 17, 19, dynamic=PeakDynamic(prioritise_residual_load=False, min_end_of_peak_soe=400.0) + ) + t = TZ.localize(datetime(2026, 4, 7, 17, 30)) + power = get_peak_power( + peaks=[peak], t=t, soe=800.0, system_state=SystemState.SHORT, **self._common() + ) + self.assertEqual(power, -400.0) + + def test_late_in_peak_forces_full_discharge_even_with_reserve(self): + # At t=18:30, latest_time = 18:00, t > 18:00-30min = 17:30 → force full. + peak = _make_peak( + 17, 19, dynamic=PeakDynamic(prioritise_residual_load=False, min_end_of_peak_soe=400.0) + ) + t = TZ.localize(datetime(2026, 4, 7, 18, 30)) + power = get_peak_power( + peaks=[peak], t=t, soe=800.0, system_state=SystemState.LONG, **self._common() + ) + self.assertEqual(power, -400.0) + + def test_min_end_above_soe_clamps_to_zero_dischargeable(self): + # If min_end > soe, dischargeable=0, time_to_empty=0, latest_time=peak_end. + # So t > peak_end - time_step (~18:30) → must full discharge from 18:30 onwards; + # at 17:30 (well before latest), we're flexible → LONG → HOLD. + peak = _make_peak( + 17, 19, dynamic=PeakDynamic(prioritise_residual_load=False, min_end_of_peak_soe=1000.0) + ) + t = TZ.localize(datetime(2026, 4, 7, 17, 30)) + power = get_peak_power( + peaks=[peak], t=t, soe=800.0, system_state=SystemState.LONG, **self._common() + ) + self.assertEqual(power, 0.0) + + +if __name__ == "__main__": + unittest.main()