feat: split grid peak shaping into feed-in, demand and combined options

openapi.yaml

@@ -153,12 +153,14 @@ components:
       properties:
         charging_strategy:
           type: string
-          enum: [none, charge_before_export, attenuate_grid_peaks]
+          enum: [none, charge_before_export, attenuate_demand_peaks, attenuate_feedin_peaks, attenuate_grid_peaks]
           description: |
             Sets a strategy for charging in situations where choices are cost neutral.
             - none (default): no strategy set 
             - charge_before_export: charge batteries before exporting to grid
-            - attenuate_grid_peaks: charge at times with high solar yield to reduce the grid load
+            - attenuate_demand_peaks: level the grid import profile, charging at partial power over several time steps instead of one peak
+            - attenuate_feedin_peaks: level the grid export profile, charging to shave solar feed-in peaks
+            - attenuate_grid_peaks: level both the grid import and the grid export profile
         discharging_strategy:
           type: string
           enum: [none, discharge_before_import]

src/optimizer/optimizer.py

@@ -14,6 +14,21 @@ class OptimizationStrategy:
     discharging_strategy: str
 
 
+# charging strategies that level grid peaks, mapped to the metered sides they level.
+# 'imp' is the demand side (grid import), 'exp' the feed-in side (grid export).
+PEAK_STRATEGY_SIDES = {
+    'attenuate_demand_peaks': ('imp',),
+    'attenuate_feedin_peaks': ('exp',),
+    'attenuate_grid_peaks': ('imp', 'exp'),
+}
+
+# grid energy variable and limit exceedance variable per leveled side
+PEAK_SIDE_VARIABLES = {
+    'imp': ('n', 'e_imp_lim_exc'),
+    'exp': ('e', 'e_exp_lim_exc'),
+}
+
+
 @dataclass
 class GridConfig:
     p_max_imp: float
@@ -82,19 +97,19 @@ def __init__(self, strategy: OptimizationStrategy, grid: GridConfig, batteries:
         self.max_import_price = np.max(self.time_series.p_N)
 
         # scaling base for penalty parameters. Make sure goal_penalty is always positive.
-        penalty_base = np.max([self.max_import_price, 0.1e-3])
+        self.penalty_base = np.max([self.max_import_price, 0.1e-3])
 
         # scaling for penalty parameters
-        self.prc_e_goal_pen = penalty_base * 10e1
-        self.prc_p_goal_pen = penalty_base * np.max(self.time_series.dt) / 3600 * 10e1
-        self.prc_soc_exc_pen = penalty_base * 10e2
+        self.prc_e_goal_pen = self.penalty_base * 10e1
+        self.prc_p_goal_pen = self.penalty_base * np.max(self.time_series.dt) / 3600 * 10e1
+        self.prc_soc_exc_pen = self.penalty_base * 10e2
 
         # penalty for exceeding grid import limit. Result shall not become infeasible but report the violation
         # with helpful information
-        self.prc_e_grid_imp_pen = penalty_base * 10e1
+        self.prc_e_grid_imp_pen = self.penalty_base * 10e1
         # penalty for exceeding the grid export limit. Result shall not become infeasible but report the 'lost'
         # solar power
-        self.prc_e_grid_exp_pen = penalty_base * 10e1
+        self.prc_e_grid_exp_pen = self.penalty_base * 10e1
 
         # 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
@@ -103,6 +118,9 @@ def __init__(self, strategy: OptimizationStrategy, grid: GridConfig, batteries:
         if self.grid.p_max_imp is not None and self.grid.prc_p_exc_imp is not None:
             self.is_grid_demand_rate_active = True
 
+        # grid sides leveled by the active peak attenuation strategy, empty for all other strategies
+        self.peak_sides = PEAK_STRATEGY_SIDES.get(strategy.charging_strategy, ())
+
     def create_model(self):
         """
         Create and initialize the MILP model
@@ -192,6 +210,10 @@ 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
+        for side in self.peak_sides:
+            self.variables[f'p_{side}_peak'] = pulp.LpVariable(f"p_{side}_peak", lowBound=0)
+
         # Binary variable: power flow direction to / from grid variables
         # these variables
         # 1. avoid direct export from import if export remuneration is greater than import cost
@@ -305,11 +327,14 @@ def _setup_target_function(self):
                 for t in self.time_steps:
                     objective += - self.variables['e'][t] * self.min_import_price * 2e-5 * (self.T - t)
 
-        # prefer charging at high solar production times to unload public grid from peaks
-        if self.strategy.charging_strategy == 'attenuate_grid_peaks':
-            for i, bat in enumerate(self.batteries):
-                for t in self.time_steps:
-                    objective += self.variables['c'][i][t] * self.time_series.ft[t] * self.min_import_price * 1e-6
+        # 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.
+        # 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
 
         # prefer discharging batteries completely before importing from grid
         if self.strategy.discharging_strategy == 'discharge_before_import':
@@ -402,6 +427,17 @@ 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.
+        for side in self.peak_sides:
+            grid_var, lim_exc_var = PEAK_SIDE_VARIABLES[side]
+            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
+
         # if demand rate is applied, the maximum grid import power value
         # of all time steps drives the demand rate charge
         if self.is_grid_demand_rate_active: