Skip to content

Battery storage

bess

Battery storage: round-trip efficiency, LTSA availability, cycles and rainflow, capacity tests.

efficiency

Round-trip efficiency (RTE) of a battery at the meter, corrected for the change of stored energy.

Per interval the metered power p_meter (positive = export, discharge) is split into energy out and energy in. Where the auxiliaries (HVAC, transformers, control) sit behind the revenue meter (aux_metering = "behind_meter"), the meter sees the net flow and the flow of the power conversion system is p_meter + p_aux; where they have their own supply ("separate"), their energy is added to the input for the RTE that includes them.

Over a window that does not end at the state of charge it started with, the RTE is corrected for the change of stored energy dE with a one-way efficiency eta between the meter and the store:

dE >= 0:  RTE = (E_out + eta dE) / E_in          what is left in store is worth eta dE at the meter
dE <  0:  RTE = E_out / (E_in + |dE| / eta)      what was taken from store would need |dE| / eta to refill

eta is given or found as sqrt(RTE) of the whole record by fixed-point iteration. With dE = 0 both lines reduce to E_out / E_in. The stored energy is read at the instants that bound each window (the start of the window and the start of the next one).

Any consistent units work: MW and MWh, or kW and kWh.

MIN_EFC module-attribute

MIN_EFC: dict[str, float] = {'D': 0.3, 'M': 5.0, 'Q': 15.0, 'Y': 50.0, 'CY': 50.0}

Default minimum equivalent full cycles for a window's RTE to be assessed.

corrected_rte

corrected_rte(e_out: ArrayLike, e_in: ArrayLike, d_stored: ArrayLike, eta: float) -> np.ndarray

RTE corrected for the change of stored energy (see the module description).

Parameters:

Name Type Description Default
e_out array - like

Energy discharged and charged over each window.

required
e_in array - like

Energy discharged and charged over each window.

required
d_stored array - like

Change of stored energy over each window (end minus start); NaN gives NaN.

required
eta float

One-way efficiency between the meter and the store.

required

Examples:

A day that charged 100 and discharged 80 but ended 5 higher than it started:

>>> corrected_rte([80.0], [100.0], [5.0], 0.95).round(4).tolist()
[0.8475]

auto_eta

auto_eta(e_out: float, e_in: float, d_stored: float, iterations: int = 50) -> float

One-way efficiency sqrt(RTE) consistent with the corrected RTE of the whole record.

Examples:

>>> round(auto_eta(81.0, 100.0, 0.0), 4)
0.9

stored_energy

stored_energy(site: DataFrame, soc_energy: float | None = None) -> pd.Series | None

Stored energy at every instant: e_stored if present, else soc times the energy SoC = 1 stands for.

interval_flows

interval_flows(site: DataFrame, aux_metering: AuxMetering = 'behind_meter', soc_energy: float | None = None) -> pd.DataFrame

Energy per interval: meter out and in, PCS out and in, auxiliaries, DC, and the stored energy.

Parameters:

Name Type Description Default
site DataFrame

Regular UTC index (interval starts) and canonical columns: p_meter (required), p_aux, p_dc, e_stored or soc.

required
aux_metering ('behind_meter', 'separate', 'none')

Where the auxiliary consumption is metered (see the module description).

"behind_meter"
soc_energy float

Energy that SoC = 1 stands for, to turn soc into stored energy.

None

Returns:

Type Description
DataFrame

e_out, e_in, e_aux, e_out_pcs, e_in_pcs, (e_dc_dis, e_dc_ch) and stored (NaN columns where the signal is missing).

window_keys

window_keys(index: DatetimeIndex, freq: WindowFreq = 'M', tz: str = 'Europe/Warsaw', contract_start: Timestamp | str | None = None) -> pd.Series

Window label of every interval: local day, month, quarter, year, or contract year (CY).

window_rte

window_rte(site: DataFrame, freq: WindowFreq = 'M', *, reference_energy: float, aux_metering: AuxMetering = 'behind_meter', soc_energy: float | None = None, tz: str = 'Europe/Warsaw', contract_start: Timestamp | str | None = None, eta: float | Literal['auto'] = 'auto', min_efc: float | None = None, min_coverage: float = 0.95) -> pd.DataFrame

Energy in and out and RTE per window, with and without the auxiliaries.

Parameters:

Name Type Description Default
site DataFrame

Regular UTC index and canonical columns p_meter and e_stored or soc (needed for the SoC correction); p_aux for the split with and without auxiliaries.

required
freq ('D', 'M', 'Q', 'Y', 'CY')

Window: local day, month, quarter, year, or contract year from contract_start.

"D"
reference_energy float

Energy for the equivalent full cycles (e.g. usable energy), same unit as the energies.

required
aux_metering ('behind_meter', 'separate', 'none')

Where the auxiliary consumption is metered (see the module description).

"behind_meter"
soc_energy float

Energy that SoC = 1 stands for (needed when the site has soc and no e_stored).

None
tz str

Time zone of the windows (a local day has 23 or 25 hours at the DST changes).

"Europe/Warsaw"
contract_start Timestamp or str

Start of contract year 1 for freq="CY".

None
eta float or 'auto'

One-way efficiency for the correction; auto = sqrt(RTE) of the whole record (excluding the auxiliaries when possible).

"auto"
min_efc float

Minimum equivalent full cycles for valid (default MIN_EFC of the window).

None
min_coverage float

Minimum share of intervals with a meter reading for valid.

0.95

Returns:

Type Description
DataFrame

Per window: summed flows, coverage, hours, stored_start, stored_end, d_stored, rte_raw (uncorrected E_out / E_in), rte (including auxiliaries), rte_excl_aux, efc (PCS input energy / reference energy), eta and valid.

Examples:

One day of one cycle, ending 2 MWh higher than it started:

>>> idx = pd.date_range("2024-06-01 00:00", periods=6, freq="4h", tz="Europe/Warsaw").tz_convert("UTC")
>>> site = pd.DataFrame({"p_meter": [-25.0, 0.0, 19.5, 0.0, 0.0, 0.0],
...                      "e_stored": [10.0, 105.0, 105.0, 22.0, 12.0, 12.0]}, index=idx)
>>> w = window_rte(site, "D", reference_energy=100.0, aux_metering="none", eta=0.95, min_efc=0.5)
>>> w[["e_in", "e_out", "d_stored", "rte_raw", "rte"]].round(4)
            e_in  e_out  d_stored  rte_raw   rte
window
2024-06-01 100.0   78.0       2.0     0.78  0.799

availability

Contractual availability of a battery as typically defined in a long-term service agreement (LTSA).

Per interval i the measured availability a_i follows from the available power the EMS reports in each direction d (discharge, charge) against the contracted power, and for the energy formula also from the available energy:

time:      a_i = 1 if P_avail,d / P_contract,d >= threshold in every direction, else 0
capacity:  a_i = min_d min(1, P_avail,d / P_contract,d)          (or the mean over the directions)
energy:    a_i = capacity term x min(1, E_avail / E_contract)

Events from the operator's log decide how an interval counts:

  • deemed_available: a_i = 1 (e.g. an instruction of the grid operator, a grid outage, an owner request);
  • excluded: removed from numerator and denominator (e.g. force majeure, the capacity test itself);
  • allowance: excluded until the planned-maintenance allowance of the period is used up, counted afterwards;
  • counted: a_i as measured (forced outages; also intervals without an event).

The availability of a period is A = sum(a_i dt) / sum(dt) over the intervals not excluded. Intervals without data and without an event are excluded, counted as available or as unavailable (missing_data). When events overlap, the treatment listed first in PRIORITY wins. The allowance is given per year and pro-rated to the length of the period.

Which categories exist and how each is treated is contract-specific: the example in res_asset_kpi/data/bess_ltsa.example.yaml is an illustration, not a reading of any LTSA.

TREATMENTS module-attribute

TREATMENTS = ('excluded', 'deemed_available', 'allowance', 'counted')

Ways an event category can count (see the module description).

PRIORITY module-attribute

PRIORITY = TREATMENTS

Order in which overlapping treatments win (first wins).

measured_availability

measured_availability(site: DataFrame, *, power: float, charge_power: float | None = None, formula: Literal['time', 'capacity', 'energy'] = 'time', threshold: float = 0.95, directions: Sequence[str] = ('discharge', 'charge'), combine: Literal['min', 'mean'] = 'min', energy: float | None = None) -> pd.Series

Measured availability a_i per interval (NaN where a needed signal is missing).

Parameters:

Name Type Description Default
site DataFrame

Columns p_avail_dis and/or p_avail_ch (magnitudes), and e_avail for the energy formula.

required
power float

Contracted discharge power.

required
charge_power float

Contracted charge power (default: power).

None
formula ('time', 'capacity', 'energy')
"time"
threshold float

Time formula: share of the contracted power needed to count as available.

0.95
directions sequence of {"discharge", "charge"}
both
combine ('min', 'mean')

How the capacity formula combines the directions.

"min"
energy float

Contracted energy (energy formula).

None

Examples:

>>> site = pd.DataFrame({"p_avail_dis": [50.0, 40.0, 50.0], "p_avail_ch": [50.0, 50.0, None]})
>>> measured_availability(site, power=50.0).tolist()
[1.0, 0.0, nan]
>>> measured_availability(site, power=50.0, formula="capacity").tolist()
[1.0, 0.8, nan]

interval_treatment

interval_treatment(index: DatetimeIndex, events: DataFrame | None, treatments: Mapping[str, str]) -> pd.DataFrame

Event category and its treatment for every interval (by the interval's start time).

Parameters:

Name Type Description Default
index DatetimeIndex

Interval starts (UTC).

required
events DataFrame or None

start, end (time-zone aware) and category.

required
treatments mapping

Category -> one of TREATMENTS.

required

Returns:

Type Description
DataFrame

category (missing without an event) and treatment (counted without an event).

ltsa_availability

ltsa_availability(measured: Series, *, events: DataFrame | None = None, treatments: Mapping[str, str] | None = None, planned_allowance_h: float = 0.0, missing_data: Literal['excluded', 'available', 'unavailable'] = 'excluded', periods: ArrayLike | None = None, guaranteed: float | None = None) -> pd.DataFrame

Availability per period with event treatments, a planned-maintenance allowance and missing data.

Parameters:

Name Type Description Default
measured Series

a_i per interval (measured_availability) on a regular UTC index.

required
events DataFrame

Operator's log: start, end, category.

None
treatments mapping

Category -> treatment (see the module description).

None
planned_allowance_h float

Planned maintenance hours per year that are excluded; pro-rated to each period's length.

0.0
missing_data ('excluded', 'available', 'unavailable')

Treatment of counted intervals without a measurement.

"excluded"
periods array - like

Period label per interval (e.g. period_labels or contract years); one period by default.

None
guaranteed float

Guaranteed availability, copied to the output for convenience.

None

Returns:

Type Description
DataFrame

Per period: hours, hours_excluded, hours_deemed_available, hours_planned_in_allowance, hours_planned_counted, hours_missing, equivalent_unavailable_hours, availability and guaranteed.

Examples:

Ten hours; a forced outage of two hours and a grid-operator instruction of one hour:

>>> idx = pd.date_range("2024-05-01", periods=10, freq="1h", tz="UTC")
>>> a = pd.Series([1, 1, 0, 0, 0, 1, 1, 1, 1, 1], index=idx, dtype=float)
>>> ev = pd.DataFrame({"start": [idx[4]], "end": [idx[5]], "category": ["tso_instruction"]})
>>> r = ltsa_availability(a, events=ev, treatments={"tso_instruction": "deemed_available"})
>>> float(r["availability"].iloc[0])
0.8

cycles

Equivalent full cycles, throughput and rainflow counting of the state of charge.

  • Equivalent full cycles (EFC) on the basis a warranty names: discharged energy at the meter or at the DC terminals, or half the metered throughput (charge + discharge) / 2, divided by the reference energy.
  • Rainflow counting of the state of charge with the three-point method of ASTM E1049-85, the residue counted as half cycles, and an optional hysteresis filter against measurement noise. sum(depth x count) is the rainflow EFC.

DEPTH_EDGES module-attribute

DEPTH_EDGES = (0.0, 0.05, 0.1, 0.2, 0.4, 0.6, 0.8, 1.0001)

Default bin edges of the depth of discharge histogram (fractions of the reference energy).

throughput

throughput(site: DataFrame, interval_h: float, basis: ThroughputBasis = 'discharge_meter') -> pd.Series

Energy per interval that a warranty counts as throughput.

Parameters:

Name Type Description Default
site DataFrame

p_meter (positive = discharge) and, for discharge_dc, p_dc.

required
interval_h float

Interval length, hours.

required
basis ('discharge_meter', 'discharge_dc', 'half_throughput_meter')
"discharge_meter"

Examples:

>>> site = pd.DataFrame({"p_meter": [-40.0, 30.0]})
>>> throughput(site, 1.0, "half_throughput_meter").tolist()
[20.0, 15.0]

equivalent_full_cycles

equivalent_full_cycles(energy: ArrayLike, reference_energy: float) -> float

Equivalent full cycles: summed throughput energy / reference energy.

Examples:

>>> equivalent_full_cycles([60.0, 90.0, 50.0], 100.0)
2.0

reversals

reversals(x: ArrayLike, min_range: float = 0.0) -> tuple[np.ndarray, np.ndarray]

Positions and values of the turning points of a signal.

The first and last points are kept; NaNs and plateaus are dropped. With min_range > 0 a turning point only counts once the signal has moved back by more than that (hysteresis).

Examples:

>>> pos, val = reversals([0.5, 0.6, 0.9, 0.9, 0.2, 0.4, 0.3])
>>> pos.tolist(), val.tolist()
([0, 2, 4, 5, 6], [0.5, 0.9, 0.2, 0.4, 0.3])

rainflow

rainflow(x: ArrayLike, min_range: float = 0.0) -> pd.DataFrame

Rainflow cycles of a signal (three-point method, residue as half cycles).

Parameters:

Name Type Description Default
x array - like

Signal in time order, e.g. the state of charge as a fraction.

required
min_range float

Hysteresis of reversals.

0.0

Returns:

Type Description
DataFrame

One row per cycle: depth (range), mean, count (1 or 0.5), start and end (positions).

Examples:

The ASTM E1049-85 example (loads -2, 1, -3, 5, -1, 3, -4, 4, -2) gives two full cycles of range 4 and half cycles of ranges 3, 4, 8, 9, 8 and 6:

>>> c = rainflow([-2, 1, -3, 5, -1, 3, -4, 4, -2])
>>> c.groupby("depth")["count"].sum().to_dict()
{3.0: 0.5, 4.0: 1.5, 6.0: 0.5, 8.0: 1.0, 9.0: 0.5}

rainflow_efc

rainflow_efc(cycles: DataFrame) -> float

Equivalent full cycles from rainflow cycles: sum(depth x count).

Examples:

>>> rainflow_efc(rainflow([0.1, 0.9, 0.1, 0.9, 0.1]))
1.6

depth_histogram

depth_histogram(cycles: DataFrame, years: float = 1.0, edges: ArrayLike = DEPTH_EDGES) -> pd.DataFrame

Cycles and their EFC per depth bin, per year.

Examples:

>>> h = depth_histogram(rainflow([0.1, 0.9, 0.1, 0.9, 0.1]))
>>> float(h.loc["80%-100%", "cycles"])
2.0

capacity

Capacity test evaluation: discharged energy, temperature correction, measurement uncertainty and verdict.

A capacity test (in the spirit of IEC 62933-2-1 and common LTSA practice; the procedure itself is set by the contract) charges the battery to full, rests, and discharges at the test power until the EMS stops at its lower limit. The discharged energy is integrated at the meter (or the DC terminals), optionally with the auxiliary energy added back, and corrected to the reference temperature with a linear coefficient:

E_corr = E_meas / (1 + alpha (T_test - T_ref))

Uncertainty, following the GUM with relative standard uncertainties added in quadrature and expanded with a coverage factor k:

u_meter = class / sqrt(3),   u_CT = ct / sqrt(3),   u_VT = vt / sqrt(3)       (rectangular error limits)
u_temp  = sqrt((alpha u_T)^2 + (u_alpha |T_test - T_ref|)^2)                  (temperature correction)
U       = k sqrt(u_meter^2 + u_CT^2 + u_VT^2 + u_temp^2) E_corr

Decision rules: simple passes when E_corr >= E_guaranteed; guard_band passes only when E_corr - U >= E_guaranteed, fails when E_corr + U < E_guaranteed and is inconclusive in between.

UncertaintyBudget dataclass

UncertaintyBudget(meter: float, ct: float, vt: float, temperature: float)

Relative standard uncertainties of one capacity test.

combined property

combined: float

Combined relative standard uncertainty (root sum of squares).

CapacityTest dataclass

CapacityTest(e_measured: float, temp_factor: float, e_corrected: float, u_rel_combined: float, expanded_uncertainty: float, e_guaranteed: float, margin: float, verdict: str)

Result of one capacity test.

Attributes:

Name Type Description
e_measured, e_corrected, e_guaranteed float

Measured, temperature-corrected and guaranteed energy.

temp_factor float

1 + alpha (T_test - T_ref).

u_rel_combined float

Combined relative standard uncertainty.

expanded_uncertainty float

U in energy units.

margin float

E_corr - E_guaranteed.

verdict str

pass, fail or inconclusive.

as_dict

as_dict() -> dict[str, object]

Fields as a dictionary.

relative_uncertainty

relative_uncertainty(*, meter_class_pct: float = 0.2, ct_pct: float = 0.2, vt_pct: float = 0.2, temp_c: float = float('nan'), t_ref_c: float = 25.0, temp_coeff_per_k: float = 0.003, temp_coeff_u_per_k: float = 0.001, temp_u_k: float = 1.0) -> UncertaintyBudget

Relative standard uncertainties of a capacity test (see the module description).

Parameters:

Name Type Description Default
meter_class_pct float

Error limits of the energy meter and the current and voltage transformers, percent.

0.2
ct_pct float

Error limits of the energy meter and the current and voltage transformers, percent.

0.2
vt_pct float

Error limits of the energy meter and the current and voltage transformers, percent.

0.2
temp_c float

Mean cell temperature during the test, degC; NaN means no temperature correction.

float('nan')
t_ref_c float

Reference temperature.

25.0
temp_coeff_per_k float

Temperature coefficient of the dischargeable energy (1/K) and its standard uncertainty.

0.003 and 0.001
temp_coeff_u_per_k float

Temperature coefficient of the dischargeable energy (1/K) and its standard uncertainty.

0.003 and 0.001
temp_u_k float

Standard uncertainty of the test temperature, K.

1.0

Examples:

>>> u = relative_uncertainty(temp_c=30.0)
>>> round(u.meter, 6), round(u.temperature, 6), round(u.combined, 6)
(0.001155, 0.005831, 0.006164)

evaluate_capacity_test

evaluate_capacity_test(e_measured: float, e_guaranteed: float, *, temp_c: float = float('nan'), t_ref_c: float = 25.0, temp_coeff_per_k: float = 0.003, temp_coeff_u_per_k: float = 0.001, temp_u_k: float = 1.0, meter_class_pct: float = 0.2, ct_pct: float = 0.2, vt_pct: float = 0.2, coverage_k: float = 2.0, decision_rule: Literal['simple', 'guard_band'] = 'guard_band') -> CapacityTest

Temperature correction, expanded uncertainty and verdict of one capacity test.

Parameters:

Name Type Description Default
e_measured float

Discharged energy of the test (e.g. from discharge_segment).

required
e_guaranteed float

Guaranteed energy at the time of the test, same unit.

required
temp_c float

Mean cell temperature during the test; without it no correction is applied.

float('nan')
coverage_k float

Coverage factor of the expanded uncertainty.

2.0
decision_rule ('simple', 'guard_band')
"simple"
t_ref_c float 25.0
temp_coeff_per_k float 25.0
temp_coeff_u_per_k float 25.0
temp_u_k float 25.0
meter_class_pct float 25.0
ct_pct float 25.0
vt_pct float 25.0

Returns:

Type Description
CapacityTest

Examples:

>>> r = evaluate_capacity_test(91.0, 90.0, temp_c=30.0)
>>> round(r.e_corrected, 3), round(r.expanded_uncertainty, 3), r.verdict
(89.655, 1.105, 'inconclusive')
>>> evaluate_capacity_test(91.0, 90.0, temp_c=30.0, decision_rule="simple").verdict
'fail'

discharge_segment

discharge_segment(p_discharge: ArrayLike, interval_h: float, test_power: float, steady_tolerance: float = 0.05, idle_share: float = 0.02) -> tuple[int, int, float]

Find the test discharge in a power series and integrate it.

The longest run at the test power (within steady_tolerance) is widened to the whole contiguous discharge around it (power above idle_share of the test power), so that the ramps count too.

Parameters:

Name Type Description Default
p_discharge array - like

Power per interval, positive = discharge (at the meter, or at the DC terminals).

required
interval_h float

Interval length, hours.

required
test_power float

Test power, same unit.

required
steady_tolerance float

Relative band around the test power.

0.05
idle_share float

Power below this share of the test power ends the discharge.

0.02

Returns:

Type Description
tuple

First and last position of the discharge and its energy.

Raises:

Type Description
ValueError

If the series never reaches the test power.

Examples:

>>> discharge_segment([0.0, 20.0, 50.0, 50.0, 49.0, 10.0, 0.0], 0.25, 50.0)
(1, 5, 44.75)