feat(battery): plan a periodic full charge of the stationary battery (#1486)

LFP batteries can only recalibrate their state of charge close to full
charge, so many systems expect a full charge every few weeks and some
start one on their own at an arbitrary, possibly expensive time. EOS can
now plan it instead (GENETIC only, off by default).

New battery settings: full_charge_interval_days (unset = off),
full_charge_soc_percentage, full_charge_detect_percentage,
full_charge_allow_grid and the informational cell_chemistry. The last
full charge is read from the measured SoC history; nothing is stored.
When the next one is due within the horizon (or overdue), the maximum
SoC is raised to the target. With full_charge_allow_grid the optimizer
must reach the target once before the deadline, which is at least 24 h
away (penalty battery_full_charge_miss on the best SoC reached), so it
picks the cheapest time itself. Without it, PV surplus may fill the
battery but the target is not enforced.

Signed-off-by: SchaafAlexander <86607129+SchaafAlexander@users.noreply.github.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
SchaafAlexander
2026-11-08 07:40:13 +01:00
committed by GitHub
co-authored by Claude Opus 5.5
parent b33953d445
commit a06473b6d6
12 changed files with 1085 additions and 6 deletions
+43 -1
View File
@@ -1,7 +1,7 @@
from typing import Any, Iterator, Optional
import numpy as np
from pydantic import Field, field_validator
from pydantic import Field, field_validator, model_validator
from akkudoktoreos.devices.settings.batterysettings import BATTERY_DEFAULT_CHARGE_RATES
from akkudoktoreos.optimization.genetic.geneticdevices import DeviceParameters
@@ -110,6 +110,48 @@ class SolarPanelBatteryParameters(BaseBatteryParameters):
},
)
max_charge_power_w: Optional[float] = max_charging_power_field()
full_charge_soc_percentage: Optional[int] = Field(
default=None,
ge=1,
le=100,
json_schema_extra={
"description": (
"State of charge a due full charge has to reach at least once "
"before 'full_charge_deadline_datetime' [%]. Must not exceed "
"'max_soc_percentage'. None means no full charge is due."
),
"examples": [None, 100],
},
)
full_charge_deadline_datetime: Optional[DateTime] = Field(
default=None,
json_schema_extra={
"description": (
"Moment by which 'full_charge_soc_percentage' has to be reached. "
"A date time without timezone is read as local time. None means "
"end of the optimization horizon."
),
"examples": [None, "2026-11-20T07:00:00+01:00"],
},
)
@field_validator("full_charge_deadline_datetime", mode="before")
@classmethod
def transform_full_charge_deadline(cls, value: Any) -> Optional[DateTime]:
"""Accept the usual date time representations, naive input is local time."""
if value is None:
return None
return to_datetime(value)
@model_validator(mode="after")
def validate_full_charge_target(self) -> "SolarPanelBatteryParameters":
"""A full charge target above the maximum SoC could never be reached."""
if (
self.full_charge_soc_percentage is not None
and self.full_charge_soc_percentage > self.max_soc_percentage
):
raise ValueError("full_charge_soc_percentage must be <= max_soc_percentage")
return self
class ElectricVehicleParameters(BaseBatteryParameters):
@@ -4,7 +4,7 @@ Note: Used for the GENETIC and GENETIC0 algorithm.
"""
import re
from typing import TYPE_CHECKING, Any, Optional
from typing import TYPE_CHECKING, Any, Literal, Optional
import numpy as np
from numpydantic import NDArray, Shape
@@ -212,6 +212,80 @@ class BatteriesCommonSettings(DevicesBaseSettings):
return None
return to_datetime(value)
# ------------------------------------------------------------------
# Periodic full charge (stationary batteries, GENETIC only)
# ------------------------------------------------------------------
cell_chemistry: Optional[Literal["LFP", "NMC", "NCA", "LTO"]] = Field(
default=None,
json_schema_extra={
"x-scope": [str(ConfigScope.GENETIC)],
"description": (
"Cell chemistry of the battery. Informational only, it does not "
"enable anything. LFP batteries usually need a periodic full charge "
"to recalibrate their state of charge, see "
"'full_charge_interval_days'."
),
"examples": [None, "LFP"],
},
)
full_charge_interval_days: Optional[float] = Field(
default=None,
gt=0,
allow_inf_nan=False,
json_schema_extra={
"x-scope": [str(ConfigScope.GENETIC)],
"description": (
"Maximum number of days between two full charges of a stationary "
"battery [d]. The last full charge is taken from the measured state "
"of charge. When the next one falls due within the optimization "
"horizon, the maximum state of charge is raised to "
"'full_charge_soc_percentage'. None disables the periodic full "
"charge. Ignored for electric vehicles."
),
"examples": [None, 21],
},
)
full_charge_soc_percentage: int = Field(
default=100,
ge=1,
le=100,
json_schema_extra={
"x-scope": [str(ConfigScope.GENETIC)],
"description": (
"State of charge a periodic full charge has to reach [%]. May be "
"above 'max_soc_percentage', which then applies on the other days."
),
"examples": [100],
},
)
full_charge_detect_percentage: int = Field(
default=99,
ge=1,
le=100,
json_schema_extra={
"x-scope": [str(ConfigScope.GENETIC)],
"description": (
"Measured state of charge from which the battery counts as fully "
"charged [%]. Many battery management systems never report exactly "
"100 %."
),
"examples": [99],
},
)
full_charge_allow_grid: bool = Field(
default=False,
json_schema_extra={
"x-scope": [str(ConfigScope.GENETIC)],
"description": (
"Enforce a due full charge, charging from the grid if needed. "
"False only raises the maximum state of charge, so that PV surplus "
"can fill the battery, without enforcing the target."
),
"examples": [False, True],
},
)
def to_genetic_pv_bat_param(self) -> "SolarPanelBatteryParameters":
"""Return SolarPanelBatteryParameters for the GENETIC optimizer."""
from akkudoktoreos.devices.genetic.battery import SolarPanelBatteryParameters
@@ -317,6 +391,8 @@ class BatteriesCommonSettings(DevicesBaseSettings):
def _validate_soc_range(self) -> "BatteriesCommonSettings":
if self.min_soc_percentage > self.max_soc_percentage:
raise ValueError("min_soc_percentage must be <= max_soc_percentage")
if self.full_charge_detect_percentage > self.full_charge_soc_percentage:
raise ValueError("full_charge_detect_percentage must be <= full_charge_soc_percentage")
if (
self.min_charge_power_w is not None
and self.max_charge_power_w is not None
@@ -15,6 +15,10 @@ from akkudoktoreos.core.coreabc import (
)
from akkudoktoreos.devices.settings.batterysettings import BatteriesCommonSettings
from akkudoktoreos.optimization.genetic.forecast import bounded_forecast_array
from akkudoktoreos.optimization.genetic.fullcharge import (
enforced_full_charge_deadline,
full_charge_deadline,
)
from akkudoktoreos.optimization.genetic.geneticabc import GeneticParametersBaseModel
from akkudoktoreos.optimization.genetic.geneticparams import (
GeneticEnergyManagementParameters,
@@ -133,11 +137,64 @@ class ConfigOptimizationRequest(
)
return 0
async def full_charge(device: BatteriesCommonSettings) -> Optional[DateTime]:
"""Deadline of the next periodic full charge, None if disabled or unknown."""
interval_days = device.full_charge_interval_days
if interval_days is None:
return None
try:
dates, values = await self.measurement.key_to_lists(
key=device.measurement_key_soc_factor,
start_datetime=observation_time.subtract(days=math.ceil(interval_days)),
end_datetime=observation_time.add(seconds=1),
dropna=False,
)
except KeyError:
dates, values = [], []
deadline = full_charge_deadline(
zip(dates, values),
now=observation_time,
interval_days=interval_days,
detect_percentage=device.full_charge_detect_percentage,
)
if deadline is None:
logger.warning(
"No state of charge history for {}; cannot schedule its periodic full charge.",
device.device_id,
)
return deadline
battery = None
if groups["batteries"]:
device = groups["batteries"][0]
battery = device.to_genetic_pv_bat_param()
battery.initial_soc_percentage = await state(device)
deadline = await full_charge(device)
horizon_end = start.add(seconds=settings.horizon * settings.interval_sec)
if deadline is not None and deadline.timestamp() <= horizon_end.timestamp():
# Due within the horizon: lift the upper limit so the battery may
# reach the target, and enforce the target only if grid charging
# is allowed for it.
target = device.full_charge_soc_percentage
battery.max_soc_percentage = max(battery.max_soc_percentage, target)
if device.full_charge_allow_grid:
enforced = enforced_full_charge_deadline(deadline, observation_time)
battery.full_charge_soc_percentage = target
battery.full_charge_deadline_datetime = enforced
logger.info(
"Periodic full charge of {} to {}% due by {}; enforced by {}.",
device.device_id,
target,
deadline,
enforced,
)
else:
logger.info(
"Periodic full charge of {} to {}% due by {}; PV surplus only.",
device.device_id,
target,
deadline,
)
ev = None
if groups["electric_vehicles"]:
device = groups["electric_vehicles"][0]
@@ -0,0 +1,83 @@
"""Periodic full charge of a stationary battery.
LFP batteries can only recalibrate their state of charge close to full charge,
so many battery management systems expect a full charge every few weeks. The
last full charge is read from the measured state of charge; nothing is stored.
"""
import math
from datetime import timedelta
from typing import Any, Iterable, Optional
from akkudoktoreos.utils.datetimeutil import DateTime
# A due full charge never has to be finished in less time than this, so the
# optimizer can still pick a cheap slot even when the deadline is close or
# already overdue.
FULL_CHARGE_MIN_LEAD_HOURS = 24
def full_charge_deadline(
samples: Iterable[tuple[DateTime, Any]],
now: DateTime,
interval_days: float,
detect_percentage: float,
) -> Optional[DateTime]:
"""Return the moment by which the next full charge is due.
Args:
samples: Measured state of charge as (time, factor 0.0 ... 1.0) pairs,
covering at least the last ``interval_days`` before ``now``. Samples
after ``now`` and invalid values are ignored.
now: Current time.
interval_days: Maximum number of days between two full charges.
detect_percentage: State of charge from which a sample counts as full [%].
Returns:
The deadline, which may already have passed, or None if there is no
valid sample at all.
Without a full charge among the samples, the interval counts from the oldest
sample. A history that covers the whole interval therefore makes the full
charge due now, while a fresh installation is not forced into a full charge
on its first day.
"""
interval = timedelta(days=interval_days)
valid = [
(date, value)
for date, value in samples
if value is not None
and isinstance(value, (int, float))
and math.isfinite(value)
and date.timestamp() <= now.timestamp()
]
if not valid:
return None
full = [date for date, value in valid if value * 100 >= detect_percentage - 1e-9]
if full:
reference = max(full, key=lambda date: date.timestamp())
else:
reference = min((date for date, _ in valid), key=lambda date: date.timestamp())
return reference + interval
def enforced_full_charge_deadline(deadline: DateTime, now: DateTime) -> DateTime:
"""Deadline handed to the optimizer, at least the minimum lead time from now.
Whether a full charge is due is decided on the real deadline. Only the
moment it has to be finished by is moved, so that a close or overdue full
charge can still use the cheapest slot of the next hours.
Args:
deadline: Deadline from ``full_charge_deadline``.
now: Current time.
Returns:
``deadline``, or ``FULL_CHARGE_MIN_LEAD_HOURS`` after ``now`` if that is later.
"""
earliest = now.add(hours=FULL_CHARGE_MIN_LEAD_HOURS)
if deadline.timestamp() < earliest.timestamp():
return earliest
return deadline
@@ -1082,6 +1082,10 @@ class GeneticOptimization(OptimizationBase):
# Slot by which the EV has to reach its target SoC. None means the SoC is
# only required at the end of the horizon (the behaviour without a deadline).
self._ev_soc_deadline_slot: Optional[int] = None
# Slot by which a due periodic full charge of the stationary battery has
# to have happened. None means no full charge is enforced in this run.
self._full_charge_deadline_slot: Optional[int] = None
self._full_charge_target_soc: Optional[int] = None
# Value of the energy left in the battery at the end of the
# horizon. None means the fixed scalar terminal value is used instead.
self._terminal_value_curve: Optional[TerminalValueCurve] = None
@@ -1236,6 +1240,32 @@ class GeneticOptimization(OptimizationBase):
# A deadline in the past means the target is due right now.
return max(deadline_slot, start_slot)
def _full_charge_slot(self, parameters: GeneticOptimizationParameters) -> Optional[int]:
"""Slot by which the battery has to reach its full charge target once.
Args:
parameters: Optimization parameters of this run.
Returns:
Absolute slot index, capped to the end of the control horizon, or None
when no full charge is due.
"""
battery = parameters.pv_battery
if battery is None or battery.full_charge_soc_percentage is None:
return None
start_slot = self._control_start_slot()
deadline = battery.full_charge_deadline_datetime
if deadline is None:
return self.control_end_slot
if self._slot0_datetime is None:
raise ValueError("Full charge deadline requires a run start timestamp.")
seconds = (
deadline.in_timezone(self._slot0_datetime.timezone) - self._slot0_datetime
).total_seconds()
deadline_slot = math.floor(seconds / (self.slot_duration_h * 3600) + 1e-9)
return min(max(deadline_slot, start_slot), self.control_end_slot)
def _validate_forecast_availability(self) -> None:
"""Use only the contiguous finite forecast prefix after now."""
start = self._control_start_slot()
@@ -3560,6 +3590,38 @@ class GeneticOptimization(OptimizationBase):
except (TypeError, ValueError):
return _pack_genes([int(value) for value in relevant])
def _battery_max_soc_until_full_charge_deadline(
self, simulation_result: dict[str, Any], start_slot: int
) -> Optional[float]:
"""Highest battery SoC reached before the full charge deadline [%].
The full charge only has to happen once, so the maximum counts, not the
SoC at the deadline. ``akku_soc_pro_stunde`` holds the SoC at the start of
each slot; the SoC at the start of the deadline slot includes every charge
that finished in time. A deadline at the end of the horizon also counts
the SoC after the last slot.
Args:
simulation_result: Result of the simulation run for this individual.
start_slot: Slot index the result arrays start at.
Returns:
State of charge in percent, or None when no full charge is due.
"""
deadline_slot = getattr(self, "_full_charge_deadline_slot", None)
battery = self.simulation.battery
if deadline_slot is None or battery is None:
return None
soc_per_slot = simulation_result.get("akku_soc_pro_stunde")
if soc_per_slot is None:
return battery.current_soc_percentage()
index = deadline_slot - start_slot
reached = [float(value) for value in soc_per_slot[: max(index, 0) + 1]]
if index >= len(soc_per_slot):
reached.append(battery.current_soc_percentage())
reached = [value for value in reached if math.isfinite(value)]
return max(reached) if reached else battery.current_soc_percentage()
def _ev_soc_at_deadline(self, simulation_result: dict[str, Any], start_slot: int) -> float:
"""EV state of charge the target is checked against [%].
@@ -3794,6 +3856,25 @@ class GeneticOptimization(OptimizationBase):
excess_cost_per_wh = break_even_price - best_uncovered_price
gesamtbilanz += ac_wh * excess_cost_per_wh * ac_penalty_factor
# --- Periodic full charge (e.g. LFP recalibration) ---
# Only the shortfall of the best SoC reached before the deadline counts,
# so the optimizer is free to pick the cheapest slot for it.
# Lightweight callers construct the optimizer without __init__ (see
# evaluate()); a missing target must behave like no full charge.
target = getattr(self, "_full_charge_target_soc", None)
if target is not None:
reached = self._battery_max_soc_until_full_charge_deadline(
simulation_result, start_hour
)
if reached is not None and reached < target:
try:
full_charge_penalty = float(
self._genetic_cfg.penalties["battery_full_charge_miss"]
)
except Exception:
full_charge_penalty = 10.0
gesamtbilanz += (target - reached) * full_charge_penalty
if self.optimize_ev and parameters.ev and self.simulation.ev:
try:
penalty = self._genetic_cfg.penalties["ev_soc_miss"]
@@ -4242,6 +4323,16 @@ class GeneticOptimization(OptimizationBase):
# EV charging deadline (departure). Resolved once per run; the seeding
# heuristic and the SoC penalty both read it.
self._ev_soc_deadline_slot = self._ev_deadline_slot(parameters)
self._full_charge_deadline_slot = self._full_charge_slot(parameters)
self._full_charge_target_soc = (
parameters.pv_battery.full_charge_soc_percentage
if self._full_charge_deadline_slot is not None and parameters.pv_battery
else None
)
if self._full_charge_deadline_slot is not None:
logger.debug(
"Battery full charge required by slot {}.", self._full_charge_deadline_slot
)
if self._ev_soc_deadline_slot is not None:
logger.debug(
"EV target SoC required by slot {} ({}).",
@@ -213,6 +213,7 @@ class GeneticCommonSettings(SettingsBaseModel):
default_factory=lambda: dict[str, float | int | str](
ev_soc_miss=10,
ac_charge_break_even=1.0,
battery_full_charge_miss=10,
),
json_schema_extra={
"description": "Penalty parameters used in fitness evaluation.",