logic: support minimum grid charge soc

config/batcontrol_config_dummy.yaml

@@ -18,6 +18,7 @@ battery_control:
                                  #    _rel helps to avoid charging at high prices with less efficiency
   always_allow_discharge_limit: 0.90 # 0.00 to 1.00 above this SOC limit using energy from the battery is always allowed
   max_charging_from_grid_limit: 0.89 # 0.00 to 1.00 charging from the grid is only allowed until this SOC limit
+  # min_grid_charge_soc: 0.55        # optional 0.00 to 1.00 target to preserve/charge before expensive slots
   min_recharge_amount: 100           # in Wh, start & minimum amount of energy to recharge the battery
 
 #--------------------------
@@ -32,6 +33,7 @@ battery_control_expert:
   round_price_digits: 4 # round price to n digits after the comma
   production_offset_percent: 1.0 # Adjust production forecast by a percentage (1.0 = 100%, 0.8 = 80%, etc.)
                                  # Useful for winter mode when solar panels are covered with snow
+  preserve_min_grid_charge_soc: false # If true, also preserve min_grid_charge_soc as reserved energy during cheap/pre-expensive slots
 
 #--------------------------
 #  Peak Shaving

src/batcontrol/core.py

@@ -57,6 +57,29 @@
 logger = logging.getLogger(__name__)
 
 
+def _parse_optional_ratio(value, config_key: str) -> Optional[float]:
+    """Parse an optional 0..1 ratio config value."""
+    if value is None:
+        return None
+    if isinstance(value, bool):
+        raise ValueError(
+            f"{config_key} must be numeric between 0 and 1 or None, "
+            f"got {type(value).__name__}"
+        )
+    try:
+        ratio = float(value)
+    except (TypeError, ValueError) as exc:
+        raise ValueError(
+            f"{config_key} must be numeric between 0 and 1 or None, "
+            f"got {value!r}"
+        ) from exc
+    if not 0 <= ratio <= 1:
+        raise ValueError(
+            f"{config_key} must be between 0 and 1 or None, got {value!r}"
+        )
+    return ratio
+
+
 @dataclass
 class PeakShavingConfig:
     """ Holds peak shaving configuration parameters, initialized from the config dict. """
@@ -277,6 +300,21 @@ def __init__(self, configdict: dict):
             'min_price_difference', 0.05)
         self.min_price_difference_rel = self.batconfig.get(
             'min_price_difference_rel', 0)
+        self.min_grid_charge_soc = _parse_optional_ratio(
+            self.batconfig.get('min_grid_charge_soc', None),
+            'battery_control.min_grid_charge_soc'
+        )
+        self.preserve_min_grid_charge_soc = False
+        if (self.min_grid_charge_soc is not None
+                and self.min_grid_charge_soc > self.max_charging_from_grid_limit):
+            logger.warning(
+                'min_grid_charge_soc (%.2f) is above '
+                'max_charging_from_grid_limit (%.2f); grid charging cannot '
+                'reach the minimum SoC target. Increase '
+                'max_charging_from_grid_limit or lower min_grid_charge_soc.',
+                self.min_grid_charge_soc,
+                self.max_charging_from_grid_limit
+            )
 
         self.round_price_digits = 4
         self.production_offset_percent = 1.0  # Default: no offset
@@ -290,6 +328,9 @@ def __init__(self, configdict: dict):
             self.production_offset_percent = battery_control_expert.get(
                 'production_offset_percent',
                 self.production_offset_percent)
+            self.preserve_min_grid_charge_soc = battery_control_expert.get(
+                'preserve_min_grid_charge_soc',
+                self.preserve_min_grid_charge_soc)
 
         self.general_logic = CommonLogic.get_instance(
             charge_rate_multiplier=self.batconfig.get(
@@ -633,6 +674,8 @@ def run(self):
             self.min_price_difference,
             self.min_price_difference_rel,
             self.get_max_capacity(),
+            min_grid_charge_soc=self.min_grid_charge_soc,
+            preserve_min_grid_charge_soc=self.preserve_min_grid_charge_soc,
             peak_shaving_enabled=peak_shaving_config_enabled and not evcc_disable_peak_shaving,
             peak_shaving_allow_full_after=self.peak_shaving_config.allow_full_battery_after,
             peak_shaving_mode=self.peak_shaving_config.mode,
@@ -928,6 +971,9 @@ def refresh_static_values(self) -> None:
                 self.get_always_allow_discharge_limit())
             self.mqtt_api.publish_max_charging_from_grid_limit(
                 self.max_charging_from_grid_limit)
+            if self.min_grid_charge_soc is not None:
+                self.mqtt_api.publish_min_grid_charge_soc(
+                    self.min_grid_charge_soc)
             #
             self.mqtt_api.publish_min_price_difference(
                 self.min_price_difference)

src/batcontrol/logic/common.py

@@ -5,6 +5,7 @@
   - charge_rate multiplier
   """
 import logging
+from typing import Optional
 
 
 # Minimum charge rate to controlling loops between charging and
@@ -114,6 +115,43 @@ def is_charging_above_minimum(self, needed_energy: float) -> bool:
             round(needed_energy, 0))
         return False
 
+    def apply_min_grid_charge_soc_target(self, recharge_energy: float,
+                                         stored_energy: float,
+                                         min_grid_charge_soc: Optional[float]) -> float:
+        """Apply an optional minimum SoC target to grid recharge energy."""
+        if min_grid_charge_soc is None:
+            return recharge_energy
+
+        target_energy = self.max_capacity * min_grid_charge_soc
+        soc_recharge_energy = max(0.0, target_energy - stored_energy)
+        if soc_recharge_energy > recharge_energy:
+            logger.debug(
+                'Recharge increased to meet min_grid_charge_soc %.1f%%: %.1f Wh',
+                min_grid_charge_soc * 100,
+                soc_recharge_energy
+            )
+        return max(recharge_energy, soc_recharge_energy)
+
+    def apply_min_grid_charge_soc_reserve(self, reserved_energy: float,
+                                          stored_energy: float,
+                                          stored_usable_energy: float,
+                                          min_grid_charge_soc: Optional[float],
+                                          active: bool) -> float:
+        """Apply an optional minimum SoC target to protected reserve energy."""
+        if min_grid_charge_soc is None or not active:
+            return reserved_energy
+
+        min_soc_energy = max(0.0, stored_energy - stored_usable_energy)
+        target_energy = self.max_capacity * min_grid_charge_soc
+        target_usable_energy = max(0.0, target_energy - min_soc_energy)
+        if target_usable_energy > reserved_energy:
+            logger.debug(
+                'Reserve increased to meet min_grid_charge_soc %.1f%%: %.1f Wh',
+                min_grid_charge_soc * 100,
+                target_usable_energy
+            )
+        return max(reserved_energy, target_usable_energy)
+
     def calculate_charge_rate(self, charge_rate: float) -> int:
         """ Calculate the charge rate based on the charge rate multiplier.
         Args:

src/batcontrol/logic/default.py

@@ -304,6 +304,25 @@ def __is_discharge_allowed(self, calc_input: CalculationInput,
             # add_remaining required_energy to reserved_storage
             reserved_storage += required_energy
 
+        min_grid_charge_soc_active = (
+            self.calculation_parameters.preserve_min_grid_charge_soc
+            and reserved_storage > 0
+            and self.__has_grid_charge_soc_price_signal(
+                consumption,
+                prices,
+                max_slots,
+                current_price,
+                min_dynamic_price_difference
+            )
+        )
+        reserved_storage = self.common.apply_min_grid_charge_soc_reserve(
+            reserved_storage,
+            calc_input.stored_energy,
+            calc_input.stored_usable_energy,
+            self.calculation_parameters.min_grid_charge_soc,
+            min_grid_charge_soc_active
+        )
+
         self.calculation_output.reserved_energy = reserved_storage
 
         if len(higher_price_slots) > 0:
@@ -340,6 +359,19 @@ def __is_discharge_allowed(self, calc_input: CalculationInput,
 
         return False
 
+    def __has_grid_charge_soc_price_signal(self, consumption: np.ndarray,
+                                           prices: dict,
+                                           max_slots: int,
+                                           current_price: float,
+                                           min_dynamic_price_difference: float) -> bool:
+        """Return True when a future price justifies preserving a grid-charge SoC target."""
+        for slot in range(max_slots):
+            future_price = prices[slot]
+            if (consumption[slot] > 0
+                    and future_price > current_price + min_dynamic_price_difference):
+                return True
+        return False
+
  # %%
     def __get_required_recharge_energy(self, calc_input: CalculationInput,
                                               net_consumption: list, prices: dict) -> float:
@@ -429,6 +461,12 @@ def __get_required_recharge_energy(self, calc_input: CalculationInput,
             self.calculation_output.required_recharge_energy = recharge_energy
             return recharge_energy
 
+        recharge_energy = self.common.apply_min_grid_charge_soc_target(
+            recharge_energy,
+            calc_input.stored_energy,
+            self.calculation_parameters.min_grid_charge_soc
+        )
+
         free_capacity = calc_input.free_capacity
 
         if recharge_energy > free_capacity:

src/batcontrol/logic/logic_interface.py

@@ -26,6 +26,12 @@ class CalculationParameters:
     min_price_difference: float
     min_price_difference_rel: float
     max_capacity: float # Maximum capacity of the battery in Wh (excludes MAX_SOC)
+    # Optional minimum SoC target used when grid charging is already economical.
+    # None disables the target. Values are ratios from 0.0 to 1.0.
+    min_grid_charge_soc: Optional[float] = None
+    # Expert option: also preserve the target as reserved energy during
+    # cheap/pre-expensive windows.
+    preserve_min_grid_charge_soc: bool = False
     # Peak shaving parameters
     peak_shaving_enabled: bool = False
     peak_shaving_allow_full_after: int = 14  # Hour (0-23)
@@ -40,6 +46,18 @@ class CalculationParameters:
     peak_shaving_price_limit: Optional[float] = None
 
     def __post_init__(self):
+        if self.min_grid_charge_soc is not None:
+            if (isinstance(self.min_grid_charge_soc, bool)
+                    or not isinstance(self.min_grid_charge_soc, (int, float))):
+                raise ValueError(
+                    f"min_grid_charge_soc must be numeric between 0 and 1 or None, "
+                    f"got {type(self.min_grid_charge_soc).__name__}"
+                )
+            if not 0 <= self.min_grid_charge_soc <= 1:
+                raise ValueError(
+                    f"min_grid_charge_soc must be between 0 and 1 or None, "
+                    f"got {self.min_grid_charge_soc}"
+                )
         if not 0 <= self.peak_shaving_allow_full_after <= 23:
             raise ValueError(
                 f"peak_shaving_allow_full_after must be 0-23, "

src/batcontrol/logic/next.py

@@ -594,6 +594,25 @@ def _is_discharge_allowed(self, calc_input: CalculationInput,
             # add_remaining required_energy to reserved_storage
             reserved_storage += required_energy
 
+        min_grid_charge_soc_active = (
+            self.calculation_parameters.preserve_min_grid_charge_soc
+            and reserved_storage > 0
+            and self._has_grid_charge_soc_price_signal(
+                consumption,
+                prices,
+                max_slots,
+                current_price,
+                min_dynamic_price_difference
+            )
+        )
+        reserved_storage = self.common.apply_min_grid_charge_soc_reserve(
+            reserved_storage,
+            calc_input.stored_energy,
+            calc_input.stored_usable_energy,
+            self.calculation_parameters.min_grid_charge_soc,
+            min_grid_charge_soc_active
+        )
+
         self.calculation_output.reserved_energy = reserved_storage
 
         if len(higher_price_slots) > 0:
@@ -625,6 +644,19 @@ def _is_discharge_allowed(self, calc_input: CalculationInput,
 
         return False
 
+    def _has_grid_charge_soc_price_signal(self, consumption: np.ndarray,
+                                          prices: dict,
+                                          max_slots: int,
+                                          current_price: float,
+                                          min_dynamic_price_difference: float) -> bool:
+        """Return True when a future price justifies preserving a grid-charge SoC target."""
+        for slot in range(max_slots):
+            future_price = prices[slot]
+            if (consumption[slot] > 0
+                    and future_price > current_price + min_dynamic_price_difference):
+                return True
+        return False
+
     # ------------------------------------------------------------------ #
     #  Recharge energy calculation (same as DefaultLogic)                 #
     # ------------------------------------------------------------------ #
@@ -714,6 +746,12 @@ def _get_required_recharge_energy(self, calc_input: CalculationInput,
             self.calculation_output.required_recharge_energy = recharge_energy
             return recharge_energy
 
+        recharge_energy = self.common.apply_min_grid_charge_soc_target(
+            recharge_energy,
+            calc_input.stored_energy,
+            self.calculation_parameters.min_grid_charge_soc
+        )
+
         free_capacity = calc_input.free_capacity
 
         if recharge_energy > free_capacity:

src/batcontrol/mqtt_api.py

@@ -9,6 +9,8 @@
 - /mode: operational mode (-1 = charge from grid, 0 = avoid discharge, 8 = limit battery charge, 10 = discharge allowed)
 - /max_charging_from_grid_limit: charge limit in 0.1-1
 - /max_charging_from_grid_limit_percent: charge limit in %
+- /min_grid_charge_soc: optional minimum grid-charge target in 0.0-1.0
+- /min_grid_charge_soc_percent: optional minimum grid-charge target in %
 - /always_allow_discharge_limit: always discharge limit in 0.1-1
 - /always_allow_discharge_limit_percent: always discharge limit in %
 - /always_allow_discharge_limit_capacity: always discharge limit in Wh
@@ -386,6 +388,21 @@ def publish_max_charging_from_grid_limit(
                 f'{charge_limit:.2f}'
             )
 
+    def publish_min_grid_charge_soc(self, min_grid_charge_soc: float) -> None:
+        """ Publish the optional minimum grid-charge SoC target to MQTT
+            /min_grid_charge_soc_percent
+            /min_grid_charge_soc as digit.
+        """
+        if self.client.is_connected():
+            self.client.publish(
+                self.base_topic + '/min_grid_charge_soc_percent',
+                f'{min_grid_charge_soc * 100:.0f}'
+            )
+            self.client.publish(
+                self.base_topic + '/min_grid_charge_soc',
+                f'{min_grid_charge_soc:.2f}'
+            )
+
     def publish_min_price_difference(
             self, min_price_difference: float) -> None:
         """ Publish the minimum price difference to MQTT found in config
@@ -595,6 +612,16 @@ def send_mqtt_discovery_messages(self) -> None:
             step_value=0.1,
             initial_value=0.9)
 
+        self.publish_mqtt_discovery_message(
+            "Minimum Grid Charge SOC",
+            "batcontrol_min_grid_charge_soc",
+            "sensor",
+            "battery",
+            "%",
+            self.base_topic +
+            "/min_grid_charge_soc_percent",
+            entity_category="diagnostic")
+
         self.publish_mqtt_discovery_message(
             "Min Price Difference",
             "batcontrol_min_price_difference",