From f481e22d801129c3b1fdcb9a6bf6ce4e5013a1dc Mon Sep 17 00:00:00 2001 From: andig Date: Fri, 24 Jul 2026 13:00:49 +0200 Subject: [PATCH 1/2] fix: level the grid profile below the peak, not only the absolute peak The attenuate_* charging strategies penalized only p_{side}_peak, the horizon maximum of the grid power. Once that maximum is pinned by a time step the batteries cannot shave - because a battery is already full, or because the surplus is far beyond c_max - the objective is completely flat over every profile below the cap, so the solver returns an arbitrary vertex. The result was a capped but jagged feed-in curve, e.g. 172, 0, 429, 525, 0, 0, 0, 943, 0. Penalize the step to step ramp of the grid power as well: p_{side}_ramp[t] >= |p_grid[t] - p_grid[t-1]| at prc_p_ramp = penalty_base * 1e-5, a hundred times below the peak weight, so lowering the peak still wins wherever the two disagree. The peak constraint is rewritten in power terms (p_grid[t] <= p_{side}_peak) so both share the same expression. On the reported case the schedule economics are unchanged (6.686112), the peak drops from 1853 W to 1806 W and the total variation of the feed-in curve from 20666 W to 8899 W. Solving also gets ~20x faster (12.7s -> 0.5s) because the ramp term removes the degeneracy the solver was branching through. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/optimizer/optimizer.py | 38 ++++++++++++++++++----- tests/test_peak_leveling.py | 61 +++++++++++++++++++++++++++++++++++++ 2 files changed, 91 insertions(+), 8 deletions(-) create mode 100644 tests/test_peak_leveling.py diff --git a/src/optimizer/optimizer.py b/src/optimizer/optimizer.py index c3cf96140..5e561f813 100644 --- a/src/optimizer/optimizer.py +++ b/src/optimizer/optimizer.py @@ -111,6 +111,12 @@ def __init__(self, strategy: OptimizationStrategy, grid: GridConfig, batteries: # solar power self.prc_e_grid_exp_pen = self.penalty_base * 10e1 + # weights of the grid peak leveling strategies. capping the horizon maximum alone leaves the + # profile below the cap arbitrary, so the step to step ramp is penalized as well. The ramp + # weight is the smaller one, so lowering the peak wins wherever the two disagree. + self.prc_p_peak = self.penalty_base * 1e-3 + self.prc_p_ramp = self.penalty_base * 1e-5 + # if there is a demand rate given in the input, the grid import limit will be interpreted as the # threshold beyond wich the demand rate is to be applied. Compute a demand rate flag for use in the # build constraint and build objective methods. @@ -210,9 +216,14 @@ def _setup_variables(self): if self.is_grid_demand_rate_active: self.variables['p_max_imp_exc'] = pulp.LpVariable("p_max_imp_exc", lowBound=0) - # highest grid power over the whole horizon (W) per side, used by the peak attenuation strategies + # highest grid power over the whole horizon (W) and step to step ramp of the grid power (W) + # per side, used by the peak attenuation strategies for side in self.peak_sides: self.variables[f'p_{side}_peak'] = pulp.LpVariable(f"p_{side}_peak", lowBound=0) + self.variables[f'p_{side}_ramp'] = [ + pulp.LpVariable(f"p_{side}_ramp_{t}", lowBound=0) + for t in self.time_steps + ] # Binary variable: power flow direction to / from grid variables # these variables @@ -329,12 +340,14 @@ def _setup_target_function(self): # level the grid profile to unload the public grid from peaks. attenuate_demand_peaks levels # grid import, attenuate_feedin_peaks levels grid export, attenuate_grid_peaks levels both. - # the penalty sits on the horizon maximum instead of on charge power, so the optimizer spreads - # charging at partial power over several time steps rather than running one step at full power. + # the penalty sits on the horizon maximum and on the step to step ramp instead of on charge + # power, so the optimizer spreads charging at partial power over several time steps rather + # than running one step at full power, and keeps the profile below the cap leveled too. # penalty_base is used instead of min_import_price because negative market prices would turn # this penalty into a reward for peaks. for side in self.peak_sides: - objective += - self.variables[f'p_{side}_peak'] * self.penalty_base * 1e-3 + objective += - self.variables[f'p_{side}_peak'] * self.prc_p_peak + objective += - pulp.lpSum(self.variables[f'p_{side}_ramp']) * self.prc_p_ramp # prefer discharging batteries completely before importing from grid if self.strategy.discharging_strategy == 'discharge_before_import': @@ -427,16 +440,25 @@ def _add_energy_balance_constraints(self): self.problem += (self.variables['e_exp_lim_exc'][t] <= self.M * (1 - self.variables['z_exp_lim'][t])) - # track the horizon maximum of the total grid power for every side the strategy levels. - # the peak includes the portion beyond p_max_imp / p_max_exp, so it stays correct in - # demand rate mode and when a limit is violated. + # track the horizon maximum and the step to step ramp of the total grid power for every side + # the strategy levels. Both include the portion beyond p_max_imp / p_max_exp, so they stay + # correct in demand rate mode and when a limit is violated. for side in self.peak_sides: grid_var, lim_exc_var = PEAK_SIDE_VARIABLES[side] + # total grid power of this side per time step (W) + p_grid = [] for t in self.time_steps: e_grid = self.variables[grid_var][t] if lim_exc_var in self.variables: e_grid = e_grid + self.variables[lim_exc_var][t] - self.problem += e_grid <= self.variables[f'p_{side}_peak'] * self.time_series.dt[t] / 3600 + p_grid.append(e_grid * 3600 / self.time_series.dt[t]) + + for t in self.time_steps: + self.problem += p_grid[t] <= self.variables[f'p_{side}_peak'] + # ramp magnitude: p_ramp[t] >= |p_grid[t] - p_grid[t-1]| + for t in range(1, self.T): + self.problem += self.variables[f'p_{side}_ramp'][t] >= p_grid[t] - p_grid[t - 1] + self.problem += self.variables[f'p_{side}_ramp'][t] >= p_grid[t - 1] - p_grid[t] # if demand rate is applied, the maximum grid import power value # of all time steps drives the demand rate charge diff --git a/tests/test_peak_leveling.py b/tests/test_peak_leveling.py new file mode 100644 index 000000000..883faea45 --- /dev/null +++ b/tests/test_peak_leveling.py @@ -0,0 +1,61 @@ +import pathlib +import sys + +import pytest + +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1] / 'src')) + +from optimizer.optimizer import ( # noqa: E402 + BatteryConfig, + GridConfig, + OptimizationStrategy, + Optimizer, + TimeSeriesData, +) + + +def _optimize(charging_strategy, ft, gt, c_max, s_max): + """single battery, 1 h steps, flat prices, no grid limits""" + T = len(ft) + ts = TimeSeriesData( + dt=[3600] * T, ft=ft, gt=gt, + p_N=[0.30e-3] * T, p_E=[0.08e-3] * T, + ) + bat = BatteryConfig( + charge_from_grid=False, discharge_to_grid=False, + s_capacity=s_max, s_min=0, s_max=s_max, s_initial=0, + c_min=0, c_max=c_max, d_max=0, p_a=0.25e-3, + ) + opt = Optimizer( + strategy=OptimizationStrategy(charging_strategy, 'none'), + grid=GridConfig(p_max_imp=None, p_max_exp=None, prc_p_exc_imp=None), + batteries=[bat], time_series=ts, eta_c=1.0, eta_d=1.0, + ) + res = opt.solve() + assert res['status'] == 'Optimal' + return res + + +@pytest.mark.parametrize('strategy', ['attenuate_feedin_peaks', 'attenuate_grid_peaks']) +def test_feedin_profile_is_leveled_below_the_peak(strategy): + """ + The surplus in the first step is far beyond the charging power, so the export peak is pinned + there and charging the remaining 170 Wh anywhere in steps 1..4 leaves that peak untouched. + Capping the peak alone would therefore accept an arbitrary split; the profile has to come out + leveled instead. + """ + res = _optimize(strategy, ft=[1000, 100, 100, 100, 100], gt=[0] * 5, c_max=50, s_max=220) + + # peak is pinned by the first step: 1000 Wh surplus less the 50 Wh the battery can take + assert res['grid_export'][0] == pytest.approx(950, abs=1e-3) + # remaining 170 Wh charged in equal shares -> equal export in every following step + assert res['grid_export'][1:] == pytest.approx([100 - 170 / 4] * 4, abs=1e-3) + + +def test_demand_strategy_leaves_feedin_alone(): + """attenuate_demand_peaks levels grid import only, so the export side keeps the free choice""" + res = _optimize('attenuate_demand_peaks', ft=[1000, 100, 100, 100, 100], gt=[0] * 5, + c_max=50, s_max=220) + + # the battery still fills up, but the export profile is not required to be leveled + assert sum(res['batteries'][0]['charging_power']) == pytest.approx(220, abs=1e-3) From 9c141d38dbd30b90d9fadd1554796eaf50378154 Mon Sep 17 00:00:00 2001 From: andig Date: Fri, 24 Jul 2026 13:11:47 +0200 Subject: [PATCH 2/2] test: move the leveling test into the data driven harness The coverage moves to test_cases/026-attenuate-feedin-peaks-below-cap.json on https://github.com/evcc-io/optimizer/pull/101, where the peak shaping cases and the strict schedule comparison live. Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/test_peak_leveling.py | 61 ------------------------------------- 1 file changed, 61 deletions(-) delete mode 100644 tests/test_peak_leveling.py diff --git a/tests/test_peak_leveling.py b/tests/test_peak_leveling.py deleted file mode 100644 index 883faea45..000000000 --- a/tests/test_peak_leveling.py +++ /dev/null @@ -1,61 +0,0 @@ -import pathlib -import sys - -import pytest - -sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1] / 'src')) - -from optimizer.optimizer import ( # noqa: E402 - BatteryConfig, - GridConfig, - OptimizationStrategy, - Optimizer, - TimeSeriesData, -) - - -def _optimize(charging_strategy, ft, gt, c_max, s_max): - """single battery, 1 h steps, flat prices, no grid limits""" - T = len(ft) - ts = TimeSeriesData( - dt=[3600] * T, ft=ft, gt=gt, - p_N=[0.30e-3] * T, p_E=[0.08e-3] * T, - ) - bat = BatteryConfig( - charge_from_grid=False, discharge_to_grid=False, - s_capacity=s_max, s_min=0, s_max=s_max, s_initial=0, - c_min=0, c_max=c_max, d_max=0, p_a=0.25e-3, - ) - opt = Optimizer( - strategy=OptimizationStrategy(charging_strategy, 'none'), - grid=GridConfig(p_max_imp=None, p_max_exp=None, prc_p_exc_imp=None), - batteries=[bat], time_series=ts, eta_c=1.0, eta_d=1.0, - ) - res = opt.solve() - assert res['status'] == 'Optimal' - return res - - -@pytest.mark.parametrize('strategy', ['attenuate_feedin_peaks', 'attenuate_grid_peaks']) -def test_feedin_profile_is_leveled_below_the_peak(strategy): - """ - The surplus in the first step is far beyond the charging power, so the export peak is pinned - there and charging the remaining 170 Wh anywhere in steps 1..4 leaves that peak untouched. - Capping the peak alone would therefore accept an arbitrary split; the profile has to come out - leveled instead. - """ - res = _optimize(strategy, ft=[1000, 100, 100, 100, 100], gt=[0] * 5, c_max=50, s_max=220) - - # peak is pinned by the first step: 1000 Wh surplus less the 50 Wh the battery can take - assert res['grid_export'][0] == pytest.approx(950, abs=1e-3) - # remaining 170 Wh charged in equal shares -> equal export in every following step - assert res['grid_export'][1:] == pytest.approx([100 - 170 / 4] * 4, abs=1e-3) - - -def test_demand_strategy_leaves_feedin_alone(): - """attenuate_demand_peaks levels grid import only, so the export side keeps the free choice""" - res = _optimize('attenuate_demand_peaks', ft=[1000, 100, 100, 100, 100], gt=[0] * 5, - c_max=50, s_max=220) - - # the battery still fills up, but the export profile is not required to be leveled - assert sum(res['batteries'][0]['charging_power']) == pytest.approx(220, abs=1e-3)