9e77b35ff4
Single algorithm with two passes:
Pass 1 — greedy self-consumption (real-firmware default):
For each hour, charge any surplus, discharge into any demand.
This nails sunny days: battery fills from morning surplus, exports
only after capacity is reached, drains during evening peak.
Pass 2 — daily price-spread arbitrage on the residual capacity:
For each UTC day, repeatedly find the most profitable
cheap-charge → expensive-discharge pair (positive after round-trip
efficiency), execute it, recompute SoC trajectory, repeat until no
profitable cycle remains. Discharge must fit within hourly demand
for plug-in batteries (no grid push). Sees only that day's prices,
matching what a Tibber/Frank-style smart-charging controller does
with day-ahead price visibility.
Effect:
- Sunny weeks (June 17-19): unchanged — greedy fill + evening discharge.
- No-sun weeks (Feb 7+): now generates arbitrage savings instead of €0.
- Mixed weeks: greedy handles surplus, arbitrage handles the rest.
Updated: test_grid_arbitrage_kicks_in_on_no_sun_days replaces the prior
"greedy doesn't arbitrage" test; now verifies the dispatcher charges
cheap hours and discharges expensive ones when no PV is available.
187 lines
7.7 KiB
Python
187 lines
7.7 KiB
Python
"""Smoke tests for the battery state engine and oracle."""
|
||
from __future__ import annotations
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
import pytest
|
||
|
||
from pluginbattery.sim import (
|
||
Battery,
|
||
apply_nl_tariff,
|
||
oracle_daily_schedule,
|
||
simulate,
|
||
synthesize_pv,
|
||
)
|
||
|
||
|
||
def make_df(prices: list[float], demands_kwh: list[float]) -> pd.DataFrame:
|
||
idx = pd.date_range("2024-01-01", periods=len(prices), freq="h", tz="UTC")
|
||
return pd.DataFrame(
|
||
{
|
||
"eur_per_kwh": prices,
|
||
"power_w": np.array(demands_kwh) * 1000.0,
|
||
"irradiance_w_m2": 0.0,
|
||
"demand_kwh": demands_kwh,
|
||
},
|
||
index=idx,
|
||
)
|
||
|
||
|
||
def test_simulate_clamps_charge_to_capacity():
|
||
df = make_df([0.1] * 6, [0.5] * 6)
|
||
bat = Battery(capacity_kwh=1.0, max_charge_kw=0.8, max_discharge_kw=0.8,
|
||
round_trip_eff=1.0)
|
||
schedule = np.array([[0.8, 0.0]] * 6) # ask for full charge every hour
|
||
out = simulate(df, bat, schedule)
|
||
assert out["soc_kwh"].max() <= 1.0 + 1e-9
|
||
assert out["soc_kwh"].min() >= 0.0
|
||
|
||
|
||
def test_plugin_never_exports():
|
||
df = make_df([0.5, 0.5, 0.5], [0.3, 0.3, 0.3])
|
||
bat = Battery(capacity_kwh=2.0, max_charge_kw=0.8, max_discharge_kw=0.8,
|
||
round_trip_eff=1.0, allows_export=False, initial_soc_kwh=2.0)
|
||
schedule = np.array([[0.0, 0.8]] * 3) # try to dump at full power
|
||
out = simulate(df, bat, schedule)
|
||
assert (out["discharge_kwh"] <= out["demand_kwh"] + 1e-9).all()
|
||
assert (out["grid_kwh_with_battery"] >= -1e-9).all()
|
||
|
||
|
||
def test_grid_arbitrage_kicks_in_on_no_sun_days():
|
||
"""Without PV but with a daily price spread, the dispatcher should
|
||
charge during the cheapest hours and discharge during the most
|
||
expensive — same dynamic-tariff behaviour Tibber-style controllers do."""
|
||
prices = [0.05] * 12 + [0.50] * 12
|
||
demands = [1.0] * 24
|
||
df = make_df(prices, demands)
|
||
bat = Battery(capacity_kwh=2.0, max_charge_kw=0.8, max_discharge_kw=0.8,
|
||
round_trip_eff=0.9, allows_export=False)
|
||
schedule = oracle_daily_schedule(df, bat)
|
||
out = simulate(df, bat, schedule)
|
||
# Charging happened during the cheap morning hours (0..11)
|
||
assert out["charge_kwh"].iloc[:12].sum() > 0
|
||
# Discharging happened during the expensive afternoon (12..23)
|
||
assert out["discharge_kwh"].iloc[12:].sum() > 0
|
||
# Arbitrage produced some saving
|
||
assert out["savings"].sum() > 0
|
||
|
||
|
||
def test_greedy_fills_from_surplus_then_overflows():
|
||
"""With PV surplus, greedy fills the battery as fast as power allows
|
||
until capacity is reached, then lets the rest export."""
|
||
prices = [0.20] * 24
|
||
demand = [0.1] * 24
|
||
pv = [0.0] * 6 + [3.0] * 6 + [0.0] * 12 # 6 sunny hours, 3 kWh/h surplus
|
||
df = make_df(prices, demand)
|
||
df["pv_kwh"] = pv
|
||
bat = Battery(capacity_kwh=2.0, max_charge_kw=0.8, max_discharge_kw=0.8,
|
||
round_trip_eff=1.0, allows_export=False)
|
||
schedule = oracle_daily_schedule(df, bat)
|
||
out = simulate(df, bat, schedule)
|
||
# Charging happens during the first surplus hours, capped at 0.8 kW
|
||
assert out["charge_kwh"].iloc[6:9].sum() == pytest.approx(2.0, abs=1e-6)
|
||
# Once full, no more charging even though surplus continues
|
||
assert out["charge_kwh"].iloc[9:12].sum() == 0
|
||
# Battery discharges into evening demand
|
||
assert out["discharge_kwh"].iloc[12:].sum() > 0
|
||
|
||
|
||
def test_apply_nl_tariff_matches_user_formula():
|
||
df = make_df([0.0, 0.10, -0.05], [1.0, 1.0, 1.0])
|
||
out = apply_nl_tariff(df)
|
||
expected = [0.0 * 1.21 + 0.136, 0.10 * 1.21 + 0.136, -0.05 * 1.21 + 0.136]
|
||
assert np.allclose(out["eur_per_kwh"].to_numpy(), expected)
|
||
assert np.allclose(out["epex_eur_per_kwh"].to_numpy(), [0.0, 0.10, -0.05])
|
||
|
||
|
||
def test_fixed_tax_reduces_optimal_cycle_count():
|
||
"""A flat per-kWh charge makes round-trip losses more expensive, so the
|
||
oracle should run fewer cycles when the same EPEX series is consumer-priced."""
|
||
prices = [0.05] * 12 + [0.20] * 12
|
||
df_raw = make_df(prices, [1.0] * 24)
|
||
df_consumer = apply_nl_tariff(df_raw)
|
||
bat = Battery(capacity_kwh=2.0, max_charge_kw=0.8, max_discharge_kw=0.8,
|
||
round_trip_eff=0.9, allows_export=False)
|
||
sched_raw = oracle_daily_schedule(df_raw, bat)
|
||
sched_cons = oracle_daily_schedule(df_consumer, bat)
|
||
# On consumer prices the opportunity cost of efficiency loss is higher,
|
||
# so total charge_kwh should not increase (often decreases or stays equal).
|
||
assert sched_cons[:, 0].sum() <= sched_raw[:, 0].sum() + 1e-6
|
||
|
||
|
||
def test_oracle_skips_arbitrage_when_eff_kills_it():
|
||
# Spread 0.20 → 0.21 against η_rt=0.5 means every cycle loses money.
|
||
df = make_df([0.20] * 12 + [0.21] * 12, [1.0] * 24)
|
||
bat = Battery(capacity_kwh=2.0, max_charge_kw=0.8, max_discharge_kw=0.8,
|
||
round_trip_eff=0.5, allows_export=False)
|
||
schedule = oracle_daily_schedule(df, bat)
|
||
out = simulate(df, bat, schedule)
|
||
assert out["savings"].sum() < 1e-6
|
||
|
||
|
||
def test_synthesize_pv_hits_target_annual():
|
||
"""synthesize_pv should calibrate so annual output ≈ target × kWp."""
|
||
n = 24 * 30 # 30 days
|
||
idx = pd.date_range("2024-06-01", periods=n, freq="h", tz="UTC")
|
||
# Simple square wave: 600 W/m² for 8 daylight hours, zero otherwise.
|
||
irr = np.zeros(n)
|
||
for d in range(30):
|
||
irr[d * 24 + 8 : d * 24 + 16] = 600.0
|
||
df = pd.DataFrame({
|
||
"eur_per_kwh": 0.20, "power_w": 0.0,
|
||
"irradiance_w_m2": irr, "demand_kwh": 0.0,
|
||
}, index=idx)
|
||
out = synthesize_pv(df, kwp=3.0, target_kwh_per_kwp_per_year=900.0)
|
||
annual_pv = out["pv_kwh"].sum() * (8766 / n)
|
||
assert abs(annual_pv - 3.0 * 900.0) < 1.0
|
||
|
||
|
||
def test_plugin_with_pv_does_not_push_to_grid():
|
||
"""Plug-in battery + surplus solar: discharge must be 0 in surplus hours."""
|
||
df = make_df([0.30] * 24, [0.5] * 24)
|
||
df["pv_kwh"] = [3.0] * 12 + [0.0] * 12 # huge midday surplus
|
||
bat = Battery(capacity_kwh=2.0, max_charge_kw=0.8, max_discharge_kw=0.8,
|
||
round_trip_eff=0.9, allows_export=False, initial_soc_kwh=2.0)
|
||
schedule = np.array([[0.0, 0.8]] * 24) # try to dump every hour
|
||
out = simulate(df, bat, schedule)
|
||
surplus_hours = out["pv_kwh"] > out["demand_kwh"]
|
||
assert (out.loc[surplus_hours, "discharge_kwh"] == 0).all()
|
||
|
||
|
||
def test_no_saldering_increases_battery_savings():
|
||
"""Removing saldering should make a battery on a PV system more valuable.
|
||
|
||
Reason: surplus solar that previously credited at consumer price now only
|
||
earns raw EPEX. Storing it for later self-consumption is now strictly
|
||
better than the previous opportunity cost.
|
||
"""
|
||
# Day with cheap morning EPEX, noon surplus solar, expensive evening.
|
||
prices_consumer = [0.20] * 6 + [0.15] * 6 + [0.40] * 12
|
||
prices_epex = [0.05] * 6 + [0.02] * 6 + [0.20] * 12 # before VAT/tax
|
||
demands = [0.5] * 24
|
||
pv = [0.0] * 8 + [3.0] * 6 + [0.0] * 10
|
||
n = 24
|
||
idx = pd.date_range("2024-01-01", periods=n, freq="h", tz="UTC")
|
||
base = pd.DataFrame({
|
||
"eur_per_kwh": prices_consumer,
|
||
"power_w": np.array(demands) * 1000.0,
|
||
"irradiance_w_m2": 0.0,
|
||
"demand_kwh": demands,
|
||
"pv_kwh": pv,
|
||
"epex_eur_per_kwh": prices_epex,
|
||
}, index=idx)
|
||
bat = Battery(capacity_kwh=5.0, max_charge_kw=2.5, max_discharge_kw=2.5,
|
||
round_trip_eff=0.9, allows_export=False)
|
||
|
||
# With full saldering (export = import).
|
||
df_sald = base.copy()
|
||
out_sald = simulate(df_sald, bat, oracle_daily_schedule(df_sald, bat))
|
||
savings_sald = out_sald["savings"].sum()
|
||
|
||
# Without saldering (export = raw EPEX).
|
||
df_no = base.copy()
|
||
df_no["export_eur_per_kwh"] = df_no["epex_eur_per_kwh"]
|
||
out_no = simulate(df_no, bat, oracle_daily_schedule(df_no, bat))
|
||
savings_no = out_no["savings"].sum()
|
||
|
||
assert savings_no > savings_sald
|