60e0706736
- Hourly data exporter (InfluxDB → CSV) for prices, P1, irradiance. - LP-based 24h-foresight oracle dispatch with SoC-consistent state engine. - Reverse-engineered thuisbatterijgids.nl formula (matches their quotes to within €0.50 across three battery configs). - Catalog scraper for the 52 batteries on thuisbatterijgids.net via their /wp-json REST endpoint. - Web app (Flask) that ranks every catalog battery by honest payback and contrasts with the store's quote, deployable via the included Procfile.
160 lines
6.4 KiB
Python
160 lines
6.4 KiB
Python
"""Smoke tests for the battery state engine and oracle."""
|
||
from __future__ import annotations
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
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_oracle_arbitrages_clear_spread():
|
||
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)
|
||
assert out["charge_kwh"].iloc[:12].sum() > 0
|
||
assert out["discharge_kwh"].iloc[12:].sum() > 0
|
||
assert out["savings"].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
|